商品索引添加商品热度字段，支持全量和定时增量。fix

@@ -20,5 +20,9 @@ public interface ProductHeatValueMapper {
 
     Integer selectLatestDateId();
 
-    List<ProductHeatValue> selectHeatValueOfLatestDate(@Param(value="sknList")List<Integer> sknList, @Param(value="dateId")Integer dateId);
+    Integer count(@Param(value="dateId")Integer dateId);
+
+    List<ProductHeatValue> selectHeatValueOfLatestDateBySkn(@Param(value="sknList")List<Integer> sknList, @Param(value="dateId")Integer dateId);
+
+    List<ProductHeatValue> selectHeatValueOfLatestDate(@Param(value="dateId")Integer dateId,@Param(value = "offset") Integer offset, @Param(value = "pageSize") Integer pageSize);
 }
\ No newline at end of file

@@ -79,7 +79,7 @@
       last_update_time = #{lastUpdateTime,jdbcType=INTEGER}
     where product_skn = #{productSkn,jdbcType=INTEGER}
   </update>
-    <select id="selectHeatValueOfLatestDate" resultMap="BaseResultMap" timeout="20000">
+    <select id="selectHeatValueOfLatestDateBySkn" resultMap="BaseResultMap" timeout="20000">
         <![CDATA[
 		select * from product_heat_value p
 		where
@@ -92,7 +92,18 @@
             #{item}
         </foreach>
     </select>
+    <select id="selectHeatValueOfLatestDate" resultMap="BaseResultMap" timeout="20000">
+		select * from product_heat_value p
+		where
+		date_id = #{dateId}
+		limit #{offset},#{pageSize}
+    </select>
     <select id="selectLatestDateId" resultType="java.lang.Integer">
         select max(date_id) from product_heat_value
     </select>
+    <select id="count" resultType="java.lang.Integer" timeout="20000">
+        SELECT count(*) FROM product_heat_value
+        where
+		date_id = #{dateId}
+    </select>
 </mapper>
\ No newline at end of file

+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.ProductService;
+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.ProductHeatValueMapper;
+import com.yoho.search.dal.model.Product;
+import com.yoho.search.dal.model.ProductHeatValue;
+import org.apache.commons.collections.CollectionUtils;
+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 wangnan on 2017/4/10.
+ */
+@Component
+public class ProductIndexHeatValueUpdateFlow implements RetryBusinessFlow {
+
+    private static final Logger logger = LoggerFactory.getLogger("FLOW_EXECUTOR");
+
+    @Autowired
+    private RebuildFlagService rebuildFlagService;
+    @Autowired
+    private ProductHeatValueMapper productHeatValueMapper;
+    @Autowired
+    private ProductService productService;
+    @Autowired
+    private IYohoIndexService yohoIndexService;
+
+    @Override
+    public String flowName() {
+        return this.getClass().getSimpleName();
+    }
+
+    @Override
+    public void init() {
+        rebuildFlagService.waitingRebuildingIndex();
+        rebuildFlagService.updateIsBuildingTrue();
+    }
+
+    @Override
+    public int getTotalCount() {
+        return productHeatValueMapper.count(productHeatValueMapper.selectLatestDateId());
+    }
+
+    @Override
+    public boolean doBusiness(int pageNo, int batchSize) {
+        int start = (pageNo - 1) * batchSize;
+        List<ProductHeatValue> productHeatValueList = productHeatValueMapper.selectHeatValueOfLatestDate(productHeatValueMapper.selectLatestDateId(),start, batchSize);
+        if (CollectionUtils.isEmpty(productHeatValueList)) {
+            return true;
+        }
+        List<Integer> sknList = productHeatValueList.stream().map(ProductHeatValue::getProductSkn).collect(Collectors.toList());
+        List<Product> productList = productService.selectListSkns(sknList);
+        Map<Integer, Integer> skn2IDMap = productList.stream().collect(Collectors.toMap(Product::getErpProductId, Product::getId));
+        logger.info("[{} business][pageNo={}][productHeatValueList={}]", flowName(), pageNo, productHeatValueList.size());
+        List<JSONObject> dataList = new ArrayList<JSONObject>();
+        productHeatValueList.stream().forEach((p)->{
+            if (skn2IDMap.get(p.getProductSkn()) != null) {
+                JSONObject product = new JSONObject();
+                product.put("id", skn2IDMap.get(p.getProductSkn()));
+                product.put("heatValue", p.getHeatValue());
+                dataList.add(product);
+            }
+        });
+        if (CollectionUtils.isEmpty(dataList)) {
+            return true;
+        }
+        final String indexName = ISearchConstants.INDEX_NAME_PRODUCT_INDEX;
+        List<ESBluk> bluks = new ArrayList<>(dataList.size());
+        for (JSONObject product : dataList) {
+            bluks.add(new ESBluk(product.toJSONString(), product.getInteger("id").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();
+    }
+}

+package com.yoho.search.consumer.job;
+
+import com.yoho.search.base.utils.ISearchConstants;
+import com.yoho.search.consumer.index.increment.flow.ProductIndexHeatValueUpdateFlow;
+import com.yoho.search.consumer.suggests.common.RetryBusinessFlowExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * Created by wangnan on 2017/4/10.
+ */
+@Component
+public class ProductHeatValueJob {
+    private static final Logger logger = LoggerFactory.getLogger(ProductHeatValueJob.class);
+
+    private ProductIndexHeatValueUpdateFlow productIndexHeatValueUpdateFlow;
+
+    /**
+     * 每隔半个小时
+     */
+    @Scheduled(cron = "0 0/30 * * * ?")
+    public void doUpdateProductIndexHeatValue(){
+        long begin = System.currentTimeMillis();
+        RetryBusinessFlowExecutor flowExecutor = new RetryBusinessFlowExecutor(productIndexHeatValueUpdateFlow, ISearchConstants.SEARCH_INDEX_BATCH_MAX_THREAD_SIZE, ISearchConstants.SEARCH_INDEX_BATCH_LIMIT);
+        boolean result = flowExecutor.execute();
+        logger.info("ProductHeatValueJob.doUpdateProductIndexHeatValue end----[result={}][cost={}]", result, System.currentTimeMillis() - begin);
+    }
+
+    /**
+     * 每天1时
+     */
+    @Scheduled(cron = "0 0 1 * * ?")
+    public void clean(){
+
+    }
+
+}

 package com.yoho.search.consumer.restapi;
 
+import com.yoho.search.base.utils.ISearchConstants;
+import com.yoho.search.consumer.index.increment.flow.ProductIndexHeatValueUpdateFlow;
+import com.yoho.search.consumer.job.ProductHeatValueJob;
 import com.yoho.search.consumer.service.logic.tbl.TblBrandRelationLogicService;
 import com.yoho.search.consumer.service.logic.tbl.TblSortRelationLogicService;
+import com.yoho.search.consumer.suggests.common.RetryBusinessFlowExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -15,11 +21,37 @@ import java.util.Map;
  */
 @Controller
 public class TblRelationController {
+
+    private static final Logger logger = LoggerFactory.getLogger(ProductHeatValueJob.class);
+
+
     @Autowired
     private TblSortRelationLogicService tblSortRelationLogicService;
     @Autowired
     private TblBrandRelationLogicService tblBrandRelationLogicService;
 
+    private ProductIndexHeatValueUpdateFlow productIndexHeatValueUpdateFlow;
+
+    @RequestMapping(value = "/doUpdateProductIndexHeatValue")
+    @ResponseBody
+    public Map<String, Object> doUpdateProductIndexHeatValue() {
+        Map<String, Object> rt = new HashMap<String, Object>();
+        try {
+            long begin = System.currentTimeMillis();
+            RetryBusinessFlowExecutor flowExecutor = new RetryBusinessFlowExecutor(productIndexHeatValueUpdateFlow, ISearchConstants.SEARCH_INDEX_BATCH_MAX_THREAD_SIZE, ISearchConstants.SEARCH_INDEX_BATCH_LIMIT);
+            boolean result = flowExecutor.execute();
+            logger.info("ProductHeatValueJob.doUpdateProductIndexHeatValue end----[result={}][cost={}]", result, System.currentTimeMillis() - begin);
+            rt.put("code", 200);
+            rt.put("message", "success");
+        } catch (Exception e) {
+            e.printStackTrace();
+            rt.put("code", 400);
+            rt.put("message", e.getMessage());
+        }
+        return rt;
+    }
+
+
     @RequestMapping(value = "/tblSortRelation")
     @ResponseBody
     public Map<String, Object> loadSortRelationData() {

@@ -55,4 +55,6 @@ public class ProductService {
     public Product getBySkn(Integer skn){return productMapper.selectBySkn(skn); }
 
     public List<Product> selectListByIds(List<Integer> ProductIds){return productMapper.selectListByIds(ProductIds);}
+
+    public List<Product> selectListSkns(List<Integer> ProductSkns){return productMapper.selectBySkns(ProductSkns);}
 }

@@ -22,7 +22,7 @@ public class HeatValueBuilder implements ViewBuilder {
 
     @Override
     public void build(List<ProductIndexBO> productIndexBOs, List<Integer> ids, List<Integer> sknList) {
-        List<ProductHeatValue> productHeatValueList = productHeatValueMapper.selectHeatValueOfLatestDate(sknList, productHeatValueMapper.selectLatestDateId());
+        List<ProductHeatValue> productHeatValueList = productHeatValueMapper.selectHeatValueOfLatestDateBySkn(sknList, productHeatValueMapper.selectLatestDateId());
         Map<Integer, BigDecimal> productHeatValueMap = productHeatValueList.parallelStream().collect(Collectors.toMap(ProductHeatValue::getProductSkn, ProductHeatValue::getHeatValue));
         productIndexBOs.stream().forEach((p) -> {p.setHeatValue(productHeatValueMap.get(p.getProductSkn()));});
     }