|
|
|
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.consumer.service.bo.ShopsBO;
|
|
|
|
import com.yoho.search.consumer.service.logic.ShopsLogicService;
|
|
|
|
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;
|
|
|
|
}
|
|
|
|
} |
