Merge branch 'master' into zf_suggest

+package com.yoho.search.dal;
+
+import java.util.List;
+
+import com.yoho.search.dal.model.PromotionProduct;
+
+public interface PromotionProductMapper {
+    int deleteByPrimaryKey(Integer id);
+
+    int insert(PromotionProduct record);
+
+    int insertSelective(PromotionProduct record);
+
+    PromotionProduct selectByPrimaryKey(Integer id);
+
+    int updateByPrimaryKeySelective(PromotionProduct record);
+
+    int updateByPrimaryKey(PromotionProduct record);
+
+	List<PromotionProduct> selectListBySkns(List<Integer> skns);
+	
+	List<PromotionProduct> selectListBySkn(Integer skn);
+}
\ No newline at end of file

@@ -270,7 +270,7 @@
 		from product_price limit #{offset},#{pageSize}
 	</select>
 	<select id="selectByIds" resultMap="BaseResultMap">
-		Select
+		select
 		<include refid="Base_Column_List" />
 		from product_price where product_id in
 		<foreach item="item" index="index" collection="list"

+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="com.yoho.search.dal.PromotionProductMapper" >
+  <resultMap id="BaseResultMap" type="com.yoho.search.dal.model.PromotionProduct" >
+    <id column="id" property="id" jdbcType="INTEGER" />
+    <result column="promotion_id" property="promotionId" jdbcType="INTEGER" />
+    <result column="product_skn" property="productSkn" jdbcType="INTEGER" />
+  </resultMap>
+  <sql id="Base_Column_List" >
+    id, promotion_id, product_skn
+  </sql>
+  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.Integer" >
+    select 
+    <include refid="Base_Column_List" />
+    from promotion_product
+    where id = #{id,jdbcType=INTEGER}
+  </select>
+  <select id="selectListBySkns" resultMap="BaseResultMap">
+    select 
+    <include refid="Base_Column_List" />
+    from promotion_product
+    where product_skn in
+	<foreach item="item" index="index" collection="list"
+			 open="(" separator="," close=")">
+		#{item}
+	</foreach>
+  </select>
+  <select id="selectListBySkn" resultMap="BaseResultMap" parameterType="java.lang.Integer">
+    select 
+    <include refid="Base_Column_List" />
+    from promotion_product
+    where product_skn = #{skn,jdbcType=INTEGER}
+  </select>
+  <delete id="deleteByPrimaryKey" parameterType="java.lang.Integer" >
+    delete from promotion_product
+    where id = #{id,jdbcType=INTEGER}
+  </delete>
+  <insert id="insert" parameterType="com.yoho.search.dal.model.PromotionProduct" >
+    insert into promotion_product (id, promotion_id, product_skn
+      )
+    values (#{id,jdbcType=INTEGER}, #{promotionId,jdbcType=INTEGER}, #{productSkn,jdbcType=INTEGER}
+      )
+  </insert>
+  <insert id="insertSelective" parameterType="com.yoho.search.dal.model.PromotionProduct" >
+    insert into promotion_product
+    <trim prefix="(" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        id,
+      </if>
+      <if test="promotionId != null" >
+        promotion_id,
+      </if>
+      <if test="productSkn != null" >
+        product_skn,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides="," >
+      <if test="id != null" >
+        #{id,jdbcType=INTEGER},
+      </if>
+      <if test="promotionId != null" >
+        #{promotionId,jdbcType=INTEGER},
+      </if>
+      <if test="productSkn != null" >
+        #{productSkn,jdbcType=INTEGER},
+      </if>
+    </trim>
+  </insert>
+  <update id="updateByPrimaryKeySelective" parameterType="com.yoho.search.dal.model.PromotionProduct" >
+    update promotion_product
+    <set >
+      <if test="promotionId != null" >
+        promotion_id = #{promotionId,jdbcType=INTEGER},
+      </if>
+      <if test="productSkn != null" >
+        product_skn = #{productSkn,jdbcType=INTEGER},
+      </if>
+    </set>
+    where id = #{id,jdbcType=INTEGER}
+  </update>
+  <update id="updateByPrimaryKey" parameterType="com.yoho.search.dal.model.PromotionProduct" >
+    update promotion_product
+    set promotion_id = #{promotionId,jdbcType=INTEGER},
+      product_skn = #{productSkn,jdbcType=INTEGER}
+    where id = #{id,jdbcType=INTEGER}
+  </update>
+</mapper>
\ No newline at end of file

