|
|
|
package com.yoho.search.consumer.index.increment.flow;
|
|
|
|
|
|
|
|
import com.alibaba.fastjson.JSONObject;
|
|
|
|
import com.yoho.search.base.utils.ISearchConstants;
|
|
|
|
import com.yoho.search.consumer.index.common.IYohoIndexService;
|
|
|
|
import com.yoho.search.consumer.index.rebuild.RebuildFlagService;
|
|
|
|
import com.yoho.search.consumer.service.base.ProductIndexService;
|
|
|
|
import com.yoho.search.consumer.service.base.ProductService;
|
|
|
|
import com.yoho.search.consumer.service.base.ProductVectorFeatureService;
|
|
|
|
import com.yoho.search.consumer.service.bo.ProductIndexBO;
|
|
|
|
import com.yoho.search.consumer.service.logic.ProductVectorFeatureLogicService;
|
|
|
|
import com.yoho.search.consumer.suggests.common.RetryBusinessFlow;
|
|
|
|
import com.yoho.search.core.es.IElasticsearchClient;
|
|
|
|
import com.yoho.search.core.es.model.ESBluk;
|
|
|
|
import com.yoho.search.dal.model.Product;
|
|
|
|
import org.apache.commons.collections.CollectionUtils;
|
|
|
|
import org.apache.commons.lang.StringUtils;
|
|
|
|
import org.elasticsearch.action.bulk.BulkResponse;
|
|
|
|
import org.slf4j.Logger;
|
|
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
|
import org.springframework.stereotype.Component;
|
|
|
|
|
|
|
|
import java.util.ArrayList;
|
|
|
|
import java.util.List;
|
|
|
|
import java.util.Map;
|
|
|
|
import java.util.stream.Collectors;
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Created by ginozhang on 2017/3/17.
|
|
|
|
*/
|
|
|
|
@Component
|
|
|
|
public class ProductIndexFeatureVectorUpdateFlow implements RetryBusinessFlow {
|
|
|
|
|
|
|
|
private static final Logger logger = LoggerFactory.getLogger("FLOW_EXECUTOR");
|
|
|
|
|
|
|
|
private static final int VALID_COUNT_LIMIT = 1000;
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private ProductVectorFeatureService productVectorFeatureService;
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private ProductVectorFeatureLogicService productVectorFeatureLogicService;
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private ProductService productService;
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private ProductIndexService productIndexService;
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private IYohoIndexService yohoIndexService;
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private RebuildFlagService rebuildFlagService;
|
|
|
|
|
|
|
|
private volatile String versionDate = null;
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public String flowName() {
|
|
|
|
return this.getClass().getSimpleName();
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void init() {
|
|
|
|
rebuildFlagService.waitingRebuildingIndex();
|
|
|
|
rebuildFlagService.updateIsBuildingTrue();
|
|
|
|
productVectorFeatureLogicService.updateGenerateDate();
|
|
|
|
this.versionDate = productVectorFeatureLogicService.getGenerateDate();
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public int getTotalCount() {
|
|
|
|
if (StringUtils.isEmpty(this.versionDate)) {
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
|
|
|
// 如果查询的skn数据小于1000个 则可能不是完整的数据 不更新索引
|
|
|
|
int total = productVectorFeatureService.selectCount(this.versionDate);
|
|
|
|
if (total <= VALID_COUNT_LIMIT) {
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
|
|
|
|
return productService.count();
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public boolean doBusiness(int pageNo, int batchSize) {
|
|
|
|
int start = (pageNo - 1) * batchSize;
|
|
|
|
List<Product> productList = productService.getPageLists(start, batchSize);
|
|
|
|
if (CollectionUtils.isEmpty(productList)) {
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
|
|
|
|
List<Integer> sknList = productList.stream().map(Product::getErpProductId).collect(Collectors.toList());
|
|
|
|
Map<Integer, Integer> skn2IDMap = productList.stream().collect(Collectors.toMap(Product::getErpProductId, Product::getId));
|
|
|
|
Map<Integer, String> productVectorFeatureMapBaseSkn = productVectorFeatureLogicService.queryProductVectorFeatureMap(sknList);
|
|
|
|
logger.info("[{} business][pageNo={}][productVectorFeatureMapBaseSknSize={}]", flowName(), pageNo, productVectorFeatureMapBaseSkn.size());
|
|
|
|
|
|
|
|
List<ProductIndexBO> dataList = new ArrayList<>();
|
|
|
|
productVectorFeatureMapBaseSkn.forEach((skn, vector) -> {
|
|
|
|
if (skn2IDMap.get(skn) != null) {
|
|
|
|
ProductIndexBO temp = new ProductIndexBO();
|
|
|
|
temp.setId(skn2IDMap.get(skn));
|
|
|
|
temp.setProductFeatureFactor(vector);
|
|
|
|
dataList.add(temp);
|
|
|
|
}
|
|
|
|
});
|
|
|
|
|
|
|
|
if (CollectionUtils.isEmpty(dataList)) {
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
|
|
|
|
final String indexName = ISearchConstants.INDEX_NAME_PRODUCT_INDEX;
|
|
|
|
List<ESBluk> bluks = new ArrayList<>(dataList.size());
|
|
|
|
for (ProductIndexBO ob : dataList) {
|
|
|
|
bluks.add(new ESBluk(JSONObject.toJSONString(productIndexService.beanToMap(ob)), ob.getId().toString(), indexName, indexName, false));
|
|
|
|
}
|
|
|
|
|
|
|
|
IElasticsearchClient client = yohoIndexService.getIndex(indexName).getIndexClient();
|
|
|
|
BulkResponse bulkResponse = client.bulk(bluks);
|
|
|
|
if (bulkResponse.hasFailures()) {
|
|
|
|
throw new RuntimeException(String.format("bulk has failure,[yohoIndexName=[%s]],[pageNo=%s],[failureMessage=%s]", indexName, pageNo,
|
|
|
|
bulkResponse.buildFailureMessage()));
|
|
|
|
}
|
|
|
|
|
|
|
|
logger.info("[{} business][pageNo={}][message=bulk succeed]", flowName(), pageNo);
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void finish(boolean doBusinessResult, Exception exception) {
|
|
|
|
logger.info("[{} business][doBusinessResult={}][exception={}]", flowName(), doBusinessResult, exception);
|
|
|
|
rebuildFlagService.updateIsBuildingFalse();
|
|
|
|
if (exception == null) {
|
|
|
|
productVectorFeatureLogicService.publishGenerateDate();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
} |
...
|
...
|
|
