package cn.lili.modules.search.serviceimpl;
|
|
import cn.hutool.core.collection.CollUtil;
|
import cn.hutool.core.date.DateUtil;
|
import cn.hutool.core.text.CharSequenceUtil;
|
import cn.hutool.core.thread.ThreadUtil;
|
import cn.hutool.core.util.ReflectUtil;
|
import cn.hutool.json.JSONObject;
|
import cn.hutool.json.JSONUtil;
|
import cn.lili.cache.Cache;
|
import cn.lili.cache.CachePrefix;
|
import cn.lili.common.enums.PromotionTypeEnum;
|
import cn.lili.common.enums.ResultCode;
|
import cn.lili.common.exception.RetryException;
|
import cn.lili.common.exception.ServiceException;
|
import cn.lili.common.properties.RocketmqCustomProperties;
|
import cn.lili.common.vo.PageVO;
|
import cn.lili.elasticsearch.BaseElasticsearchService;
|
import cn.lili.elasticsearch.EsSuffix;
|
import cn.lili.elasticsearch.config.ElasticsearchProperties;
|
import cn.lili.modules.goods.entity.dos.GoodsSku;
|
import cn.lili.modules.goods.entity.dto.GoodsParamsDTO;
|
import cn.lili.modules.goods.entity.dto.GoodsSkuDTO;
|
import cn.lili.modules.goods.entity.enums.GoodsAuthEnum;
|
import cn.lili.modules.goods.entity.enums.GoodsSalesModeEnum;
|
import cn.lili.modules.goods.entity.enums.GoodsStatusEnum;
|
import cn.lili.modules.goods.service.BrandService;
|
import cn.lili.modules.goods.service.CategoryService;
|
import cn.lili.modules.goods.service.GoodsSkuService;
|
import cn.lili.modules.goods.service.StoreGoodsLabelService;
|
import cn.lili.modules.promotion.entity.dos.BasePromotions;
|
import cn.lili.modules.promotion.entity.dos.PromotionGoods;
|
import cn.lili.modules.promotion.entity.enums.PromotionsScopeTypeEnum;
|
import cn.lili.modules.promotion.entity.enums.PromotionsStatusEnum;
|
import cn.lili.modules.promotion.service.PromotionGoodsService;
|
import cn.lili.modules.promotion.service.PromotionService;
|
import cn.lili.modules.promotion.tools.PromotionTools;
|
import cn.lili.modules.search.entity.dos.CustomWords;
|
import cn.lili.modules.search.entity.dos.EsGoodsAttribute;
|
import cn.lili.modules.search.entity.dos.EsGoodsIndex;
|
import cn.lili.modules.search.entity.dto.EsGoodsSearchDTO;
|
import cn.lili.modules.search.repository.EsGoodsIndexRepository;
|
import cn.lili.modules.search.service.CustomWordsService;
|
import cn.lili.modules.search.service.EsGoodsIndexService;
|
import cn.lili.modules.search.service.EsGoodsSearchService;
|
import cn.lili.rocketmq.RocketmqSendCallbackBuilder;
|
import cn.lili.rocketmq.tags.GoodsTagsEnum;
|
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
|
import com.baomidou.mybatisplus.core.metadata.IPage;
|
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.rocketmq.spring.core.RocketMQTemplate;
|
import org.elasticsearch.action.ActionListener;
|
import org.elasticsearch.action.bulk.BulkRequest;
|
import org.elasticsearch.action.bulk.BulkResponse;
|
import org.elasticsearch.action.update.UpdateRequest;
|
import org.elasticsearch.client.RequestOptions;
|
import org.elasticsearch.index.query.BoolQueryBuilder;
|
import org.elasticsearch.index.query.QueryBuilders;
|
import org.elasticsearch.index.query.TermQueryBuilder;
|
import org.elasticsearch.index.reindex.BulkByScrollResponse;
|
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
|
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
|
import org.elasticsearch.script.Script;
|
import org.elasticsearch.script.ScriptType;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.data.domain.PageRequest;
|
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
|
import org.springframework.data.elasticsearch.core.SearchHit;
|
import org.springframework.data.elasticsearch.core.SearchHits;
|
import org.springframework.data.elasticsearch.core.SearchPage;
|
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
|
import org.springframework.stereotype.Service;
|
|
import java.lang.reflect.Field;
|
import java.util.*;
|
import java.util.concurrent.TimeUnit;
|
import java.util.stream.Collectors;
|
|
/**
|
* 商品索引业务层实现
|
*
|
* @author paulG
|
* @since 2020/10/14
|
**/
|
@Slf4j
|
@Service
|
public class EsGoodsIndexServiceImpl extends BaseElasticsearchService implements EsGoodsIndexService {
|
|
private static final String IGNORE_FIELD = "serialVersionUID,promotionMap,id,goodsId";
|
private static final String KEY_SUCCESS = "success";
|
private static final String KEY_FAIL = "fail";
|
private static final String KEY_PROCESSED = "processed";
|
private final Map<String, Field> fieldMap = ReflectUtil.getFieldMap(EsGoodsIndex.class);
|
@Autowired
|
private ElasticsearchProperties elasticsearchProperties;
|
@Autowired
|
private EsGoodsIndexRepository goodsIndexRepository;
|
@Autowired
|
private EsGoodsSearchService goodsSearchService;
|
@Autowired
|
private PromotionService promotionService;
|
|
@Autowired
|
private PromotionGoodsService promotionGoodsService;
|
|
@Autowired
|
private CustomWordsService customWordsService;
|
|
@Autowired
|
private GoodsSkuService goodsSkuService;
|
@Autowired
|
private BrandService brandService;
|
|
@Autowired
|
private CategoryService categoryService;
|
|
@Autowired
|
private StoreGoodsLabelService storeGoodsLabelService;
|
@Autowired
|
private Cache<Object> cache;
|
/**
|
* rocketMq
|
*/
|
@Autowired
|
private RocketMQTemplate rocketMQTemplate;
|
/**
|
* rocketMq配置
|
*/
|
@Autowired
|
private RocketmqCustomProperties rocketmqCustomProperties;
|
@Autowired
|
private ElasticsearchOperations restTemplate;
|
|
/**
|
* 去除 重复元素
|
*
|
* @param list
|
* @return
|
*/
|
public static void removeDuplicate(List<String> list) {
|
HashSet<String> h = new HashSet<>(list);
|
list.clear();
|
list.addAll(h);
|
}
|
|
@Override
|
public void init() {
|
//获取索引任务标识
|
Boolean flag = (Boolean) cache.get(CachePrefix.INIT_INDEX_FLAG.getPrefix());
|
//为空则默认写入没有任务
|
if (flag == null) {
|
cache.put(CachePrefix.INIT_INDEX_FLAG.getPrefix(), false);
|
}
|
//有正在初始化的任务，则提示异常
|
if (Boolean.TRUE.equals(flag)) {
|
throw new ServiceException(ResultCode.INDEX_BUILDING);
|
}
|
|
//初始化标识
|
cache.put(CachePrefix.INIT_INDEX_PROCESS.getPrefix(), null);
|
cache.put(CachePrefix.INIT_INDEX_FLAG.getPrefix(), true, 10L, TimeUnit.MINUTES);
|
|
|
ThreadUtil.execAsync(() -> {
|
try {
|
|
QueryWrapper<GoodsSkuDTO> skuQueryWrapper = new QueryWrapper<>();
|
skuQueryWrapper.eq("gs.auth_flag", GoodsAuthEnum.PASS.name());
|
skuQueryWrapper.eq("gs.market_enable", GoodsStatusEnum.UPPER.name());
|
skuQueryWrapper.eq("gs.delete_flag", false);
|
skuQueryWrapper.gt("gs.quantity", 0);
|
|
|
Map<String, Long> resultMap = (Map<String, Long>) cache.get(CachePrefix.INIT_INDEX_PROCESS.getPrefix());
|
|
if (CollUtil.isEmpty(resultMap)) {
|
QueryWrapper<GoodsSku> skuCountQueryWrapper = new QueryWrapper<>();
|
skuCountQueryWrapper.eq("auth_flag", GoodsAuthEnum.PASS.name());
|
skuCountQueryWrapper.eq("market_enable", GoodsStatusEnum.UPPER.name());
|
skuCountQueryWrapper.eq("delete_flag", false);
|
skuCountQueryWrapper.gt("quantity", 0);
|
resultMap = new HashMap<>();
|
resultMap.put(KEY_SUCCESS, 0L);
|
resultMap.put(KEY_FAIL, 0L);
|
resultMap.put(KEY_PROCESSED, 0L);
|
resultMap.put("total", this.goodsSkuService.count(skuCountQueryWrapper));
|
cache.put(CachePrefix.INIT_INDEX_PROCESS.getPrefix(), resultMap);
|
}
|
|
for (int i = 1; ; i++) {
|
List<EsGoodsIndex> esGoodsIndices = new ArrayList<>();
|
Page<GoodsSkuDTO> skuPage = new Page<>(i, 2000);
|
IPage<GoodsSkuDTO> skuIPage = goodsSkuService.getGoodsSkuDTOByPage(skuPage, skuQueryWrapper);
|
|
if (skuIPage == null || CollUtil.isEmpty(skuIPage.getRecords())) {
|
break;
|
}
|
List<String> skuIds = skuIPage.getRecords().stream().map(GoodsSku::getId).collect(Collectors.toList());
|
List<PromotionGoods> skuValidPromotions = promotionGoodsService.findSkuValidPromotions(skuIds);
|
|
List<String> brandIds = new ArrayList<>();
|
|
List<String> categoryPaths = new ArrayList<>();
|
|
List<String> storeCategoryPaths = new ArrayList<>();
|
|
for (GoodsSkuDTO goodsSkuDTO : skuIPage.getRecords()) {
|
if (CharSequenceUtil.isNotEmpty(goodsSkuDTO.getBrandId())) {
|
brandIds.add(goodsSkuDTO.getBrandId());
|
}
|
if (CharSequenceUtil.isNotEmpty(goodsSkuDTO.getStoreCategoryPath())) {
|
storeCategoryPaths.addAll(Arrays.asList(goodsSkuDTO.getStoreCategoryPath().split(",")));
|
}
|
if (CharSequenceUtil.isNotEmpty((goodsSkuDTO.getCategoryPath()))) {
|
categoryPaths.addAll(Arrays.asList(goodsSkuDTO.getCategoryPath().split(",")));
|
}
|
}
|
|
List<Map<String, Object>> brandList = new ArrayList<>();
|
if (CollUtil.isNotEmpty(brandIds)) {
|
brandList = this.brandService.getBrandsMapsByCategory(CollUtil.distinct(brandIds), "id,name,logo");
|
}
|
List<Map<String, Object>> categoryList = new ArrayList<>();
|
if (CollUtil.isNotEmpty(categoryPaths)) {
|
categoryList = this.categoryService.listMapsByIdsOrderByLevel(CollUtil.distinct(categoryPaths), "id,name");
|
}
|
List<Map<String, Object>> storeCategoryList = new ArrayList<>();
|
if (CollUtil.isNotEmpty(storeCategoryPaths)) {
|
storeCategoryList = this.storeGoodsLabelService.listMapsByStoreIds(CollUtil.distinct(storeCategoryPaths), "id,label_name");
|
}
|
|
for (GoodsSkuDTO goodsSku : skuIPage.getRecords()) {
|
int skuSource = 100;
|
EsGoodsIndex esGoodsIndex = wrapperEsGoodsIndex(goodsSku, brandList, categoryList, storeCategoryList);
|
long count = esGoodsIndices.stream().filter(j -> j.getGoodsId().equals(esGoodsIndex.getGoodsId())).count();
|
if (count >= 1) {
|
skuSource -= count;
|
}
|
if (skuSource <= 0) {
|
skuSource = 1;
|
}
|
esGoodsIndex.setSkuSource(skuSource);
|
|
|
//设置促销信息
|
List<PromotionGoods> promotionGoods = skuValidPromotions.stream()
|
.filter(j ->
|
(CharSequenceUtil.isNotEmpty(j.getSkuId()) && j.getSkuId().equals(goodsSku.getId())) ||
|
(j.getScopeType().equals(PromotionsScopeTypeEnum.ALL.name()) && j.getStoreId().equals("0")) ||
|
(j.getScopeType().equals(PromotionsScopeTypeEnum.ALL.name()) && j.getStoreId().equals(esGoodsIndex.getStoreId())) ||
|
(j.getScopeType().equals(PromotionsScopeTypeEnum.PORTION_GOODS_CATEGORY.name()) && j.getStoreId().equals("0") && j.getScopeId().contains(goodsSku.getCategoryPath()))||
|
(j.getScopeType().equals(PromotionsScopeTypeEnum.PORTION_GOODS_CATEGORY.name()) && j.getStoreId().equals(goodsSku.getStoreId()) && j.getScopeId().contains(goodsSku.getCategoryPath()))
|
)
|
.collect(Collectors.toList());
|
if (CollUtil.isNotEmpty(promotionGoods)) {
|
esGoodsIndex.setPromotionMapJson(JSONUtil.toJsonStr(promotionService.wrapperPromotionMapList(promotionGoods)));
|
}
|
|
esGoodsIndices.add(esGoodsIndex);
|
//库存锁是在redis做的，所以生成索引，同时更新一下redis中的库存数量
|
cache.put(GoodsSkuService.getStockCacheKey(goodsSku.getId()), goodsSku.getQuantity());
|
}
|
|
//批量插入索引，如果为第一次则删除原索引并创建新索引
|
this.initIndex(esGoodsIndices, i == 1);
|
}
|
|
cache.put(CachePrefix.INIT_INDEX_FLAG.getPrefix(), false);
|
|
//初始化商品索引
|
} catch (Exception e) {
|
log.error("商品索引生成异常：", e);
|
//如果出现异常，则将进行中的任务标识取消掉，打印日志
|
cache.put(CachePrefix.INIT_INDEX_PROCESS.getPrefix(), null);
|
cache.put(CachePrefix.INIT_INDEX_FLAG.getPrefix(), false);
|
}
|
});
|
|
}
|
|
@Override
|
public Map<String, Long> getProgress() {
|
Map<String, Long> map = (Map<String, Long>) cache.get(CachePrefix.INIT_INDEX_PROCESS.getPrefix());
|
if (map == null) {
|
return Collections.emptyMap();
|
}
|
Boolean flag = (Boolean) cache.get(CachePrefix.INIT_INDEX_FLAG.getPrefix());
|
map.put("flag", Boolean.TRUE.equals(flag) ? 1L : 0L);
|
return map;
|
}
|
|
@Override
|
public void initIndex() {
|
//索引名称拼接
|
String indexName = this.getIndexName();
|
|
//索引初始化，因为mapping结构问题：
|
//但是如果索引已经自动生成过，这里就不会创建索引，设置mapping，所以这里决定在初始化索引的同时，将已有索引删除，重新创建
|
|
boolean indexExist = this.indexExist(indexName);
|
log.info("检测 {} 索引结构是否存在：{}", indexName, indexExist);
|
if (!indexExist) {
|
|
log.info("初始化索引结构 {}", indexName);
|
//如果索引不存在，则创建索引
|
createIndexRequest(indexName);
|
}
|
|
}
|
|
@Override
|
public void addIndex(EsGoodsIndex goods) {
|
try {
|
//分词器分词
|
this.analyzeAndSaveWords(goods);
|
//生成索引
|
goodsIndexRepository.save(goods);
|
} catch (Exception e) {
|
log.error("为商品[" + goods.getGoodsName() + "]生成索引异常", e);
|
}
|
}
|
|
/**
|
* 添加商品索引
|
*
|
* @param goods 商品索引信息
|
*/
|
@Override
|
public void addIndex(List<EsGoodsIndex> goods) {
|
try {
|
for (EsGoodsIndex esGoodsIndex : goods) {
|
this.analyzeAndSaveWords(esGoodsIndex);
|
}
|
goodsIndexRepository.saveAll(goods);
|
} catch (Exception e) {
|
log.error("批量为商品生成索引异常", e);
|
}
|
}
|
|
@Override
|
public void updateIndex(EsGoodsIndex goods) {
|
this.analyzeAndSaveWords(goods);
|
goodsIndexRepository.save(goods);
|
}
|
|
/**
|
* 商品分词
|
*
|
* @param goods 商品信息
|
*/
|
private void analyzeAndSaveWords(EsGoodsIndex goods) {
|
try {
|
List<String> keywordsList = new ArrayList<>();
|
//根据商品参数分词
|
if (goods.getAttrList() != null && !goods.getAttrList().isEmpty()) {
|
//保存分词
|
for (EsGoodsAttribute esGoodsAttribute : goods.getAttrList()) {
|
if (keywordsList.stream().noneMatch(i -> i.toLowerCase(Locale.ROOT).equals(esGoodsAttribute.getValue().toLowerCase(Locale.ROOT)))) {
|
keywordsList.add(esGoodsAttribute.getValue());
|
}
|
}
|
}
|
//根据商品名称生成分词
|
keywordsList.add(goods.getGoodsName().substring(0, Math.min(goods.getGoodsName().length(), 10)));
|
|
//去除重复词
|
removeDuplicate(keywordsList);
|
//入库自定义分词
|
List<CustomWords> customWordsArrayList = new ArrayList<>();
|
keywordsList.forEach(item -> customWordsArrayList.add(new CustomWords(item)));
|
//这里采用先批量删除再插入的方法，故意这么做。否则需要挨个匹配是否存在，性能消耗更大
|
if (CollUtil.isNotEmpty(customWordsArrayList)) {
|
customWordsService.insertBatchCustomWords(customWordsArrayList);
|
}
|
} catch (Exception e) {
|
log.info(goods + "自定义分词错误", e);
|
}
|
}
|
|
/**
|
* 更新商品索引的的部分属性（只填写更新的字段，不需要更新的字段不要填写）
|
*
|
* @param id 商品索引id
|
* @param goods 更新后的购买数量
|
*/
|
@Override
|
public void updateIndex(String id, EsGoodsIndex goods) {
|
EsGoodsIndex goodsIndex = this.findById(id);
|
// 通过反射获取全部字段，在根据参数字段是否为空，设置要更新的字段
|
for (Map.Entry<String, Field> entry : fieldMap.entrySet()) {
|
Object fieldValue = ReflectUtil.getFieldValue(goods, entry.getValue());
|
if (fieldValue != null && !IGNORE_FIELD.contains(entry.getKey())) {
|
ReflectUtil.setFieldValue(goodsIndex, entry.getValue(), fieldValue);
|
}
|
}
|
goodsIndexRepository.save(goodsIndex);
|
}
|
|
/**
|
* 更新商品索引的的部分属性（只填写更新的字段，不需要更新的字段不要填写）
|
*
|
* @param queryFields 查询字段
|
* @param updateFields 更新字段
|
*/
|
@Override
|
public void updateIndex(Map<String, Object> queryFields, Map<String, Object> updateFields) {
|
UpdateByQueryRequest update = new UpdateByQueryRequest(getIndexName());
|
BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
|
for (Map.Entry<String, Object> entry : queryFields.entrySet()) {
|
TermQueryBuilder termQueryBuilder = new TermQueryBuilder(entry.getKey(), entry.getValue());
|
queryBuilder.filter(termQueryBuilder);
|
}
|
update.setQuery(queryBuilder);
|
StringBuilder script = new StringBuilder();
|
for (Map.Entry<String, Object> entry : updateFields.entrySet()) {
|
script.append("ctx._source.").append(entry.getKey()).append("=").append("'").append(entry.getValue()).append("'").append(";");
|
}
|
update.setScript(new Script(script.toString()));
|
update.setConflicts("proceed");
|
|
this.client.updateByQueryAsync(update, RequestOptions.DEFAULT, new ActionListener<BulkByScrollResponse>() {
|
@Override
|
public void onResponse(BulkByScrollResponse bulkByScrollResponse) {
|
if (bulkByScrollResponse.getVersionConflicts() > 0) {
|
throw new RetryException("更新商品索引失败，es内容版本冲突");
|
}
|
}
|
|
@Override
|
public void onFailure(Exception e) {
|
log.error("更新商品索引异常", e);
|
}
|
});
|
}
|
|
/**
|
* 批量商品索引的的属性（ID 必填, 其他字段只填写更新的字段，不需要更新的字段不要填写。）
|
*
|
* @param goodsIndices 商品索引列表
|
*/
|
@Override
|
public void updateBulkIndex(List<EsGoodsIndex> goodsIndices) {
|
//索引名称拼接
|
String indexName = getIndexName();
|
|
BulkRequest request = new BulkRequest();
|
|
for (EsGoodsIndex goodsIndex : goodsIndices) {
|
UpdateRequest updateRequest = new UpdateRequest(indexName, goodsIndex.getId());
|
|
JSONObject jsonObject = JSONUtil.parseObj(goodsIndex);
|
jsonObject.set("releaseTime", goodsIndex.getReleaseTime());
|
updateRequest.doc(jsonObject);
|
request.add(updateRequest);
|
}
|
this.client.bulkAsync(request, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
|
@Override
|
public void onResponse(BulkResponse bulkItemResponses) {
|
// 判断索引如果不存在的处理
|
log.info("批量更新商品索引结果：{}", bulkItemResponses.buildFailureMessage());
|
}
|
|
@Override
|
public void onFailure(Exception e) {
|
log.error("批量更新商品索引异常", e);
|
}
|
});
|
}
|
|
/**
|
* 删除索引
|
*
|
* @param queryFields 查询条件
|
*/
|
@Override
|
public void deleteIndex(Map<String, Object> queryFields) {
|
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
|
for (Map.Entry<String, Object> entry : queryFields.entrySet()) {
|
boolQueryBuilder.filter(QueryBuilders.termsQuery(entry.getKey(), entry.getValue().toString()));
|
}
|
|
DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest();
|
deleteByQueryRequest.setQuery(boolQueryBuilder);
|
deleteByQueryRequest.indices(getIndexName());
|
deleteByQueryRequest.setConflicts("proceed");
|
this.client.deleteByQueryAsync(deleteByQueryRequest, RequestOptions.DEFAULT, new ActionListener<BulkByScrollResponse>() {
|
|
@Override
|
public void onResponse(BulkByScrollResponse bulkByScrollResponse) {
|
if (bulkByScrollResponse.getVersionConflicts() > 0) {
|
throw new RetryException("删除索引失败，es内容版本冲突");
|
}
|
}
|
|
@Override
|
public void onFailure(Exception e) {
|
throw new RetryException("删除索引失败，" + e.getMessage());
|
}
|
});
|
|
}
|
|
/**
|
* 删除索引
|
*
|
* @param id 商品索引信息
|
*/
|
@Override
|
public void deleteIndexById(String id) {
|
goodsIndexRepository.deleteById(id);
|
}
|
|
/**
|
* 删除索引
|
*
|
* @param ids 商品索引id集合
|
*/
|
@Override
|
public void deleteIndexByIds(List<String> ids) {
|
NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
|
queryBuilder.withQuery(QueryBuilders.termsQuery("id", ids.toArray()));
|
this.restTemplate.delete(queryBuilder.build(), EsGoodsIndex.class);
|
}
|
|
@Override
|
public void initIndex(List<EsGoodsIndex> goodsIndexList, boolean regeneratorIndex) {
|
if (goodsIndexList == null || goodsIndexList.isEmpty()) {
|
//初始化标识
|
cache.put(CachePrefix.INIT_INDEX_PROCESS.getPrefix(), null);
|
cache.put(CachePrefix.INIT_INDEX_FLAG.getPrefix(), false);
|
return;
|
}
|
//索引名称拼接
|
String indexName = this.getIndexName();
|
|
//索引初始化，因为mapping结构问题：
|
//但是如果索引已经自动生成过，这里就不会创建索引，设置mapping，所以这里决定在初始化索引的同时，将已有索引删除，重新创建
|
|
//如果索引存在，则删除，重新生成。 这里应该有更优解。
|
boolean indexExist = this.indexExist(indexName);
|
if (regeneratorIndex || !indexExist) {
|
if (indexExist) {
|
this.deleteIndexRequest(indexName);
|
}
|
//如果索引不存在，则创建索引
|
this.createIndexRequest(indexName);
|
}
|
|
Map<String, Long> resultMap = (Map<String, Long>) cache.get(CachePrefix.INIT_INDEX_PROCESS.getPrefix());
|
if (!goodsIndexList.isEmpty()) {
|
for (EsGoodsIndex goodsIndex : goodsIndexList) {
|
try {
|
log.info("生成商品索引：{}", goodsIndex);
|
this.addIndex(goodsIndex);
|
resultMap.put(KEY_SUCCESS, resultMap.get(KEY_SUCCESS) + 1);
|
} catch (Exception e) {
|
log.error("商品{}生成索引错误！", goodsIndex);
|
resultMap.put(KEY_FAIL, resultMap.get(KEY_FAIL) + 1);
|
}
|
resultMap.put(KEY_PROCESSED, resultMap.get(KEY_PROCESSED) + 1);
|
cache.put(CachePrefix.INIT_INDEX_PROCESS.getPrefix(), resultMap);
|
}
|
}
|
cache.put(CachePrefix.INIT_INDEX_PROCESS.getPrefix(), resultMap);
|
}
|
|
@Override
|
public UpdateRequest updateEsGoodsIndexPromotions(String id, BasePromotions promotion, String key) {
|
EsGoodsIndex goodsIndex = findById(id);
|
if (goodsIndex != null) {
|
// 批发商品不参与促销（除优惠券和满减）
|
if (PromotionTools.isPromotionsTypeNeedsToChecked(key) && GoodsSalesModeEnum.WHOLESALE.name().equals(goodsIndex.getSalesModel())) {
|
return null;
|
}
|
//更新索引
|
return this.updateGoodsIndexPromotion(goodsIndex, key, promotion);
|
} else {
|
log.error("更新索引商品促销信息失败！skuId 为 {} 的索引不存在！", id);
|
return null;
|
}
|
}
|
|
/**
|
* 更新商品索引的促销信息
|
*
|
* @param ids skuId集合
|
* @param promotion 促销信息
|
* @param key 促销信息的key
|
*/
|
@Override
|
public void updateEsGoodsIndexPromotions(List<String> ids, BasePromotions promotion, String key) {
|
BulkRequest bulkRequest = new BulkRequest();
|
log.info("更新商品索引的促销信息----------");
|
log.info("商品ids: {}", ids);
|
log.info("活动: {}", promotion);
|
log.info("key: {}", key);
|
for (String id : ids) {
|
UpdateRequest updateRequest = this.updateEsGoodsIndexPromotions(id, promotion, key);
|
if (updateRequest != null) {
|
bulkRequest.add(updateRequest);
|
}
|
}
|
this.executeBulkUpdateRequest(bulkRequest);
|
}
|
|
|
@Override
|
public void updateEsGoodsIndexByList(List<PromotionGoods> promotionGoodsList, BasePromotions promotion, String key) {
|
BulkRequest bulkRequest = new BulkRequest();
|
log.info("修改商品活动索引");
|
log.info("促销商品信息: {}", promotionGoodsList);
|
log.info("活动关键字: {}", key);
|
log.info("活动: {}", promotion);
|
if (promotionGoodsList != null) {
|
//循环更新 促销商品索引
|
for (PromotionGoods promotionGoods : promotionGoodsList) {
|
promotion.setStartTime(promotionGoods.getStartTime());
|
promotion.setEndTime(promotionGoods.getEndTime());
|
UpdateRequest updateRequest = this.updateEsGoodsIndexPromotions(promotionGoods.getSkuId(), promotion, key);
|
if (updateRequest != null) {
|
bulkRequest.add(updateRequest);
|
}
|
}
|
}
|
this.executeBulkUpdateRequest(bulkRequest);
|
}
|
|
/**
|
* 更新全部商品索引的促销信息
|
*
|
* @param promotion 促销信息
|
* @param key 促销信息的key
|
*/
|
@Override
|
public void updateEsGoodsIndexAllByList(BasePromotions promotion, String key) {
|
ThreadUtil.execAsync(() -> this.executeUpdateEsGoodsIndexAll(promotion, key));
|
|
}
|
|
private void executeUpdateEsGoodsIndexAll(BasePromotions promotion, String key) {
|
for (int i = 0; ; i++) {
|
List<String> skuIds;
|
PageVO pageVO = new PageVO();
|
pageVO.setPageNumber(i);
|
pageVO.setPageSize(1000);
|
EsGoodsSearchDTO searchDTO = new EsGoodsSearchDTO();
|
if (PromotionTools.isPromotionsTypeNeedsToChecked(key)) {
|
searchDTO.setSalesModel(GoodsSalesModeEnum.RETAIL.name());
|
}
|
//如果storeId不为空，则表示是店铺活动
|
if (promotion.getStoreId() != null && !promotion.getStoreId().equals(PromotionTools.PLATFORM_ID)) {
|
searchDTO.setStoreId(promotion.getStoreId());
|
}
|
|
//查询出店铺商品
|
SearchPage<EsGoodsIndex> esGoodsIndices = goodsSearchService.searchGoods(searchDTO, pageVO);
|
|
skuIds = esGoodsIndices.isEmpty() ? new ArrayList<>() :
|
esGoodsIndices.getContent().stream().map(SearchHit::getId).collect(Collectors.toList());
|
if (skuIds.isEmpty()) {
|
break;
|
}
|
this.deleteEsGoodsPromotionByPromotionKey(skuIds, key);
|
this.updateEsGoodsIndexPromotions(skuIds, promotion, key);
|
}
|
}
|
|
@Override
|
public void deleteEsGoodsPromotionByPromotionKey(List<String> skuIds, String promotionsKey) {
|
BulkRequest bulkRequest = new BulkRequest();
|
log.info("删除商品活动索引");
|
log.info("商品skuIds: {}", skuIds);
|
log.info("活动Key: {}", promotionsKey);
|
if (skuIds == null || skuIds.isEmpty()) {
|
return;
|
}
|
for (String skuId : skuIds) {
|
EsGoodsIndex goodsIndex = findById(skuId);
|
//商品索引不为空
|
if (goodsIndex != null) {
|
UpdateRequest updateRequest = this.removePromotionByPromotionKey(goodsIndex, promotionsKey);
|
if (updateRequest != null) {
|
bulkRequest.add(updateRequest);
|
}
|
} else {
|
log.error("更新索引商品促销信息失败！skuId 为 【{}】的索引不存在！", skuId);
|
}
|
}
|
this.executeBulkUpdateRequest(bulkRequest);
|
}
|
|
/**
|
* 删除索引中指定的促销活动id的促销活动
|
*
|
* @param promotionsKey 促销活动Key
|
*/
|
@Override
|
public void deleteEsGoodsPromotionByPromotionKey(String promotionsKey) {
|
ThreadUtil.execAsync(() -> {
|
for (int i = 0; ; i++) {
|
BulkRequest bulkRequest = new BulkRequest();
|
|
NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
|
nativeSearchQueryBuilder.withQuery(QueryBuilders.matchAllQuery());
|
nativeSearchQueryBuilder.withPageable(PageRequest.of(i, 1000));
|
try {
|
SearchHits<EsGoodsIndex> esGoodsIndices = this.restTemplate.search(nativeSearchQueryBuilder.build(), EsGoodsIndex.class);
|
if (esGoodsIndices.isEmpty() || esGoodsIndices.getSearchHits().isEmpty()) {
|
break;
|
}
|
for (SearchHit<EsGoodsIndex> searchHit : esGoodsIndices.getSearchHits()) {
|
EsGoodsIndex goodsIndex = searchHit.getContent();
|
UpdateRequest updateRequest = this.removePromotionByPromotionKey(goodsIndex, promotionsKey);
|
if (updateRequest != null) {
|
bulkRequest.add(updateRequest);
|
}
|
}
|
this.executeBulkUpdateRequest(bulkRequest);
|
} catch (Exception e) {
|
log.error("删除索引中指定的促销活动id的促销活动失败！key: {}", promotionsKey, e);
|
return;
|
}
|
}
|
});
|
}
|
|
/**
|
* 从索引中删除指定促销活动id的促销活动
|
*
|
* @param goodsIndex 索引
|
* @param promotionsKey 促销活动key
|
*/
|
private UpdateRequest removePromotionByPromotionKey(EsGoodsIndex goodsIndex, String promotionsKey) {
|
Map<String, Object> promotionMap = goodsIndex.getOriginPromotionMap();
|
if (promotionMap != null && !promotionMap.isEmpty()) {
|
//如果存在同促销ID的活动删除
|
Map<String, Object> filterPromotionMap = promotionMap.entrySet().stream().filter(i -> !i.getKey().equals(promotionsKey)).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
|
return this.getGoodsIndexPromotionUpdateRequest(goodsIndex.getId(), filterPromotionMap);
|
}
|
return null;
|
}
|
|
/**
|
* 清除所有商品索引的无效促销活动
|
*/
|
@Override
|
public void cleanInvalidPromotion() {
|
ThreadUtil.execAsync(this::executeCleanInvalidPromotions);
|
}
|
|
private void executeCleanInvalidPromotions() {
|
for (int i = 1; ; i++) {
|
org.springframework.data.domain.Page<EsGoodsIndex> all = goodsIndexRepository.findAll(PageRequest.of(i, 1000));
|
if (all.isEmpty()) {
|
break;
|
}
|
for (EsGoodsIndex goodsIndex : all.toList()) {
|
Map<String, Object> promotionMap = goodsIndex.getOriginPromotionMap();
|
//获取商品索引
|
if (promotionMap != null && !promotionMap.isEmpty()) {
|
//促销不为空则进行清洗
|
promotionMap.entrySet().removeIf(j -> {
|
JSONObject promotionJson = JSONUtil.parseObj(j.getValue());
|
BasePromotions promotion = promotionJson.toBean(BasePromotions.class);
|
return promotion.getEndTime() != null && promotion.getEndTime().getTime() < DateUtil.date().getTime();
|
});
|
}
|
}
|
goodsIndexRepository.saveAll(all);
|
}
|
}
|
|
|
@Override
|
public EsGoodsIndex findById(String id) {
|
Optional<EsGoodsIndex> goodsIndex = goodsIndexRepository.findById(id);
|
if (!goodsIndex.isPresent()) {
|
log.error("商品skuId为" + id + "的es索引不存在！");
|
return null;
|
}
|
return goodsIndex.get();
|
}
|
|
/**
|
* 根据id获取商品索引信息的促销信息
|
*
|
* @param id skuId
|
* @return 促销信息map
|
*/
|
@Override
|
public Map<String, Object> getPromotionMap(String id) {
|
EsGoodsIndex goodsIndex = this.findById(id);
|
|
//如果商品索引不为空，返回促销信息，否则返回空
|
if (goodsIndex != null) {
|
Map<String, Object> promotionMap = goodsIndex.getOriginPromotionMap();
|
if (promotionMap == null || promotionMap.isEmpty()) {
|
return new HashMap<>(16);
|
}
|
return promotionMap;
|
}
|
return new HashMap<>();
|
}
|
|
/**
|
* 根据id获取商品索引信息的指定促销活动的id
|
*
|
* @param id skuId
|
* @param promotionTypeEnum 促销活动类型
|
* @return 当前商品参与的促销活动id集合
|
*/
|
@Override
|
public List<String> getPromotionIdByPromotionType(String id, PromotionTypeEnum promotionTypeEnum) {
|
Map<String, Object> promotionMap = this.getPromotionMap(id);
|
//如果没有促销信息，则返回新的
|
if (promotionMap == null || promotionMap.isEmpty()) {
|
return new ArrayList<>();
|
}
|
//对促销进行过滤
|
List<String> keyCollect = promotionMap.keySet().stream().filter(i -> i.contains(promotionTypeEnum.name())).collect(Collectors.toList());
|
List<String> promotionIds = new ArrayList<>();
|
//写入促销id
|
for (String key : keyCollect) {
|
BasePromotions promotion = (BasePromotions) promotionMap.get(key);
|
promotionIds.add(promotion.getId());
|
}
|
return promotionIds;
|
}
|
|
/**
|
* 获取重置的商品索引
|
*
|
* @param goodsSku 商品sku信息
|
* @param goodsParamDTOS 商品参数
|
* @return 商品索引
|
*/
|
@Override
|
public EsGoodsIndex getResetEsGoodsIndex(GoodsSku goodsSku, List<GoodsParamsDTO> goodsParamDTOS) {
|
EsGoodsIndex index = new EsGoodsIndex(goodsSku, goodsParamDTOS);
|
//获取活动信息
|
Map<String, Object> goodsCurrentPromotionMap = promotionService.getGoodsSkuPromotionMap(index.getStoreId(), index.getId());
|
//写入促销信息
|
index.setPromotionMapJson(JSONUtil.toJsonStr(goodsCurrentPromotionMap));
|
|
//发送mq消息
|
String destination = rocketmqCustomProperties.getGoodsTopic() + ":" + GoodsTagsEnum.RESET_GOODS_INDEX.name();
|
rocketMQTemplate.asyncSend(destination, JSONUtil.toJsonStr(Collections.singletonList(index)), RocketmqSendCallbackBuilder.commonCallback());
|
return index;
|
}
|
|
/**
|
* 修改商品活动索引
|
*
|
* @param goodsIndex 商品索引
|
* @param key 关键字
|
* @param promotion 活动
|
*/
|
private UpdateRequest updateGoodsIndexPromotion(EsGoodsIndex goodsIndex, String key, BasePromotions promotion) {
|
Map<String, Object> promotionMap;
|
//数据非空处理，如果空给一个新的信息
|
if (goodsIndex.getOriginPromotionMap() == null || goodsIndex.getOriginPromotionMap().isEmpty()) {
|
promotionMap = new HashMap<>(1);
|
} else {
|
promotionMap = goodsIndex.getOriginPromotionMap();
|
}
|
// log.info("ES修改商品活动索引-原商品索引信息:{}", goodsIndex);
|
// log.info("ES修改商品活动索引-原商品索引活动信息:{}", promotionMap);
|
//如果活动已结束
|
if (promotion.getPromotionStatus().equals(PromotionsStatusEnum.END.name()) || promotion.getPromotionStatus().equals(PromotionsStatusEnum.CLOSE.name())) {//如果存在活动
|
//删除活动
|
promotionMap.remove(key);
|
} else {
|
promotionMap.put(key, promotion);
|
}
|
// log.info("ES修改商品活动索引-过滤后商品索引活动信息:{}", promotionMap);
|
return this.getGoodsIndexPromotionUpdateRequest(goodsIndex.getId(), promotionMap);
|
}
|
|
/**
|
* 以更新部分字段的方式更新索引促销信息
|
*
|
* @param id 索引id
|
* @param promotionMap 促销信息
|
*/
|
private UpdateRequest getGoodsIndexPromotionUpdateRequest(String id, Map<String, Object> promotionMap) {
|
UpdateRequest updateRequest = new UpdateRequest();
|
updateRequest.index(getIndexName());
|
updateRequest.id(id);
|
updateRequest.retryOnConflict(3);
|
// updateRequest.version(promotionMap.size());
|
Map<String, Object> params = new HashMap<>();
|
params.put("promotionMap", JSONUtil.toJsonStr(promotionMap));
|
Script script = new Script(ScriptType.INLINE, "painless", "ctx._source.promotionMapJson=params.promotionMap;", params);
|
updateRequest.script(script);
|
return updateRequest;
|
}
|
|
/**
|
* 执行批量更新商品索引
|
*
|
* @param bulkRequest 批量请求
|
*/
|
private void executeBulkUpdateRequest(BulkRequest bulkRequest) {
|
if (bulkRequest.requests().isEmpty()) {
|
return;
|
}
|
this.client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
|
@Override
|
public void onResponse(BulkResponse bulkItemResponses) {
|
if (bulkItemResponses.hasFailures()) {
|
log.info("批量更新商品索引的促销信息中出现部分异常：{}", bulkItemResponses.buildFailureMessage());
|
} else {
|
log.info("批量更新商品索引的促销信息结果：{}", bulkItemResponses.status());
|
}
|
}
|
|
@Override
|
public void onFailure(Exception e) {
|
log.error("批量更新商品索引的促销信息出现异常！", e);
|
}
|
});
|
|
}
|
|
/**
|
* 移除需要移除的促销活动
|
*
|
* @param currentKey 当前操作的促销活动key
|
* @param promotionMap 促销活动
|
* @param needRemoveKeys 需移除的促销活动
|
*/
|
private void removePromotionKey(String currentKey, Map<String, Object> promotionMap, String... needRemoveKeys) {
|
//判定是否需要移除
|
if (CharSequenceUtil.containsAny(currentKey, needRemoveKeys)) {
|
|
List<String> removeKeys = new ArrayList<>();
|
//促销循环
|
for (String entry : promotionMap.keySet()) {
|
//需要移除则进行移除处理
|
for (String needRemoveKey : needRemoveKeys) {
|
if (entry.contains(needRemoveKey) && currentKey.contains(needRemoveKey)) {
|
removeKeys.add(entry);
|
break;
|
}
|
}
|
}
|
//移除促销信息
|
removeKeys.forEach(promotionMap.keySet()::remove);
|
}
|
}
|
|
private String getIndexName() {
|
//索引名称拼接
|
return elasticsearchProperties.getIndexPrefix() + "_" + EsSuffix.GOODS_INDEX_NAME;
|
}
|
|
private EsGoodsIndex wrapperEsGoodsIndex(GoodsSkuDTO goodsSku, List<Map<String, Object>> brandList, List<Map<String, Object>> categoryList, List<Map<String, Object>> storeCategoryList) {
|
EsGoodsIndex index = new EsGoodsIndex(goodsSku);
|
|
//商品参数索引
|
if (CharSequenceUtil.isNotEmpty(goodsSku.getParams())) {
|
List<GoodsParamsDTO> goodsParamDTOS = JSONUtil.toList(goodsSku.getParams(), GoodsParamsDTO.class);
|
index = new EsGoodsIndex(goodsSku, goodsParamDTOS);
|
}
|
//商品分类索引
|
if (CollUtil.isNotEmpty(categoryList) && CharSequenceUtil.isNotEmpty(goodsSku.getCategoryPath())) {
|
StringBuilder categoryNamePath = new StringBuilder();
|
categoryList.stream().filter(o -> goodsSku.getCategoryPath().contains(o.get("id").toString())).forEach(p -> categoryNamePath.append(p.get("name")).append(","));
|
if (CharSequenceUtil.isNotEmpty(categoryNamePath)) {
|
categoryNamePath.deleteCharAt(categoryNamePath.length() - 1);
|
index.setCategoryNamePath(categoryNamePath.toString());
|
}
|
}
|
//商品品牌索引
|
if (CollUtil.isNotEmpty(brandList) && CharSequenceUtil.isNotEmpty(goodsSku.getBrandId())) {
|
Optional<Map<String, Object>> brandInfo = brandList.stream().filter(p -> p.get("id").equals(goodsSku.getBrandId())).findFirst();
|
if (brandInfo.isPresent()) {
|
index.setBrandName(brandInfo.get().get("name").toString());
|
index.setBrandUrl(brandInfo.get().get("logo").toString());
|
}
|
}
|
//店铺分类索引
|
if (CollUtil.isNotEmpty(storeCategoryList) && CharSequenceUtil.isNotEmpty(goodsSku.getStoreCategoryPath())) {
|
StringBuilder storeCategoryNamePath = new StringBuilder();
|
storeCategoryList.stream().filter(o -> goodsSku.getStoreCategoryPath().contains(o.get("id").toString())).forEach(p -> storeCategoryNamePath.append(p.get("label_name").toString()).append(","));
|
if (CharSequenceUtil.isNotEmpty(storeCategoryNamePath)) {
|
storeCategoryNamePath.deleteCharAt(storeCategoryNamePath.length() - 1);
|
index.setStoreCategoryNamePath(storeCategoryNamePath.toString());
|
}
|
}
|
return index;
|
}
|
}
|
