|
|
|
package com.yoho.search.consumer.index.increment.bulks;
|
|
|
|
|
|
|
|
import com.alibaba.fastjson.JSON;
|
|
|
|
import com.yoho.search.base.models.PromotionIndexBO;
|
|
|
|
import com.yoho.search.base.utils.ISearchConstants;
|
|
|
|
import com.yoho.search.consumer.common.IYohoIndexService;
|
|
|
|
import com.yoho.search.consumer.index.fullbuild.promotion.PromotionIndexIndexBuilder;
|
|
|
|
import com.yoho.search.consumer.service.utils.CostStatistics;
|
|
|
|
import com.yoho.search.core.es.model.ESBluk;
|
|
|
|
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 javax.annotation.PostConstruct;
|
|
|
|
import java.util.ArrayList;
|
|
|
|
import java.util.List;
|
|
|
|
import java.util.concurrent.ArrayBlockingQueue;
|
|
|
|
import java.util.concurrent.ExecutorService;
|
|
|
|
import java.util.concurrent.Executors;
|
|
|
|
|
|
|
|
/**
|
|
|
|
* 促销索引BULK服务
|
|
|
|
* @author wangnan
|
|
|
|
* @version 2018/12/17
|
|
|
|
*/
|
|
|
|
@Component
|
|
|
|
public class PromotionBulkService {
|
|
|
|
|
|
|
|
private static final Logger logger = LoggerFactory.getLogger(PromotionBulkService.class);
|
|
|
|
private static final String INDEX_NAME_PROMOTION_INDEX = ISearchConstants.INDEX_NAME_PROMOTIONINDEX;
|
|
|
|
private static final long THREAD_SLEEP_IDLE = 2000;
|
|
|
|
private static final long THREAD_SLEEP_WORK = 50;
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
private IYohoIndexService indexService;
|
|
|
|
@Autowired
|
|
|
|
private PromotionIndexIndexBuilder promotionIndexIndexBuilder;
|
|
|
|
|
|
|
|
private ArrayBlockingQueue<Integer> idUpdateQueue = new ArrayBlockingQueue<Integer>(200);
|
|
|
|
private ArrayBlockingQueue<Integer> idDeleteQueue = new ArrayBlockingQueue<Integer>(200);
|
|
|
|
|
|
|
|
private ExecutorService updateExecutorService = Executors.newSingleThreadExecutor();
|
|
|
|
private ExecutorService deleteExecutorService = Executors.newSingleThreadExecutor();
|
|
|
|
|
|
|
|
@PostConstruct
|
|
|
|
void init() {
|
|
|
|
updateExecutorService.submit((Runnable) () -> {
|
|
|
|
while (true) {
|
|
|
|
try {
|
|
|
|
long threadSleep = doBulkUpdateIndex();
|
|
|
|
Thread.sleep(threadSleep);
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
});
|
|
|
|
deleteExecutorService.submit((Runnable) () -> {
|
|
|
|
while (true) {
|
|
|
|
try {
|
|
|
|
long threadSleep = doBulkDeleteIndex();
|
|
|
|
Thread.sleep(threadSleep);
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
});
|
|
|
|
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
|
|
|
|
updateExecutorService.shutdown();
|
|
|
|
deleteExecutorService.shutdown();
|
|
|
|
}));
|
|
|
|
}
|
|
|
|
|
|
|
|
public void updateIndex(Integer id) {
|
|
|
|
try {
|
|
|
|
if (id == null) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
idUpdateQueue.put(id);
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
public void deleteIndex(Integer id) {
|
|
|
|
try {
|
|
|
|
if (id == null) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
idDeleteQueue.put(id);
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private long doBulkUpdateIndex() {
|
|
|
|
try {
|
|
|
|
CostStatistics costStatistics = new CostStatistics();
|
|
|
|
List<PromotionIndexBO> promotionIndexBOList = new ArrayList<>();
|
|
|
|
// 1、根据id去取数据库对象
|
|
|
|
List<Integer> idList = this.getElementsFromArrayBlockingQueue(idUpdateQueue, 200);
|
|
|
|
if (!idList.isEmpty()) {
|
|
|
|
promotionIndexBOList = promotionIndexIndexBuilder.buildPromotionIndexBOList(idList);
|
|
|
|
logger.info("doBulkUpdatePromotionIndex buildPromotionIndexBOList, size is [{}], cost [{}]ms", promotionIndexBOList.size(), costStatistics.getCost());
|
|
|
|
}
|
|
|
|
// 2、没数据直接返回[并让线程休息1000ms]
|
|
|
|
if (promotionIndexBOList.isEmpty()) {
|
|
|
|
return THREAD_SLEEP_IDLE;
|
|
|
|
}
|
|
|
|
// 3、批量更新ES
|
|
|
|
this.updateIndexByList(promotionIndexBOList);
|
|
|
|
logger.info("doBulkUpdatePromotionIndex updateIndexByList, cost [{}]ms", costStatistics.getCost());
|
|
|
|
// 4、打印总耗时
|
|
|
|
logger.info("doBulkUpdatePromotionIndex end , costStatistics is [{}] ", costStatistics.getCostStatisticsString());
|
|
|
|
// 让线程休息50ms继续工作
|
|
|
|
return THREAD_SLEEP_WORK;
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
return THREAD_SLEEP_WORK;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private long doBulkDeleteIndex() {
|
|
|
|
try {
|
|
|
|
CostStatistics costStatistics = new CostStatistics();
|
|
|
|
// 1、根据id去取数据库对象
|
|
|
|
List<Integer> ids = this.getElementsFromArrayBlockingQueue(idDeleteQueue, 200);
|
|
|
|
// 2、没数据直接返回[并让线程休息1000ms]
|
|
|
|
if (ids.isEmpty()) {
|
|
|
|
return THREAD_SLEEP_IDLE;
|
|
|
|
}
|
|
|
|
// 3、批量更新ES
|
|
|
|
this.deleteIndexByList(ids);
|
|
|
|
logger.info("doBulkDeletePromotionIndex start, ids size is [{}], cost [{}]ms", ids.size(), costStatistics.getCost());
|
|
|
|
// 4、打印总耗时
|
|
|
|
logger.info("doBulkDeletePromotionIndex end , costStatistics is [{}] ", costStatistics.getCostStatisticsString());
|
|
|
|
return THREAD_SLEEP_WORK;// 让线程休息50ms继续工作
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
return THREAD_SLEEP_WORK;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private void updateIndexByList(List<PromotionIndexBO> promotionIndexBOList) {
|
|
|
|
try {
|
|
|
|
if (CollectionUtils.isEmpty(promotionIndexBOList)) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
List<ESBluk> results = new ArrayList<ESBluk>();
|
|
|
|
for (PromotionIndexBO promotionIndexBO : promotionIndexBOList) {
|
|
|
|
results.add(new ESBluk(JSON.toJSONString(promotionIndexBO), promotionIndexBO.getId().toString(), INDEX_NAME_PROMOTION_INDEX, INDEX_NAME_PROMOTION_INDEX, false));
|
|
|
|
}
|
|
|
|
indexService.bulk(results);
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private void deleteIndexByList(List<Integer> ids) {
|
|
|
|
try {
|
|
|
|
if (CollectionUtils.isEmpty(ids)) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
List<ESBluk> results = new ArrayList<ESBluk>();
|
|
|
|
for (Integer id : ids) {
|
|
|
|
results.add(new ESBluk(null, id.toString(), INDEX_NAME_PROMOTION_INDEX, INDEX_NAME_PROMOTION_INDEX, true));
|
|
|
|
}
|
|
|
|
indexService.bulk(results);
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private <T> List<T> getElementsFromArrayBlockingQueue(ArrayBlockingQueue<T> queue, int limit) {
|
|
|
|
List<T> results = new ArrayList<T>();
|
|
|
|
while (results.size() < limit && (!queue.isEmpty())) {
|
|
|
|
results.add(queue.poll());
|
|
|
|
}
|
|
|
|
return results;
|
|
|
|
}
|
|
|
|
} |
...
|
...
|
|