+package com.yoho.search.consumer.index.increment;
+
+import java.util.HashMap;
+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;
+import com.yoho.search.base.utils.ConvertUtils;
+import com.yoho.search.base.utils.EventReportEnum;
+import com.yoho.search.base.utils.ISearchConstants;
+import com.yoho.search.consumer.index.common.IYohoIndexService;
+import com.yoho.search.consumer.service.base.ProductService;
+import com.yoho.search.consumer.service.base.PromotionProductService;
+import com.yoho.search.core.es.utils.IgnoreSomeException;
+import com.yoho.search.dal.model.PromotionProduct;
+
+@Component
+public class PromotionProductMqListener extends AbstractMqListener implements ChannelAwareMessageListener {
+
+	private static final Logger logger = LoggerFactory.getLogger(PromotionProductMqListener.class);
+	@Autowired
+	private IYohoIndexService indexService;
+	@Autowired
+	private PromotionProductService promotionProductService;
+	@Autowired
+	private ProductService productService;
+
+	@Override
+	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=PromotionProductMqListener][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, true);
+			} else {
+				updateData(json.getObject("data", Map.class), key, false);
+			}
+		} catch (Exception e) {
+			publisher.publishEvent(new SearchEvent(EventReportEnum.PROMOTIONPRODUCTMQLISTENER_ONMESSAGE.getEventName(), EventReportEnum.PROMOTIONPRODUCTMQLISTENER_ONMESSAGE
+					.getFunctionName(), EventReportEnum.PROMOTIONPRODUCTMQLISTENER_ONMESSAGE.getMoudleName(), "exception", IgnoreSomeException.filterSomeException(e), null));
+			Thread.sleep(1000);
+			throw e;
+		}
+	}
+
+	public void deleteData(final String id, final String key) throws Exception {
+		long begin = System.currentTimeMillis();
+		PromotionProduct promotionProduct = promotionProductService.selectById(Integer.parseInt(id));
+		if (promotionProduct == null || promotionProduct.getProductSkn()==null){
+			return;
+		}
+		//1、删除数据库信息
+		promotionProductService.delete(Integer.parseInt(id));
+		logger.info("[func=deleteData][step=saveToBb][key={}][cost={}ms]", key, System.currentTimeMillis() - begin);
+		
+		//2、更新ES索引
+		Integer skn = promotionProduct.getProductSkn();
+		this.updateProductIndex(skn, System.currentTimeMillis(), key);
+		logger.info("[func=deleteData][step=updateProductIndex][productSkn={}][cost={}ms]",skn, System.currentTimeMillis() - begin);
+
+	}
+	
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	public void updateData(final Map data, final String key, boolean isUpdate) throws Exception {
+		long begin = System.currentTimeMillis();
+		PromotionProduct promotionProduct = (PromotionProduct) ConvertUtils.transMap2Bean(PromotionProduct.class, data);
+		if (promotionProduct == null || promotionProduct.getPromotionId() == null || promotionProduct.getProductSkn() == null || promotionProduct.getId() == null) {
+			return;
+		}
+		// 1、获取变更的新的SKN和老的SKN信息
+		Integer newProductSkn = promotionProduct.getProductSkn();
+		Integer oldProductSkn = null;
+		PromotionProduct promotionProductOld = null;
+		if (isUpdate) {
+			promotionProductOld = promotionProductService.selectById(promotionProduct.getId());
+		}
+		if (promotionProductOld != null) {
+			oldProductSkn = promotionProductOld.getProductSkn();
+		}
+		logger.info("[func=updateData][step=getSknInfo][key={}][cost={}ms]", key, System.currentTimeMillis() - begin);
+
+		// 2、保存至数据库
+		begin = System.currentTimeMillis();
+		promotionProductService.saveOrUpdate(promotionProduct);
+		logger.info("[func=updateData][step=saveToBb][key={}][cost={}ms]", key, System.currentTimeMillis() - begin);
+
+		// 3、更新ProductIndex
+		this.updateProductIndex(newProductSkn, System.currentTimeMillis(), key);
+		logger.info("[func=updateData][step=updateProductIndex][productSkn={}][cost={}ms]",newProductSkn, System.currentTimeMillis() - begin);
+		
+		// 4、如果老的存在并且被更新掉了，则同时更新老SKN的信息
+		if (oldProductSkn != null && !oldProductSkn.equals(newProductSkn)) {
+			this.updateProductIndex(oldProductSkn, System.currentTimeMillis(), key);
+			logger.info("[func=updateData][step=updateProductIndex][productSkn={}][cost={}ms]",oldProductSkn, System.currentTimeMillis() - begin);
+		}
+	}
+	
+	private void updateProductIndex(Integer productSkn, long currentTimeMillis, String key) {
+		Integer productId = productService.selectProductIdBySkn(productSkn);
+		if(productId==null){
+			return;
+		}
+		String promotionIds = promotionProductService.getPromotionIds(productSkn);
+		if(StringUtils.isBlank(promotionIds)){
+			promotionIds = "";
+		}
+		logger.info("[func=updateProductIndex][key={}][productId={}][promotionIds={}]", key, productId, promotionIds);
+		Map<String, Object> indexData = new HashMap<String, Object>();
+		indexData.put("productId", productId);
+		indexData.put("promotionIds", promotionIds);
+		// 更新商品索引
+		this.updateProductIndexWithDataMap(indexData, productId, key, System.currentTimeMillis());
+	}
+
+}

