caojiahao/dev_caojiahao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

!257 优化生成索引，增加批量生成参数

Browse Source 
Merge pull request !257 from OceansDeep/pg
This commit is contained in:
OceansDeep 2023-02-21 04:09:16 +00:00 committed by Gitee
commit 0a3696ac56
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
1 changed files with 63 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

107
consumer/src/main/java/cn/lili/listener/GoodsMessageListener.java
View File

@ -8,6 +8,7 @@ import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil; import cn.hutool.json.JSONUtil;
import cn.lili.common.aop.annotation.RetryOperation; import cn.lili.common.aop.annotation.RetryOperation;
import cn.lili.common.exception.RetryException; import cn.lili.common.exception.RetryException;
import cn.lili.common.vo.PageVO;
import cn.lili.event.GoodsCommentCompleteEvent; import cn.lili.event.GoodsCommentCompleteEvent;
import cn.lili.modules.distribution.entity.dos.DistributionGoods; import cn.lili.modules.distribution.entity.dos.DistributionGoods;
import cn.lili.modules.distribution.entity.dto.DistributionGoodsSearchParams; import cn.lili.modules.distribution.entity.dto.DistributionGoodsSearchParams;
@ -33,6 +34,8 @@ import cn.lili.modules.promotion.service.PromotionService;
import cn.lili.modules.search.entity.dos.EsGoodsIndex; import cn.lili.modules.search.entity.dos.EsGoodsIndex;
import cn.lili.modules.search.service.EsGoodsIndexService; import cn.lili.modules.search.service.EsGoodsIndexService;
import cn.lili.rocketmq.tags.GoodsTagsEnum; import cn.lili.rocketmq.tags.GoodsTagsEnum;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
@ -54,6 +57,8 @@ import java.util.stream.Collectors;
@RocketMQMessageListener(topic = "${lili.data.rocketmq.goods-topic}", consumerGroup = "${lili.data.rocketmq.goods-group}") @RocketMQMessageListener(topic = "${lili.data.rocketmq.goods-topic}", consumerGroup = "${lili.data.rocketmq.goods-group}")
public class GoodsMessageListener implements RocketMQListener<MessageExt> { public class GoodsMessageListener implements RocketMQListener<MessageExt> {
private static final int BATCH_SIZE = 10;
/** /**
* ES商品 * ES商品
*/ */
@ -175,10 +180,8 @@ public class GoodsMessageListener implements RocketMQListener<MessageExt> {
try { try {
String updateIndexFieldsJsonStr = new String(messageExt.getBody()); String updateIndexFieldsJsonStr = new String(messageExt.getBody());
JSONObject updateIndexFields = JSONUtil.parseObj(updateIndexFieldsJsonStr); JSONObject updateIndexFields = JSONUtil.parseObj(updateIndexFieldsJsonStr);
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked") Map<String, Object> queryFields = updateIndexFields.get("queryFields", Map.class);
Map<String, Object> queryFields = updateIndexFields.get("queryFields", Map.class); @SuppressWarnings("unchecked") Map<String, Object> updateFields = updateIndexFields.get("updateFields", Map.class);
@SuppressWarnings("unchecked")
Map<String, Object> updateFields = updateIndexFields.get("updateFields", Map.class);
goodsIndexService.updateIndex(queryFields, updateFields); goodsIndexService.updateIndex(queryFields, updateFields);
} catch (Exception e) { } catch (Exception e) {
log.error("更新商品索引事件执行异常，商品信息: " + new String(messageExt.getBody()), e); log.error("更新商品索引事件执行异常，商品信息: " + new String(messageExt.getBody()), e);
@ -238,10 +241,7 @@ public class GoodsMessageListener implements RocketMQListener<MessageExt> {
try { try {
goodsCommentCompleteEvent.goodsComment(memberEvaluation); goodsCommentCompleteEvent.goodsComment(memberEvaluation);
} catch (Exception e) { } catch (Exception e) {
log.error("评价{},在{}业务中，状态修改事件执行异常", log.error("评价{},在{}业务中，状态修改事件执行异常", new String(messageExt.getBody()), goodsCommentCompleteEvent.getClass().getName(), e);
new String(messageExt.getBody()),
goodsCommentCompleteEvent.getClass().getName(),
e);
} }
} }
break; break;
@ -260,26 +260,42 @@ public class GoodsMessageListener implements RocketMQListener<MessageExt> {
log.info("更新商品索引促销信息: {}", promotionsJsonStr); log.info("更新商品索引促销信息: {}", promotionsJsonStr);
JSONObject jsonObject = JSONUtil.parseObj(promotionsJsonStr); JSONObject jsonObject = JSONUtil.parseObj(promotionsJsonStr);
// 转换为详细的促销信息（注：促销信息必须继承自 BasePromotions，且必须保证派生类存在与sdk包下） // 转换为详细的促销信息（注：促销信息必须继承自 BasePromotions，且必须保证派生类存在与sdk包下）
BasePromotions promotions = (BasePromotions) jsonObject.get("promotions", BasePromotions promotions = (BasePromotions) jsonObject.get("promotions", ClassLoaderUtil.loadClass(jsonObject.get("promotionsType").toString()));
ClassLoaderUtil.loadClass(jsonObject.get("promotionsType").toString()));
// 获取促销唯一key,由 促销类型 + 促销id 组成 // 获取促销唯一key,由 促销类型 + 促销id 组成
String esPromotionKey = jsonObject.get("esPromotionKey").toString(); String esPromotionKey = jsonObject.get("esPromotionKey").toString();
if (PromotionsScopeTypeEnum.PORTION_GOODS.name().equals(promotions.getScopeType())) { if (PromotionsScopeTypeEnum.PORTION_GOODS.name().equals(promotions.getScopeType())) {
PromotionGoodsSearchParams searchParams = new PromotionGoodsSearchParams(); for (int i = 0; ; i++) {
searchParams.setPromotionId(promotions.getId()); PromotionGoodsSearchParams searchParams = new PromotionGoodsSearchParams();
List<PromotionGoods> promotionGoodsList = this.promotionGoodsService.listFindAll(searchParams); searchParams.setPromotionId(promotions.getId());
List<String> skuIds = promotionGoodsList.stream().map(PromotionGoods::getSkuId).collect(Collectors.toList()); PageVO pageVO = new PageVO();
// 更新商品索引促销信息（删除原索引中相关的促销信息，更新索引中促销信息） pageVO.setPageNumber(i);
this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(skuIds, esPromotionKey); pageVO.setPageSize(BATCH_SIZE);
this.goodsIndexService.updateEsGoodsIndexByList(promotionGoodsList, promotions, esPromotionKey); Page<PromotionGoods> promotionGoodsPage = this.promotionGoodsService.pageFindAll(searchParams, pageVO);
if (promotionGoodsPage == null || promotionGoodsPage.getRecords().isEmpty()) {
break;
}
List<String> skuIds = promotionGoodsPage.getRecords().stream().map(PromotionGoods::getSkuId).collect(Collectors.toList());
// 更新商品索引促销信息（删除原索引中相关的促销信息，更新索引中促销信息）
this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(skuIds, esPromotionKey);
this.goodsIndexService.updateEsGoodsIndexByList(promotionGoodsPage.getRecords(), promotions, esPromotionKey);
}
} else if (PromotionsScopeTypeEnum.PORTION_GOODS_CATEGORY.name().equals(promotions.getScopeType())) { } else if (PromotionsScopeTypeEnum.PORTION_GOODS_CATEGORY.name().equals(promotions.getScopeType())) {
GoodsSearchParams searchParams = new GoodsSearchParams(); for (int i = 0; ; i++) {
searchParams.setCategoryPath(promotions.getScopeId()); GoodsSearchParams searchParams = new GoodsSearchParams();
List<GoodsSku> goodsSkuByList = this.goodsSkuService.getGoodsSkuByList(searchParams); searchParams.setCategoryPath(promotions.getScopeId());
List<String> skuIds = goodsSkuByList.stream().map(GoodsSku::getId).collect(Collectors.toList()); searchParams.setPageNumber(i);
// 更新商品索引促销信息（删除原索引中相关的促销信息，更新索引中促销信息） searchParams.setPageSize(BATCH_SIZE);
this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(skuIds, esPromotionKey); IPage<GoodsSku> goodsSkuByPage = this.goodsSkuService.getGoodsSkuByPage(searchParams);
this.goodsIndexService.updateEsGoodsIndexPromotions(skuIds, promotions, esPromotionKey); if (goodsSkuByPage == null || goodsSkuByPage.getRecords().isEmpty()) {
break;
}
List<String> skuIds = goodsSkuByPage.getRecords().stream().map(GoodsSku::getId).collect(Collectors.toList());
// 更新商品索引促销信息（删除原索引中相关的促销信息，更新索引中促销信息）
this.goodsIndexService.deleteEsGoodsPromotionByPromotionKey(skuIds, esPromotionKey);
this.goodsIndexService.updateEsGoodsIndexPromotions(skuIds, promotions, esPromotionKey);
}
} else if (PromotionsScopeTypeEnum.ALL.name().equals(promotions.getScopeType())) { } else if (PromotionsScopeTypeEnum.ALL.name().equals(promotions.getScopeType())) {
this.goodsIndexService.updateEsGoodsIndexAllByList(promotions, esPromotionKey); this.goodsIndexService.updateEsGoodsIndexAllByList(promotions, esPromotionKey);
} }
@ -321,26 +337,31 @@ public class GoodsMessageListener implements RocketMQListener<MessageExt> {
* @param goods 商品消息 * @param goods 商品消息
*/ */
private void updateGoodsIndex(Goods goods) { private void updateGoodsIndex(Goods goods) {
//如果商品通过审核&&并且已上架 for (int i = 1; ; i++) {
GoodsSearchParams searchParams = new GoodsSearchParams(); //如果商品通过审核&&并且已上架
searchParams.setGoodsId(goods.getId()); GoodsSearchParams searchParams = new GoodsSearchParams();
List<GoodsSku> goodsSkuList = this.goodsSkuService.getGoodsSkuByList(searchParams); searchParams.setGoodsId(goods.getId());
log.info("goods：{}", goods); searchParams.setPageNumber(i);
log.info("goodsSkuList：{}", goodsSkuList); searchParams.setPageSize(BATCH_SIZE);
if (goods.getAuthFlag().equals(GoodsAuthEnum.PASS.name()) IPage<GoodsSku> goodsSkuByPage = this.goodsSkuService.getGoodsSkuByPage(searchParams);
&& goods.getMarketEnable().equals(GoodsStatusEnum.UPPER.name()) if (goodsSkuByPage == null || goodsSkuByPage.getRecords().isEmpty()) {
&& Boolean.FALSE.equals(goods.getDeleteFlag())) { break;
this.generatorGoodsIndex(goods, goodsSkuList); }
} log.info("goods：{}", goods);
//如果商品状态值不支持es搜索，那么将商品信息做下架处理 log.info("goodsSkuList：{}", goodsSkuByPage.getRecords());
else { if (goods.getAuthFlag().equals(GoodsAuthEnum.PASS.name()) && goods.getMarketEnable().equals(GoodsStatusEnum.UPPER.name()) && Boolean.FALSE.equals(goods.getDeleteFlag())) {
for (GoodsSku goodsSku : goodsSkuList) { this.generatorGoodsIndex(goods, goodsSkuByPage.getRecords());
EsGoodsIndex esGoodsOld = goodsIndexService.findById(goodsSku.getId()); } else {
if (esGoodsOld != null) { //如果商品状态值不支持es搜索，那么将商品信息做下架处理
goodsIndexService.deleteIndexById(goodsSku.getId()); for (GoodsSku goodsSku : goodsSkuByPage.getRecords()) {
EsGoodsIndex esGoodsOld = goodsIndexService.findById(goodsSku.getId());
if (esGoodsOld != null) {
goodsIndexService.deleteIndexById(goodsSku.getId());
}
} }
} }
} }
} }
/** /**
@ -462,9 +483,7 @@ public class GoodsMessageListener implements RocketMQListener<MessageExt> {
goodsSku.setBuyCount(buyCount); goodsSku.setBuyCount(buyCount);
goodsSkuService.update(goodsSku); goodsSkuService.update(goodsSku);
this.goodsIndexService.updateIndex( this.goodsIndexService.updateIndex(MapUtil.builder(new HashMap<String, Object>()).put("id", goodsCompleteMessage.getSkuId()).build(), MapUtil.builder(new HashMap<String, Object>()).put("buyCount", buyCount).build());
MapUtil.builder(new HashMap<String, Object>()).put("id", goodsCompleteMessage.getSkuId()).build(),
MapUtil.builder(new HashMap<String, Object>()).put("buyCount", buyCount).build());
} else { } else {
log.error("商品SkuId为[" + goodsCompleteMessage.getGoodsId() + "的商品不存在，更新商品失败！"); log.error("商品SkuId为[" + goodsCompleteMessage.getGoodsId() + "的商品不存在，更新商品失败！");