add

+package com.yoho.search.consumer.index.increment;
+
+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.ShopsBrandsService;
+import com.yoho.search.core.es.model.ESBluk;
+import com.yoho.search.core.es.utils.IgnoreSomeException;
+import com.yoho.search.dal.model.ShopsBrands;
+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 org.springframework.util.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by wangnan on 2016/12/1.
+ */
+@Component
+public class ShopsBrandsMqListener extends AbstractMqListener implements ChannelAwareMessageListener {
+    private static final Logger logger = LoggerFactory.getLogger(ShopsBrandsMqListener.class);
+    @Autowired
+    private IYohoIndexService indexService;
+    @Autowired
+    private ShopsBrandsService shopsBrandsService;
+    @Autowired
+    private ShopsLogicService shopsLogicService;
+
+    @Override
+    public void onMessage(Message message, Channel channel) throws Exception {
+        try {
+            String messageStr = new String(message.getBody(), "UTF-8");
+            logger.info("[model=ShopsBrandsMqListener][message={}]", messageStr);
+            // 如果在重建索引等待
+            this.waitingRebuildingIndex();
+            JSONObject json = JSONObject.parseObject(messageStr);
+            String indexName = ISearchConstants.INDEX_NAME_SHOPS;
+            String idField = ISearchConstants.getKeyField(indexName);
+            if (ISearchConstants.ACTION_DELETE.equals(json.getString("action"))) {
+                deleteData(json.getString("data"), indexName, idField);
+            } else if (ISearchConstants.ACTION_UPDATE.equals(json.getString("action"))) {
+                updateData(json.getObject("data", Map.class), indexName, idField);
+            } else {
+                updateData(json.getObject("data", Map.class), indexName, idField);
+            }
+        } catch (Exception e) {
+            publisher.publishEvent(new SearchEvent(EventReportEnum.SHOPSBRADNSMQLISTENER_ONMESSAGE.getEventName(),
+                    EventReportEnum.SHOPSBRADNSMQLISTENER_ONMESSAGE.getFunctionName(),
+                    EventReportEnum.SHOPSBRADNSMQLISTENER_ONMESSAGE.getMoudleName(), "exception", IgnoreSomeException.filterSomeException(e), null));
+            Thread.sleep(1000);
+            throw e;
+        }
+    }
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    public void updateData(final Map data, final String indexName, final String idField) throws Exception {
+        long begin = System.currentTimeMillis();
+        ShopsBrands shopsBrands = new ShopsBrands();
+        shopsBrands = (ShopsBrands) ConvertUtils.toJavaBean(shopsBrands, data);
+        if (shopsBrands == null || shopsBrands.getShopsId() == null) {
+            return;
+        }
+        String idValue = shopsBrands.getId().toString();
+        shopsBrandsService.saveOrUpdate(shopsBrands);
+        logger.info("[func=updateData][step=saveToDb][indexName={}] [id={}][cost={}ms]", indexName, idValue, (System.currentTimeMillis() - begin));
+        List<Integer> ids = new ArrayList<>();
+        ids.add(shopsBrands.getShopsId());
+        List<ShopsBO> shopBOs = shopsLogicService.getShopsBOs(ids);
+        if (!CollectionUtils.isEmpty(shopBOs)) {
+            ShopsBO shopsBO = shopBOs.get(0);
+            List<ESBluk> results = new ArrayList<ESBluk>();
+            results.add(new ESBluk(JSONObject.toJSONString(this.beanToMap(shopsBO)), shopsBO.getShopsId().toString(), indexName, indexName, false));
+            indexService.bulk(results);
+        }
+        logger.info("[func=updateData][step=success][indexName={}] [id={}][cost={}ms]", indexName, idValue, (System.currentTimeMillis() - begin));
+    }
+
+    public void deleteData(final String id, final String indexName, final String idField) throws Exception {
+        if (StringUtils.isBlank(id)) {
+            return;
+        }
+        long begin = System.currentTimeMillis();
+        ShopsBrands shopsBrands = shopsBrandsService.getById(Integer.valueOf(id));
+        shopsBrandsService.delete(Integer.valueOf(id));
+        logger.info("[func=deleteData][step=deleteFromDb][indexName={}] [id={}][cost={}ms]", indexName, id, (System.currentTimeMillis() - begin));
+        List<Integer> ids = new ArrayList<>();
+        ids.add(shopsBrands.getShopsId());
+        List<ShopsBO> shopBOs = shopsLogicService.getShopsBOs(ids);
+        if (!CollectionUtils.isEmpty(shopBOs)) {
+            ShopsBO shopsBO = shopBOs.get(0);
+            List<ESBluk> results = new ArrayList<ESBluk>();
+            results.add(new ESBluk(JSONObject.toJSONString(this.beanToMap(shopsBO)), shopsBO.getShopsId().toString(), indexName, indexName, false));
+            indexService.bulk(results);
+        }
+        logger.info("[func=deleteData][step=success][indexName={}][id={}][cost={}ms]", indexName, id, (System.currentTimeMillis() - begin));
+    }
+
+    private Map<String, Object> beanToMap(ShopsBO shopsBO) {
+        JSONObject josnoJsonObject = (JSONObject) JSONObject.toJSON(shopsBO);
+        return josnoJsonObject;
+    }
+}