@@ -685,6 +685,10 @@
                 "type": "string",
                 "analyzer":"comma_spliter"
             },
+            "promotionIds": {
+                "type": "string",
+                "analyzer":"comma_spliter"
+            },
             "standardIds": {
                 "type": "string",
                 "analyzer":"comma_spliter"

@@ -188,6 +188,7 @@ public class ProductIndexService {
 		map.put("searchPageProdBoost", productIndexBO.getSearchPageProdBoost());
 		map.put("specialSearchField", productIndexBO.getSpecialSearchField());
 		map.put("isForbiddenSortBrand", productIndexBO.getIsForbiddenSortBrand());
+		map.put("promotionIds", productIndexBO.getPromotionIds());
 		return map;
 	}
 

+package com.yoho.search.consumer.service.base;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import com.yoho.search.dal.PromotionProductMapper;
+import com.yoho.search.dal.model.PromotionProduct;
+
+/**
+ * erp_product.promotion_product表
+ * @author gemingdan
+ *
+ */
+@Service
+public class PromotionProductService {
+
+	@Autowired
+	private PromotionProductMapper promotionProductMapper;
+	
+	public String getPromotionIds(Integer skn){
+		List<PromotionProduct> list = promotionProductMapper.selectListBySkn(skn);
+		StringBuffer result=new StringBuffer("");
+		for(PromotionProduct pp:list){
+			result.append(pp.getPromotionId()).append(",");
+         }
+		if(result.length()<1){
+			return "";
+		}
+		return result.substring(0,result.length()-1);
+	}
+
+	public void saveOrUpdate(PromotionProduct promotionProduct) {
+		if(promotionProduct.getId()!=null){
+			promotionProductMapper.updateByPrimaryKeySelective(promotionProduct);
+		}else{
+			promotionProductMapper.insertSelective(promotionProduct);
+		}
+	}
+
+	public PromotionProduct selectById(Integer id) {
+		return promotionProductMapper.selectByPrimaryKey(id);
+	}
+
+	public void delete(Integer id) {
+		promotionProductMapper.deleteByPrimaryKey(id);
+	}
+
+}

@@ -93,6 +93,9 @@ public class ProductIndexBO extends ProductIBO implements Serializable {
 
     //form storage
     private Integer storageUpdateTime;
+    
+    //from erp_product.promotion_product
+    private String promotionIds;
 
     // get
     public BigDecimal getSpecialPrice() {
@@ -455,4 +458,12 @@ public class ProductIndexBO extends ProductIBO implements Serializable {
     public void setStorageUpdateTime(Integer storageUpdateTime) {
         this.storageUpdateTime = storageUpdateTime;
     }
+
+	public String getPromotionIds() {
+		return promotionIds;
+	}
+
+	public void setPromotionIds(String promotionIds) {
+		this.promotionIds = promotionIds;
+	}
 }
\ No newline at end of file

 package com.yoho.search.consumer.service.logic;
 
-import com.yoho.search.base.utils.DateUtil;
-import com.yoho.search.base.utils.MathUtils;
-import com.yoho.search.consumer.service.base.Product15DaySalesNumService;
-import com.yoho.search.consumer.service.base.ProductActivitiesLinkService;
-import com.yoho.search.consumer.service.bo.*;
-import com.yoho.search.dal.*;
-import com.yoho.search.dal.model.*;
-import org.apache.commons.collections.CollectionUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
 import java.math.BigDecimal;
 import java.math.RoundingMode;
 import java.text.DecimalFormat;
 import java.util.ArrayList;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import org.apache.commons.collections.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import com.yoho.search.base.utils.DateUtil;
+import com.yoho.search.base.utils.MathUtils;
+import com.yoho.search.consumer.service.base.Product15DaySalesNumService;
+import com.yoho.search.consumer.service.base.ProductActivitiesLinkService;
+import com.yoho.search.consumer.service.bo.ProductActivitiesBO;
+import com.yoho.search.consumer.service.bo.ProductGoodsBO;
+import com.yoho.search.consumer.service.bo.ProductIBO;
+import com.yoho.search.consumer.service.bo.ProductIndexBO;
+import com.yoho.search.consumer.service.bo.ProductPoolDetailSknBO;
+import com.yoho.search.consumer.service.bo.ProductSearchBrandWeightBO;
+import com.yoho.search.consumer.service.bo.ProductSearchShopWeightBO;
+import com.yoho.search.consumer.service.bo.ProductSearchSortWeightBO;
+import com.yoho.search.consumer.service.bo.ProductStandardsBO;
+import com.yoho.search.dal.BasePinRatioMapper;
+import com.yoho.search.dal.BrandMapper;
+import com.yoho.search.dal.BrokenCodeProductMapper;
+import com.yoho.search.dal.DiscountProductSknMapper;
+import com.yoho.search.dal.GoodsImagesMapper;
+import com.yoho.search.dal.GoodsMapper;
+import com.yoho.search.dal.ProductColorsMapper;
+import com.yoho.search.dal.ProductDefaultImageMapper;
+import com.yoho.search.dal.ProductMapper;
+import com.yoho.search.dal.ProductPoolDetailMapper;
+import com.yoho.search.dal.ProductSortMapper;
+import com.yoho.search.dal.ProductStyleMapper;
+import com.yoho.search.dal.ProductStyleRelationMapper;
+import com.yoho.search.dal.PromotionProductMapper;
+import com.yoho.search.dal.model.BasePinRatio;
+import com.yoho.search.dal.model.Brand;
+import com.yoho.search.dal.model.BrokenCode;
+import com.yoho.search.dal.model.DiscountProductSkn;
+import com.yoho.search.dal.model.Product;
+import com.yoho.search.dal.model.Product15DaySalesNum;
+import com.yoho.search.dal.model.ProductActivitiesLink;
+import com.yoho.search.dal.model.ProductColors;
+import com.yoho.search.dal.model.ProductDefaultImage;
+import com.yoho.search.dal.model.ProductFeature;
+import com.yoho.search.dal.model.ProductKeywords;
+import com.yoho.search.dal.model.ProductPrice;
+import com.yoho.search.dal.model.ProductSizes;
+import com.yoho.search.dal.model.ProductSort;
+import com.yoho.search.dal.model.ProductStyles;
+import com.yoho.search.dal.model.PromotionProduct;
+import com.yoho.search.dal.model.StorageUpdateTime;
+
 /**
  * Created by wangnan on 2016/6/29.
  */
@@ -85,6 +124,8 @@ public class ProductIndexLogicService {
     private ProductStyleMapper productStyleMapper;
     @Autowired
     private StorageUpdateTimeLogicService storageUpdateTimeLogicService;
+    @Autowired
+    private PromotionProductMapper promotionProductMapper;
 
     /**
      * 获取ProductIndex列表，用于取代视图。
@@ -203,7 +244,8 @@ public class ProductIndexLogicService {
         Map<Integer, ProductFeature> productFeatureSknsMap = getProductFeatureSknsMap(skns);
         Map<Integer, ProductPoolDetailSknBO> getProductPoolDetailSknMap = getProductPoolDetailSknMap(skns);
         Map<Integer, Product15DaySalesNum> productProduct15DaySalesNumMap = getProduct15DaySalesNumMap(skns);
-
+        Map<Integer, String> promotionProductMap = getPromotionProductMap(skns);
+        
 
         for (ProductIBO productIBO : productIBOs) {
             ProductIndexBO productIndexBO = new ProductIndexBO();
@@ -249,6 +291,9 @@ public class ProductIndexLogicService {
             productIndexBO = buildProductPoolDetailSkn(productIBO, productIndexBO, getProductPoolDetailSknMap);
             //来自StorageUpdateTimeBO
             productIndexBO = buildStorageUpdateTime(productIBO, productIndexBO, getStorageUpdateTimeBOsMap);
+            //来自erp_product.promotion_product
+            productIndexBO = buildPromotionProduct(productIBO, productIndexBO,promotionProductMap);
+            
             productIndexBOs.add(productIndexBO);
         }
         return productIndexBOs;
@@ -706,7 +751,14 @@ public class ProductIndexLogicService {
         }
         return pi;
     }
-
+    private ProductIndexBO buildPromotionProduct(ProductIBO p, ProductIndexBO pi, Map<Integer, String> getPromotionProductMap) {
+    	String pp = getPromotionProductMap.get(p.getProductSkn());
+        if (pp != null) {
+            pi.setPromotionIds(pp);
+        }
+        return pi;
+    }
+    
     /**
      * @param p
      * @param pi
@@ -827,6 +879,35 @@ public class ProductIndexLogicService {
         List<Product15DaySalesNum> product15DaySalesNum = product15DaySalesNumService.getSalesNumListList(skns);
         return product15DaySalesNum.stream().parallel().collect(Collectors.toMap(Product15DaySalesNum::getProductSkn, (p) -> p));
     }
+    
+    /**
+     * 从erp_product.promotion_product表中获取promotionId
+     *
+     * @param skns
+     * @return
+     */
+    private Map<Integer, String> getPromotionProductMap(List<Integer> skns) {
+    	Map<Integer, String> boList= new HashMap<Integer,String>();
+        List<PromotionProduct> list = promotionProductMapper.selectListBySkns(skns);
+        for(PromotionProduct pp:list){
+        	try {
+        		Integer promotionId = pp.getPromotionId();
+        		if(promotionId==null){
+        			continue;
+        		}
+        		String ps=boList.get(pp.getProductSkn());
+        		if(ps==null){
+            		boList.put(pp.getProductSkn(), promotionId.toString());
+            	}else{
+            		boList.put(pp.getProductSkn(), new StringBuffer(ps).append(",").append(promotionId.toString()).toString());
+            	}
+			} catch (Exception e) {
+				logger.error(e.getMessage(),e);
+			}
+        }
+		return boList; 
+    }
+    
 
     /**
      * product_price 表查询

@@ -24,6 +24,7 @@
     <rabbit:admin connection-factory="connectionFactory" />
 
     <!-- queue 队列声明 -->
+    <rabbit:queue durable="true" exclusive="false" name="data_update_channel_promotionproduct" />
     <rabbit:queue durable="true" exclusive="false" name="data_update_channel_esworddef" />
     <rabbit:queue durable="true" exclusive="false" name="data_update_channel_dynamicrule" />
     <rabbit:queue durable="true" exclusive="false" name="data_update_channel_dynamicruledetails" />
@@ -102,6 +103,7 @@
     <bean id="productPriceMqListener4" class="com.yoho.search.consumer.index.increment.ProductPriceMqListener"></bean>
 
     <rabbit:listener-container connection-factory="connectionFactory" task-executor="taskExecutor" concurrency="1">
+        <rabbit:listener queue-names="data_update_channel_promotionproduct" ref="promotionProductMqListener" />
         <rabbit:listener queue-names="data_update_channel_esworddef" ref="esWordDefMqListener" />
         <rabbit:listener queue-names="data_update_channel_dynamicrule" ref="dynamicRuleMqListener" />
         <rabbit:listener queue-names="data_update_channel_dynamicruledetails" ref="dynamicRuleDetailsMqListener" />

@@ -11,9 +11,9 @@ redis.notsync.twemproxy.addresses=${redis.notsync.twemproxy.addresses}
 redis.notsync.twemproxy.auth=${redis.notsync.twemproxy.auth}
 
 #bigData - redis servers
-bigDataRedis-search.proxy.address=${bigDataRedis.proxy.address}
-bigDataRedis-search.proxy.port=${bigDataRedis.proxy.port}
-bigDataRedis-search.proxy.auth=${bigDataRedis.proxy.auth}
+bigDataRedis-search.proxy.address=${bigDataRedis-search.proxy.address}
+bigDataRedis-search.proxy.port=${bigDataRedis-search.proxy.port}
+bigDataRedis-search.proxy.auth=${bigDataRedis-search.proxy.auth}
 
 #mq
 search.mq.server=${search.mq.server}