优化更新es商品促销流程

consumer/src/main/java/cn/lili/listener/GoodsMessageListener.java

@@ -144,11 +144,15 @@ public class GoodsMessageListener implements RocketMQListener<MessageExt> {
                 this.updateGoodsIndexPromotions(new String(messageExt.getBody()));
                 break;
             case DELETE_GOODS_INDEX_PROMOTIONS:
-                BasePromotions promotions = JSONUtil.toBean(new String(messageExt.getBody()), BasePromotions.class);
+                JSONObject jsonObject = JSONUtil.parseObj(new String(messageExt.getBody()));
-                if (CharSequenceUtil.isNotEmpty(promotions.getScopeId())) {
+                String promotionKey = jsonObject.getStr("promotionKey");
-                    this.goodsIndexService.deleteEsGoodsPromotionByPromotionId(Arrays.asList(promotions.getScopeId().split(",")), promotions.getId());
+                if (CharSequenceUtil.isEmpty(promotionKey)) {
+                    break;
+                }
+                if (CharSequenceUtil.isNotEmpty(jsonObject.getStr("scopeId"))) {
+                    this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(Arrays.asList(jsonObject.getStr("scopeId").split(",")), promotionKey);
                 } else {
-                    this.goodsIndexService.deleteEsGoodsPromotionByPromotionId(null, promotions.getId());
+                    this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(promotionKey);
                 }
                 break;
             case UPDATE_GOODS_INDEX:
@@ -252,16 +256,17 @@ public class GoodsMessageListener implements RocketMQListener<MessageExt> {
                 searchParams.setPromotionId(promotions.getId());
                 List<PromotionGoods> promotionGoodsList = this.promotionGoodsService.listFindAll(searchParams);
                 List<String> skuIds = promotionGoodsList.stream().map(PromotionGoods::getSkuId).collect(Collectors.toList());
-                this.goodsIndexService.deleteEsGoodsPromotionByPromotionId(skuIds, promotions.getId());
+                this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(skuIds, esPromotionKey);
                 this.goodsIndexService.updateEsGoodsIndexByList(promotionGoodsList, promotions, esPromotionKey);
             } else if (PromotionsScopeTypeEnum.PORTION_GOODS_CATEGORY.name().equals(promotions.getScopeType())) {
                 GoodsSearchParams searchParams = new GoodsSearchParams();
                 searchParams.setCategoryPath(promotions.getScopeId());
                 List<GoodsSku> goodsSkuByList = this.goodsSkuService.getGoodsSkuByList(searchParams);
                 List<String> skuIds = goodsSkuByList.stream().map(GoodsSku::getId).collect(Collectors.toList());
-                this.goodsIndexService.deleteEsGoodsPromotionByPromotionId(skuIds, promotions.getId());
+                this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(skuIds, esPromotionKey);
                 this.goodsIndexService.updateEsGoodsIndexPromotions(skuIds, promotions, esPromotionKey);
             } else if (PromotionsScopeTypeEnum.ALL.name().equals(promotions.getScopeType())) {
+                this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(esPromotionKey);
                 this.goodsIndexService.updateEsGoodsIndexAllByList(promotions, esPromotionKey);
             }
         } catch (Exception e) {

framework/src/main/java/cn/lili/modules/promotion/serviceimpl/AbstractPromotionsServiceImpl.java

@@ -1,5 +1,6 @@
 package cn.lili.modules.promotion.serviceimpl;
 
+import cn.hutool.core.map.MapBuilder;
 import cn.hutool.core.text.CharSequenceUtil;
 import cn.hutool.json.JSONUtil;
 import cn.lili.common.enums.ResultCode;
@@ -254,10 +255,11 @@ public abstract class AbstractPromotionsServiceImpl<M extends BaseMapper<T>, T e
     @Override
     public void updateEsGoodsIndex(T promotions) {
         if (promotions.getStartTime() == null && promotions.getEndTime() == null) {
+            Map<Object, Object> build = MapBuilder.create().put("promotionKey", this.getPromotionType() + "-" + promotions.getId()).put("scopeId", promotions.getScopeId()).build();
             //删除商品促销消息
             String destination = rocketmqCustomProperties.getGoodsTopic() + ":" + GoodsTagsEnum.DELETE_GOODS_INDEX_PROMOTIONS.name();
             //发送mq消息
-            rocketMQTemplate.asyncSend(destination, JSONUtil.toJsonStr(promotions), RocketmqSendCallbackBuilder.commonCallback());
+            rocketMQTemplate.asyncSend(destination, JSONUtil.toJsonStr(build), RocketmqSendCallbackBuilder.commonCallback());
         } else {
 
             String esPromotionKey = this.getPromotionType().name() + "-" + promotions.getId();

framework/src/main/java/cn/lili/modules/promotion/serviceimpl/SeckillServiceImpl.java

@@ -3,6 +3,7 @@ package cn.lili.modules.promotion.serviceimpl;
 import cn.hutool.core.date.DatePattern;
 import cn.hutool.core.date.DateTime;
 import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.map.MapBuilder;
 import cn.hutool.core.text.CharSequenceUtil;
 import cn.hutool.core.util.ArrayUtil;
 import cn.hutool.json.JSONUtil;
@@ -14,7 +15,6 @@ import cn.lili.modules.promotion.entity.dos.Seckill;
 import cn.lili.modules.promotion.entity.dos.SeckillApply;
 import cn.lili.modules.promotion.entity.dto.search.SeckillSearchParams;
 import cn.lili.modules.promotion.entity.enums.PromotionsApplyStatusEnum;
-import cn.lili.modules.promotion.entity.enums.PromotionsScopeTypeEnum;
 import cn.lili.modules.promotion.entity.vos.SeckillVO;
 import cn.lili.modules.promotion.mapper.SeckillMapper;
 import cn.lili.modules.promotion.service.SeckillApplyService;
@@ -40,6 +40,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 /**
  * 秒杀活动业务层实现
@@ -188,12 +189,11 @@ public class SeckillServiceImpl extends AbstractPromotionsServiceImpl<SeckillMap
      */
     @Override
     public void deleteEsGoodsSeckill(Seckill seckill, List<String> skuIds) {
-        seckill.setScopeType(PromotionsScopeTypeEnum.PORTION_GOODS.name());
+        Map<Object, Object> build = MapBuilder.create().put("promotionKey", this.getPromotionType() + "-" + seckill.getId()).put("scopeId", ArrayUtil.join(skuIds.toArray(), ",")).build();
-        seckill.setScopeId(ArrayUtil.join(skuIds.toArray(), ","));
         //删除商品促销消息
         String destination = rocketmqCustomProperties.getGoodsTopic() + ":" + GoodsTagsEnum.DELETE_GOODS_INDEX_PROMOTIONS.name();
         //发送mq消息
-        rocketMQTemplate.asyncSend(destination, JSONUtil.toJsonStr(seckill), RocketmqSendCallbackBuilder.commonCallback());
+        rocketMQTemplate.asyncSend(destination, JSONUtil.toJsonStr(build), RocketmqSendCallbackBuilder.commonCallback());
     }
 
     @Override

framework/src/main/java/cn/lili/modules/search/service/EsGoodsIndexService.java

@@ -143,9 +143,17 @@ public interface EsGoodsIndexService {
      * 删除索引中指定的促销活动id的促销活动
      *
      * @param skuIds      商品skuId
-     * @param promotionId 促销活动Id
+     * @param promotionsKey 促销活动Key
      */
-    void deleteEsGoodsPromotionByPromotionId(List<String> skuIds, String promotionId);
+    void deleteEsGoodsPromotionByPromotionKey(List<String> skuIds, String promotionsKey);
+
+
+    /**
+     * 删除索引中指定的促销活动id的促销活动
+     *
+     * @param promotionsKey 促销活动Key
+     */
+    void deleteEsGoodsPromotionByPromotionKey(String promotionsKey);
 
     /**
      * 清除所以商品索引的无效促销活动

framework/src/main/java/cn/lili/modules/search/serviceimpl/EsGoodsIndexServiceImpl.java

@@ -389,9 +389,8 @@ public class EsGoodsIndexServiceImpl extends BaseElasticsearchService implements
     @Override
     public void updateEsGoodsIndexPromotions(List<String> ids, BasePromotions promotion, String key) {
         BulkRequest bulkRequest = new BulkRequest();
-        log.info("修改商品活动索引");
+        log.info("更新商品索引的促销信息----------");
         log.info("商品ids: {}", ids);
-        log.info("活动关键字: {}", key);
         log.info("活动: {}", promotion);
         for (String id : ids) {
             UpdateRequest updateRequest = this.updateEsGoodsIndexPromotions(id, promotion, key);
@@ -449,7 +448,6 @@ public class EsGoodsIndexServiceImpl extends BaseElasticsearchService implements
             goodsIndices = new ArrayList<>(IterableUtil.toCollection(all));
         }
         List<String> skuIds = goodsIndices.stream().map(EsGoodsIndex::getId).collect(Collectors.toList());
-        this.deleteEsGoodsPromotionByPromotionId(skuIds, promotion.getId());
         this.updateEsGoodsIndexPromotions(skuIds, promotion, key);
     }
 
@@ -475,17 +473,19 @@ public class EsGoodsIndexServiceImpl extends BaseElasticsearchService implements
     }
 
     @Override
-    public void deleteEsGoodsPromotionByPromotionId(List<String> skuIds, String promotionId) {
+    public void deleteEsGoodsPromotionByPromotionKey(List<String> skuIds, String promotionsKey) {
         BulkRequest bulkRequest = new BulkRequest();
         log.info("删除商品活动索引");
         log.info("商品skuIds: {}", skuIds);
-        log.info("活动Id: {}", promotionId);
+        log.info("活动Key: {}", promotionsKey);
-        if (skuIds != null && !skuIds.isEmpty()) {
+        if (skuIds == null || skuIds.isEmpty()) {
+            return;
+        }
         for (String skuId : skuIds) {
             EsGoodsIndex goodsIndex = findById(skuId);
             //商品索引不为空
             if (goodsIndex != null) {
-                    UpdateRequest updateRequest = this.removePromotionByPromotionId(goodsIndex, promotionId);
+                UpdateRequest updateRequest = this.removePromotionByPromotionKey(goodsIndex, promotionsKey);
                 if (updateRequest != null) {
                     bulkRequest.add(updateRequest);
                 }
@@ -493,29 +493,37 @@ public class EsGoodsIndexServiceImpl extends BaseElasticsearchService implements
                 log.error("更新索引商品促销信息失败！skuId 为 【{}】的索引不存在！", skuId);
             }
         }
-        } else {
+        this.executeBulkUpdateRequest(bulkRequest);
+    }
+
+    /**
+     * 删除索引中指定的促销活动id的促销活动
+     *
+     * @param promotionsKey 促销活动Key
+     */
+    @Override
+    public void deleteEsGoodsPromotionByPromotionKey(String promotionsKey) {
+        BulkRequest bulkRequest = new BulkRequest();
         for (EsGoodsIndex goodsIndex : this.goodsIndexRepository.findAll()) {
-                UpdateRequest updateRequest = this.removePromotionByPromotionId(goodsIndex, promotionId);
+            UpdateRequest updateRequest = this.removePromotionByPromotionKey(goodsIndex, promotionsKey);
             if (updateRequest != null) {
                 bulkRequest.add(updateRequest);
             }
         }
-        }
         this.executeBulkUpdateRequest(bulkRequest);
-
     }
 
     /**
      * 从索引中删除指定促销活动id的促销活动
      *
      * @param goodsIndex  索引
-     * @param promotionId 促销活动id
+     * @param promotionsKey 促销活动key
      */
-    private UpdateRequest removePromotionByPromotionId(EsGoodsIndex goodsIndex, String promotionId) {
+    private UpdateRequest removePromotionByPromotionKey(EsGoodsIndex goodsIndex, String promotionsKey) {
         Map<String, Object> promotionMap = goodsIndex.getPromotionMap();
         if (promotionMap != null && !promotionMap.isEmpty()) {
             //如果存在同促销ID的活动删除
-            List<String> collect = promotionMap.keySet().stream().filter(i -> i.split("-")[1].equals(promotionId)).collect(Collectors.toList());
+            List<String> collect = promotionMap.keySet().stream().filter(i -> i.equals(promotionsKey)).collect(Collectors.toList());
             collect.forEach(promotionMap::remove);
             goodsIndex.setPromotionMapJson(JSONUtil.toJsonStr(promotionMap));
             return this.getGoodsIndexPromotionUpdateRequest(goodsIndex.getId(), promotionMap);
@@ -662,7 +670,8 @@ public class EsGoodsIndexServiceImpl extends BaseElasticsearchService implements
         UpdateRequest updateRequest = new UpdateRequest();
         updateRequest.index(getIndexName());
         updateRequest.id(id);
-        updateRequest.retryOnConflict(5);
+//        updateRequest.retryOnConflict(5);
+//        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);
@@ -684,7 +693,7 @@ public class EsGoodsIndexServiceImpl extends BaseElasticsearchService implements
             if (responses.hasFailures()) {
                 log.info("批量更新商品索引的促销信息中出现部分异常：{}", responses.buildFailureMessage());
             } else {
-                log.info("批量更新商品索引的促销信息结果：{}", responses);
+                log.info("批量更新商品索引的促销信息结果：{}", responses.status());
             }
         } catch (IOException e) {
             log.error("批量更新商品索引的促销信息出现异常！", e);