|
|
|
package com.yoho.search.consumer.index.increment;
|
|
|
|
|
|
|
|
import com.alibaba.fastjson.JSONObject;
|
|
|
|
import com.rabbitmq.client.Channel;
|
|
|
|
import com.yoho.search.dal.model.SearchAction;
|
|
|
|
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.ISearchConstans;
|
|
|
|
import com.yoho.search.consumer.index.common.IYohoIndexService;
|
|
|
|
import com.yoho.search.consumer.service.base.ProductService;
|
|
|
|
import com.yoho.search.consumer.service.base.SearchActionService;
|
|
|
|
import com.yoho.search.core.es.utils.IgnoreSomeException;
|
|
|
|
|
|
|
|
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 java.util.HashMap;
|
|
|
|
import java.util.Map;
|
|
|
|
import java.util.UUID;
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Created by YOHO on 15-9-2.
|
|
|
|
*/
|
|
|
|
@Component
|
|
|
|
public class SearchActionMqListener extends AbstractMqListener implements ChannelAwareMessageListener {
|
|
|
|
|
|
|
|
private static final Logger logger = LoggerFactory.getLogger(SearchActionMqListener.class);
|
|
|
|
@Autowired
|
|
|
|
private IYohoIndexService indexService;
|
|
|
|
@Autowired
|
|
|
|
private SearchActionService searchActionService;
|
|
|
|
@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=SearchActionMqListener][key={}][message={}]", key, messageStr);
|
|
|
|
// 如果在重建索引等待
|
|
|
|
this.waitingRebuildingIndex();
|
|
|
|
JSONObject json = JSONObject.parseObject(messageStr);
|
|
|
|
if (ISearchConstans.ACTION_DELETE.equals(json.getString("action"))) {
|
|
|
|
deleteData(json.getString("data"), key);
|
|
|
|
} else if (ISearchConstans.ACTION_UPDATE.equals(json.getString("action"))) {
|
|
|
|
updateData(json.getObject("data", Map.class), key);
|
|
|
|
} else {
|
|
|
|
updateData(json.getObject("data", Map.class), key);
|
|
|
|
}
|
|
|
|
} catch (Exception e) {
|
|
|
|
publisher.publishEvent(new SearchEvent(EventReportEnum.SEARCHACTIONMQLISTENER_ONMESSAGE.getEventName(),
|
|
|
|
EventReportEnum.SEARCHACTIONMQLISTENER_ONMESSAGE.getFunctionName(),
|
|
|
|
EventReportEnum.SEARCHACTIONMQLISTENER_ONMESSAGE.getMoudleName(),"exception",IgnoreSomeException.filterSomeException(e),null));
|
|
|
|
Thread.sleep(1000);
|
|
|
|
throw e;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
@SuppressWarnings({ "rawtypes", "unchecked" })
|
|
|
|
public void updateData(final Map data, final String key) {
|
|
|
|
long begin = System.currentTimeMillis();
|
|
|
|
SearchAction searchAction = new SearchAction();
|
|
|
|
searchAction = (SearchAction) ConvertUtils.toJavaBean(searchAction, data);
|
|
|
|
if (searchAction == null || searchAction.getId() == null) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
searchActionService.saveOrUpdate(searchAction);
|
|
|
|
logger.info("[func=updateData][step=saveToDb][key={}][cost={}ms]", key, System.currentTimeMillis() - begin);
|
|
|
|
Integer productSkn = searchAction.getProductSkn();
|
|
|
|
if (productSkn == null) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
updateProductIndex(productSkn, begin, key);
|
|
|
|
}
|
|
|
|
|
|
|
|
public void deleteData(final String id, final String key) {
|
|
|
|
if (StringUtils.isBlank(id)) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
long begin = System.currentTimeMillis();
|
|
|
|
Integer idInt = Integer.parseInt(id);
|
|
|
|
SearchAction searchAction = searchActionService.getById(idInt);
|
|
|
|
if (searchAction == null) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
searchActionService.delete(idInt);// 先从数据库删除数据
|
|
|
|
logger.info("[func=deleteData][step=deleteFromDb][key={}][cost={}ms]", key, System.currentTimeMillis() - begin);
|
|
|
|
this.clearProductIndex(searchAction.getProductSkn(), begin, key);
|
|
|
|
}
|
|
|
|
|
|
|
|
private void clearProductIndex(Integer productSkn, long begin, final String key) {
|
|
|
|
Integer productId = productService.selectProductIdBySkn(productSkn);
|
|
|
|
if (productId == null) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
Map<String, Object> indexData = new HashMap<String, Object>();
|
|
|
|
indexData.put("productId", productId);
|
|
|
|
indexData.put("saleAction", 0);
|
|
|
|
this.updateProductIndexWithDataMap(indexData, productId, key, begin);
|
|
|
|
logger.info("[func=clearProductIndex][step=clearProductIndex][key={}][productId={}][cost={}ms]", key, productId,System.currentTimeMillis() - begin);
|
|
|
|
}
|
|
|
|
|
|
|
|
private void updateProductIndex(Integer productSkn, long begin, final String key) {
|
|
|
|
Map<String, Object> indexData = searchActionService.getSaleActionMap(productSkn);
|
|
|
|
if (indexData == null || indexData.size() == 0) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
Object productId = indexData.get("productId");
|
|
|
|
if (productId == null) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
Integer productIdInt = Integer.valueOf(productId.toString());
|
|
|
|
this.updateProductIndexWithDataMap(indexData, productIdInt, key, begin);
|
|
|
|
logger.info("[func=updateProductIndex][step=updateProductIndex][key={}][productId={}][cost={}ms]", key,productId, System.currentTimeMillis() - begin);
|
|
|
|
}
|
|
|
|
} |
