yoho-search / yoho-search-consumer · Commits

Go to a project

Authored by wangnan
Commit 4e43f55bfc64110d85ed8c2533e6706ae07f1fb7 2 parents 26291e55 89cc6085

Merge branch 'wn_change_skn_status' into 0310

Inline Side-by-side
Showing 6 changed files with 229 additions and 199 deletions
package com.yoho.search.consumer.index.increment;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.yoho.error.event.SearchEvent;
... ... @@ -24,122 +10,177 @@ import com.yoho.search.consumer.common.CostStatistics;
import com.yoho.search.consumer.index.increment.bulks.StorageSkuIndexBulkService;
import com.yoho.search.consumer.index.increment.rule.AbstractStorageRelatedMqListener;
import com.yoho.search.consumer.service.base.GoodsService;
import com.yoho.search.consumer.service.base.ProductService;
import com.yoho.search.consumer.service.bo.StorageSkuBO;
import com.yoho.search.consumer.service.logic.StorageSkuLogicService;
import com.yoho.search.core.es.utils.IgnoreSomeException;
import com.yoho.search.dal.model.Goods;
import com.yoho.search.dal.model.Product;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.*;
@Component
public class GoodsMqListener extends AbstractStorageRelatedMqListener implements ChannelAwareMessageListener {
private static final Logger logger = LoggerFactory.getLogger(GoodsMqListener.class);
@Autowired
private GoodsService goodsService;
@Autowired
private StorageSkuIndexBulkService storageSkuIndexService;
@Autowired
private StorageSkuLogicService storageSkuLogicService;
public void onMessage(Message message, Channel channel) throws Exception {
try {
final String key = UUID.randomUUID().toString();
String messageStr = new String(message.getBody(), "UTF-8");
logger.info("[model=GoodsMqListener][key={}][message={}]", key, messageStr);
// 如果在重建索引等待
this.waitingRebuildingIndex();
JSONObject json = JSONObject.parseObject(messageStr);
if (ISearchConstants.ACTION_DELETE.equals(json.getString("action"))) {
deleteData(json.getString("data"), key);
} else if (ISearchConstants.ACTION_UPDATE.equals(json.getString("action"))) {
updateData(json.getObject("data", Map.class), key);
} else {
updateData(json.getObject("data", Map.class), key);
}
} catch (Exception e) {
publisher.publishEvent(new SearchEvent(EventReportEnum.GOODSMQLISTENER_ONMESSAGE.getEventName(), EventReportEnum.GOODSMQLISTENER_ONMESSAGE.getFunctionName(),
EventReportEnum.GOODSMQLISTENER_ONMESSAGE.getMoudleName(), "exception", IgnoreSomeException.filterSomeException(e), null));
Thread.sleep(1000);
throw e;
}
}
public void deleteData(final String id, final String key) {
if (StringUtils.isBlank(id)) {
return;
}
long begin = System.currentTimeMillis();
Goods goods = goodsService.getById(Integer.parseInt(id));
if (goods == null || goods.getId() == null) {
return;
}
// 1、获取StorageSku
List<Integer> goodsIds = new ArrayList<>();
goodsIds.add(goods.getId());
List<StorageSkuBO> storageSkuBOList = storageSkuLogicService.getStorageSkuIndexBySkcs(goodsIds);
// 2、删除数据库数据
goodsService.delete(Integer.valueOf(id));
// 3、更新productIndex索引
if (goods != null && goods.getProductId() != null) {
updateIndexNew(goods.getProductId(), begin, key);
}
// 4、更新storageSku索引
for (StorageSkuBO storageSkuBO : storageSkuBOList) {
storageSkuBO.setGoodsStatus(0);
}
storageSkuIndexService.updateStorageSkuIndexByList(storageSkuBOList);
logger.info("[func=deleteData][step=success][key={}][indexName={}][id={}][cost={}ms]", key, ISearchConstants.INDEX_NAME_PRODUCT_INDEX, id,
(System.currentTimeMillis() - begin));
}
@SuppressWarnings({ "rawtypes", "unchecked" })
public void updateData(final Map data, final String key) {
CostStatistics costStatistics = new CostStatistics();
// 1.对象转义
Goods goods = new Goods();
goods = (Goods) ConvertUtils.toJavaBean(goods, data);
if (goods == null || goods.getId() == null) {
return;
}
Integer productId = goods.getProductId();
if (productId == null) {
return;
}
logger.info("[func=updateData][step1=covertFromMap][goodId={}][cost={}ms]", goods.getId(), costStatistics.getCost());
// 2.保存到数据库
goodsService.saveOrUpdate(goods);
logger.info("[func=updateData][step2=saveToBb][goodId={}][cost={}ms]", goods.getId(), costStatistics.getCost());
// 3.更新到ES
updateIndexNew(productId, System.currentTimeMillis(), key);
logger.info("[func=updateData][step3=updateProductIndex][goodId={}][cost={}ms]", goods.getId(), costStatistics.getCost());
// 4.更新storagesku
storageSkuIndexService.updateStorageSkuIndexByGoodsId(goods.getId(), System.currentTimeMillis(), key);
logger.info("[func=updateData][step4=updateStorageSkuIndex][goodId={}][cost={}ms]", goods.getId(), costStatistics.getCost());
logger.info("[func=updateData][step5=success][goodId={}][totalCost={}ms][costStatistics={}]", goods.getId(), costStatistics.getTotalCost(),
costStatistics.getCostStatisticsString());
}
private void updateIndexNew(Integer productId, long begin, final String key) {
CostStatistics costStatistics = new CostStatistics();
Map<String, Object> indexData = new HashMap<String, Object>();
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=getGoodsListByProductId][productId={}][cost={}]", productId, costStatistics.getCost());
//第一步：获取goodsList，组装goods数据
fillColor(indexData,productId);
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=genGoodsList][productId={}][cost={}]", productId, costStatistics.getCost());
// 第二步、组装库存数据
this.fillStorageNumAndSizeInfo(productId, indexData);
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=fillProductStorage][productId={}][cost={}]", productId, costStatistics.getCost());
// 第三步：更新索引
this.updateProductIndexWithDataMap(indexData, productId, key, begin);
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=updateProductIndexWithDataMap][productId={}][cost={}]", productId, costStatistics.getCost());
}
private static final Logger logger = LoggerFactory.getLogger(GoodsMqListener.class);
@Autowired
private GoodsService goodsService;
@Autowired
private StorageSkuIndexBulkService storageSkuIndexService;
@Autowired
private StorageSkuLogicService storageSkuLogicService;
@Autowired
private ProductService productService;
public void onMessage(Message message, Channel channel) throws Exception {
try {
final String key = UUID.randomUUID().toString();
String messageStr = new String(message.getBody(), "UTF-8");
logger.info("[model=GoodsMqListener][key={}][message={}]", key, messageStr);
// 如果在重建索引等待
this.waitingRebuildingIndex();
JSONObject json = JSONObject.parseObject(messageStr);
if (ISearchConstants.ACTION_DELETE.equals(json.getString("action"))) {
deleteData(json.getString("data"), key);
} else if (ISearchConstants.ACTION_UPDATE.equals(json.getString("action"))) {
updateData(json.getObject("data", Map.class), key);
} else {
updateData(json.getObject("data", Map.class), key);
}
} catch (Exception e) {
publisher.publishEvent(new SearchEvent(EventReportEnum.GOODSMQLISTENER_ONMESSAGE.getEventName(), EventReportEnum.GOODSMQLISTENER_ONMESSAGE.getFunctionName(),
EventReportEnum.GOODSMQLISTENER_ONMESSAGE.getMoudleName(), "exception", IgnoreSomeException.filterSomeException(e), null));
Thread.sleep(1000);
throw e;
}
}
public void deleteData(final String id, final String key) {
if (StringUtils.isBlank(id)) {
return;
}
long begin = System.currentTimeMillis();
Goods goods = goodsService.getById(Integer.parseInt(id));
if (goods == null || goods.getId() == null) {
return;
}
// 1、获取StorageSku
List<Integer> goodsIds = new ArrayList<>();
goodsIds.add(goods.getId());
List<StorageSkuBO> storageSkuBOList = storageSkuLogicService.getStorageSkuIndexBySkcs(goodsIds);
//判断skn状态
Map<String, Object> indexData = new HashMap<String, Object>();
Integer productId = goods.getProductId();
this.fillSknStatus(productId, indexData);
this.updateProductIndexWithDataMap(indexData, productId, key, begin);
// 2、删除数据库数据
goodsService.delete(Integer.valueOf(id));
// 3、更新productIndex索引
if (goods != null && goods.getProductId() != null) {
updateIndexNew(goods.getProductId(), begin, key);
}
// 4、更新storageSku索引
for (StorageSkuBO storageSkuBO : storageSkuBOList) {
storageSkuBO.setGoodsStatus(0);
}
storageSkuIndexService.updateStorageSkuIndexByList(storageSkuBOList);
logger.info("[func=deleteData][step=success][key={}][indexName={}][id={}][cost={}ms]", key, ISearchConstants.INDEX_NAME_PRODUCT_INDEX, id,
(System.currentTimeMillis() - begin));
}
@SuppressWarnings({"rawtypes", "unchecked"})
public void updateData(final Map data, final String key) {
CostStatistics costStatistics = new CostStatistics();
// 1.对象转义
Goods goods = new Goods();
goods = (Goods) ConvertUtils.toJavaBean(goods, data);
if (goods == null || goods.getId() == null) {
return;
}
Integer productId = goods.getProductId();
if (productId == null) {
return;
}
logger.info("[func=updateData][step1=covertFromMap][goodId={}][cost={}ms]", goods.getId(), costStatistics.getCost());
// 2.保存到数据库
goodsService.saveOrUpdate(goods);
logger.info("[func=updateData][step2=saveToBb][goodId={}][cost={}ms]", goods.getId(), costStatistics.getCost());
// 3.更新到ES
updateIndexNew(productId, System.currentTimeMillis(), key);
logger.info("[func=updateData][step3=updateProductIndex][goodId={}][cost={}ms]", goods.getId(), costStatistics.getCost());
// 4.更新storagesku
storageSkuIndexService.updateStorageSkuIndexByGoodsId(goods.getId(), System.currentTimeMillis(), key);
logger.info("[func=updateData][step4=updateStorageSkuIndex][goodId={}][cost={}ms]", goods.getId(), costStatistics.getCost());
logger.info("[func=updateData][step5=success][goodId={}][totalCost={}ms][costStatistics={}]", goods.getId(), costStatistics.getTotalCost(),
costStatistics.getCostStatisticsString());
}
private void updateIndexNew(Integer productId, long begin, final String key) {
CostStatistics costStatistics = new CostStatistics();
Map<String, Object> indexData = new HashMap<String, Object>();
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=getGoodsListByProductId][productId={}][cost={}]", productId, costStatistics.getCost());
//第一步：获取goodsList，组装goods数据
fillColor(indexData, productId);
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=genGoodsList][productId={}][cost={}]", productId, costStatistics.getCost());
// 第二步、组装库存数据
this.fillStorageNumAndSizeInfo(productId, indexData);
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=fillProductStorage][productId={}][cost={}]", productId, costStatistics.getCost());
// 第三步，判断skn状态
this.fillSknStatus(productId, indexData);
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=fillSknStatus][productId={}][cost={}]", productId, costStatistics.getCost());
// 第四步：更新索引
this.updateProductIndexWithDataMap(indexData, productId, key, begin);
logger.info("[model=GoodsMqListener_UpdateIndexNew][step=updateProductIndexWithDataMap][productId={}][cost={}]", productId, costStatistics.getCost());
}
/**
* 如果skn下面所有skc的状态都是0，那skn的状态设为0,更新和删除都需要验证
*/
private void fillSknStatus(Integer productId, Map<String, Object> indexData) {
List<Integer> ids = new ArrayList<>();
ids.add(productId);
List<Goods> goodsList = goodsService.getListByProductId(ids);
if (CollectionUtils.isEmpty(goodsList)) {
return;
}
boolean sknInvalid = true;
for (Goods goods : goodsList) {
if (goods.getStatus() == 1) {
sknInvalid = false;
}
}
if (sknInvalid) {
indexData.put("status", 0);
} else {
//需要检查product表的status字段，如果为1才更新状态，为0则以product表的状态为准
Product product = productService.getById(productId);
if (product != null && product.getStatus() == 1) {
indexData.put("status", 1);
}
}
}
}
... ...
... ... @@ -8,7 +8,6 @@ import com.yoho.search.consumer.service.base.StorageService;
import com.yoho.search.consumer.service.bo.ProductGoodsBO;
import com.yoho.search.consumer.service.logic.productIndex.ProductGoodsLogicService;
import com.yoho.search.dal.model.Storage;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
... ... @@ -32,10 +31,11 @@ public abstract class AbstractStorageRelatedMqListener extends AbstractMqListene
@Autowired
private SizeService sizeService;
@Autowired
private GoodsService goodsService;
@Autowired
private GoodsService goodsService;
@Autowired
private ProductGoodsLogicService productGoodsLogicService;
protected void fillStorageNumAndSizeInfo(Integer productId, Map<String, Object> indexData) {
... ... @@ -77,25 +77,25 @@ public abstract class AbstractStorageRelatedMqListener extends AbstractMqListene
}
}
}
protected void fillColor(Map<String, Object> indexData,Integer productId){
Set<String> colorIdSet = new HashSet<String>();
Set<String> colorNameSet = new HashSet<String>();
// 第一步：获取goodsList
List<ProductGoodsBO> productGoodsBOList = goodsService.getProductGoodsByProductId(productId);
// 第二步、组装goods数据
if (productGoodsBOList != null && productGoodsBOList.size() == 1) {
ProductGoodsBO productGoodsBO = productGoodsBOList.get(0);
String goodsListJsonArrayStr = productGoodsBO.getGoodsList();
JSONArray goodsListJsonArray = JSONArray.parseArray(goodsListJsonArrayStr);
productGoodsLogicService.getColorSet(goodsListJsonArray, colorIdSet, colorNameSet);//获取有库存且状态正常的color
indexData.put("goodsList", goodsListJsonArray);
indexData.put("colorIds", StringUtils.join(colorIdSet, ","));
indexData.put("colorNames", StringUtils.join(colorNameSet, ","));
} else {
indexData.put("goodsList", "");
indexData.put("colorIds", "");
indexData.put("colorNames", "");
}
}
}
protected void fillColor(Map<String, Object> indexData, Integer productId) {
Set<String> colorIdSet = new HashSet<String>();
Set<String> colorNameSet = new HashSet<String>();
// 第一步：获取goodsList
List<ProductGoodsBO> productGoodsBOList = goodsService.getProductGoodsByProductId(productId);
// 第二步、组装goods数据
if (productGoodsBOList != null && productGoodsBOList.size() == 1) {
ProductGoodsBO productGoodsBO = productGoodsBOList.get(0);
String goodsListJsonArrayStr = productGoodsBO.getGoodsList();
JSONArray goodsListJsonArray = JSONArray.parseArray(goodsListJsonArrayStr);
productGoodsLogicService.getColorSet(goodsListJsonArray, colorIdSet, colorNameSet);//获取有库存且状态正常的color
indexData.put("goodsList", goodsListJsonArray);
indexData.put("colorIds", StringUtils.join(colorIdSet, ","));
indexData.put("colorNames", StringUtils.join(colorNameSet, ","));
} else {
indexData.put("goodsList", "");
indexData.put("colorIds", "");
indexData.put("colorNames", "");
}
}
}
\ No newline at end of file
... ...
package com.yoho.search.consumer.service.base;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONObject;
import com.yoho.search.consumer.service.bo.ProductGoodsBO;
import com.yoho.search.consumer.service.logic.productIndex.ProductGoodsLogicService;
import com.yoho.search.dal.GoodsMapper;
import com.yoho.search.dal.model.Goods;
import com.yoho.search.consumer.service.bo.ProductGoodsBO;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@Component
public class GoodsService {
... ... @@ -53,6 +52,8 @@ public class GoodsService {
return productGoodsLogicService.getData(Arrays.asList(productId));
}
public List<Goods> getListByProductId(List<Integer> productIds){return goodsMapper.selectListByProductIds(productIds); }
@Deprecated
public List<Map<String, Object>> getGoodsListByProductId(Integer productId) {
return goodsMapper.getGoodListByProductId(productId);
... ...
... ... @@ -4,7 +4,6 @@ import com.yoho.search.consumer.service.bo.ProductIBO;
import com.yoho.search.consumer.service.bo.ProductIndexBO;
import com.yoho.search.consumer.service.logic.productIndex.ProductILogicService;
import com.yoho.search.consumer.service.logic.productIndex.viewBuilder.GeneralDataBuilder;
import com.yoho.search.consumer.service.logic.productIndex.viewBuilder.ProductColorsBuilder;
import com.yoho.search.consumer.service.logic.productIndex.viewBuilder.ViewBuilder;
import com.yoho.search.dal.BrandMapper;
import com.yoho.search.dal.ProductMapper;
... ... @@ -134,9 +133,6 @@ public class ProductIndexLogicService implements ApplicationContextAware {
List<Integer> skns = productIBOs.stream().map(ProductIBO::getProductSkn).collect(Collectors.toList());
generalDataBuilder.buildDataFromProductI(productIndexBOs, productIBOs);
viewBuilderList.stream().forEach(viewBuilder -> {
if(viewBuilder instanceof ProductColorsBuilder){
return;
}
viewBuilder.build(productIndexBOs, ids, skns);
});
return productIndexBOs;
... ...
package com.yoho.search.consumer.service.logic.productIndex.viewBuilder;
import com.yoho.search.consumer.service.bo.ProductIndexBO;
import com.yoho.search.dal.ProductColorsMapper;
import com.yoho.search.dal.model.ProductColors;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/**
* Created by wangnan on 2017/1/6.
*/
@Component
public class ProductColorsBuilder implements ViewBuilder {
@Autowired
private ProductColorsMapper productColorsMapper;
/**
* 构建ProductColors
*/
@Override
public void build(List<ProductIndexBO> productIndexBOs, List<Integer> ids, List<Integer> skns) {
List<ProductColors> productColorses = productColorsMapper.getProductColors(ids);
Map<Integer, ProductColors> productColorsMap = productColorses.stream().parallel().collect(Collectors.toMap(ProductColors::getProductId, (p) -> p));
for (ProductIndexBO productIndexBO : productIndexBOs) {
ProductColors productColors = productColorsMap.get(productIndexBO.getProductId());
if (productColors != null) {
productIndexBO.setColorIds(productColors.getColorIds());
productIndexBO.setColorNames(productColors.getColorNames());
}
}
}
}
package com.yoho.search.consumer.service.logic.productIndex.viewBuilder;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yoho.search.consumer.service.base.ProductService;
import com.yoho.search.consumer.service.bo.ProductGoodsBO;
import com.yoho.search.consumer.service.bo.ProductIndexBO;
import com.yoho.search.consumer.service.logic.productIndex.ProductGoodsLogicService;
... ... @@ -21,9 +23,11 @@ import java.util.stream.Collectors;
public class ProductGoodsBuilder implements ViewBuilder {
@Autowired
private ProductGoodsLogicService productGoodsLogicService;
@Autowired
private ProductService productService;
/**
* 构建ProductGoods
* 构建ProductIndex的ColorIds,ColorNames属性
*/
@Override
public void build(List<ProductIndexBO> productIndexBOs, List<Integer> ids, List<Integer> skns) {
... ... @@ -34,6 +38,7 @@ public class ProductGoodsBuilder implements ViewBuilder {
for (ProductIndexBO productIndexBO : productIndexBOs) {
ProductGoodsBO productGoodsBO = productGoodsesMap.get(productIndexBO.getProductId());
if (productGoodsBO != null) {
this.checkSknStatus(productIndexBO, productGoodsBO);
productIndexBO.setGoodsList(productGoodsBO.getGoodsList());
String goodsListJsonArrayStr = productGoodsBO.getGoodsList();
if (goodsListJsonArrayStr == null || goodsListJsonArrayStr.length() < 3) {
... ... @@ -51,4 +56,27 @@ public class ProductGoodsBuilder implements ViewBuilder {
}
}
}
/**
* 如果skn下面所有skc的状态都是0，那skn的状态设为0
* 和增量不同之处：不存在status从0->1，只存在1->0
*/
private void checkSknStatus(ProductIndexBO productIndexBO, ProductGoodsBO productGoodsBO) {
JSONArray jsonArray = JSONArray.parseArray(productGoodsBO.getGoodsList());
if (jsonArray != null) {
boolean sknInvalid = true;
for (int i = 0; i < jsonArray.size(); i++) {
JSONObject jsonObject = (JSONObject) jsonArray.get(i);
if (jsonObject!=null&&jsonObject.get("status") != null) {
String status = jsonObject.get("status").toString();
if (status.equals("1")) {
sknInvalid = false;
}
}
}
if (sknInvalid) {
productIndexBO.setStatus(0);
}
}
}
}
... ...