|
|
|
package com.yoho.search.consumer.index.increment.bulks;
|
|
|
|
|
|
|
|
import com.alibaba.fastjson.JSON;
|
|
|
|
import com.alibaba.fastjson.JSONObject;
|
|
|
|
import com.yoho.error.event.SearchEvent;
|
|
|
|
import com.yoho.search.base.utils.ISearchConstants;
|
|
|
|
import com.yoho.search.base.utils.MoudleEnum;
|
|
|
|
import com.yoho.search.consumer.common.IYohoIndexService;
|
|
|
|
import com.yoho.search.consumer.index.fullbuild.ufo.UfoProductIndexBuilder;
|
|
|
|
import com.yoho.search.consumer.service.bo.UfoProductIndexBO;
|
|
|
|
import com.yoho.search.core.es.model.ESBluk;
|
|
|
|
import com.yoho.search.core.es.utils.IgnoreSomeException;
|
|
|
|
import org.apache.commons.collections.CollectionUtils;
|
|
...
|
...
|
@@ -26,10 +30,12 @@ public class UfoCommonBulkService implements ApplicationEventPublisherAware { |
|
|
|
|
|
|
|
private static final Logger logger = LoggerFactory.getLogger(UfoCommonBulkService.class);
|
|
|
|
|
|
|
|
private final ArrayBlockingQueue<ESBluk> queue = new ArrayBlockingQueue<>(200);
|
|
|
|
private final ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(200);
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private IYohoIndexService yohoIndexService;
|
|
|
|
@Autowired
|
|
|
|
private UfoProductIndexBuilder ufoProductIndexBuilder;
|
|
|
|
|
|
|
|
protected ApplicationEventPublisher publisher;
|
|
|
|
|
|
...
|
...
|
@@ -44,39 +50,59 @@ public class UfoCommonBulkService implements ApplicationEventPublisherAware { |
|
|
|
ExecutorService executorService = Executors.newSingleThreadExecutor();
|
|
|
|
executorService.submit(() -> {
|
|
|
|
while (true) {
|
|
|
|
doBulk();
|
|
|
|
}
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
|
|
|
private void doBulk() {
|
|
|
|
try {
|
|
|
|
long begin = System.currentTimeMillis();
|
|
|
|
|
|
|
|
//1、从队列中获取全部数据
|
|
|
|
List<ESBluk> blukList = new ArrayList<>();
|
|
|
|
queue.drainTo(blukList);
|
|
|
|
List<Integer> idList = new ArrayList<>();
|
|
|
|
queue.drainTo(idList);
|
|
|
|
|
|
|
|
//2、批量更新Es
|
|
|
|
if (CollectionUtils.isNotEmpty(blukList)) {
|
|
|
|
yohoIndexService.bulk(blukList);
|
|
|
|
logger.info("doBulk, the blukList size is {} and cost {} ms,", blukList.size(), System.currentTimeMillis() - begin);
|
|
|
|
Thread.sleep(50);
|
|
|
|
if (CollectionUtils.isNotEmpty(idList)) {
|
|
|
|
doBulkToEs(idList);
|
|
|
|
logger.info("doBulk, the blukList size is {} and cost {} ms,", idList.size(), System.currentTimeMillis() - begin);
|
|
|
|
Thread.sleep(100);
|
|
|
|
} else {
|
|
|
|
Thread.sleep(2000);
|
|
|
|
}
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
publisher.publishEvent(new SearchEvent("doBulk", "UfoCommonBulkService.doBulk", MoudleEnum.consumer, "exception", IgnoreSomeException.filterSomeException(e), null));
|
|
|
|
publisher.publishEvent(new SearchEvent("doBulk", "UfoCommonBulkByIdService.doBulkToEs", MoudleEnum.consumer, "exception", IgnoreSomeException.filterSomeException(e), null));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
|
|
|
private void doBulkToEs(List<Integer> idList) {
|
|
|
|
List<UfoProductIndexBO> ufoProductIndexBOList = ufoProductIndexBuilder.buildUfoProductIndexBOIncrease(idList);
|
|
|
|
if (ufoProductIndexBOList == null || ufoProductIndexBOList.isEmpty()) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
List<ESBluk> esBlukList = new ArrayList<>();
|
|
|
|
for (UfoProductIndexBO ufoProductIndexBO : ufoProductIndexBOList) {
|
|
|
|
JSONObject ufoProductIndexBOJSONObject = (JSONObject) JSON.toJSON(ufoProductIndexBO);
|
|
|
|
ESBluk esBluk = new ESBluk(ufoProductIndexBOJSONObject.toJSONString(), ufoProductIndexBO.getId().toString(), ISearchConstants.INDEX_NAME_UFO_PRODUCT_INDEX, ISearchConstants.INDEX_NAME_UFO_PRODUCT_INDEX, false);
|
|
|
|
esBlukList.add(esBluk);
|
|
|
|
}
|
|
|
|
yohoIndexService.bulk(esBlukList);
|
|
|
|
}
|
|
|
|
|
|
|
|
public void addUpdateData(String jsonString, Integer productId) {
|
|
|
|
try {
|
|
|
|
// ESBluk esBluk = new ESBluk(jsonString, productId.toString(), ISearchConstants.INDEX_NAME_UFO_PRODUCT_INDEX, ISearchConstants.INDEX_NAME_UFO_PRODUCT_INDEX, false);
|
|
|
|
// dataIncrease.put(esBluk);
|
|
|
|
queue.put(productId);
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
public void addUpdateData(String jsonString,Integer id) {
|
|
|
|
public void addUfoUpdateProductId(Integer ufoProductId) {
|
|
|
|
try {
|
|
|
|
ESBluk esBluk = new ESBluk(jsonString, id.toString(), ISearchConstants.INDEX_NAME_UFO_PRODUCT_INDEX, ISearchConstants.INDEX_NAME_UFO_PRODUCT_INDEX, false);
|
|
|
|
queue.put(esBluk);
|
|
|
|
queue.put(ufoProductId);
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
} |
...
|
...
|
|
