对es搜索使用fork/join优化搜索

需求：查询某个日期区间关于吉利相关舆情

es索引库有12个，吉利相关词有50+个
查询12个es索引库，查询出日期区间的结果且满足这50+词中其中一个词则将对应的文章查询出来

1个索引
       查询
           一个月区间日期 2019-10-01 2019-10-31
           多个词(吉利、帝豪....)包含这50个词的都查出来
   思路：按索引、日期、词维度来拆分
   如：按5天时间和 10个词一组去es查询返回，将日期区间缩短和查询的词量减少

   5天日期区间查询 + 10个词，
   第一组日期分组
       2019-10-01、2019-10-02、2019-10-03、2019-10-04、2019-10-05
       吉利、帝豪、Geely、博越、帝豪GL、帝豪GS、帝豪RS、帝豪吉利、帝豪汽车、远景X1
       2019-10-01、2019-10-02、2019-10-03、2019-10-04、2019-10-05
       远景SUV、远景X3、吉利自由舰、吉利熊猫、吉利金刚、吉利博瑞、吉利新博瑞、吉利英伦、英伦C5、英伦两厢
       2019-10-01、2019-10-02、2019-10-03、2019-10-04、2019-10-05
       吉利海景.......
       2019-10-01、2019-10-02、2019-10-03、2019-10-04、2019-10-05
       新美日汽车......
       2019-10-01、2019-10-02、2019-10-03、2019-10-04、2019-10-05
       吉利新美日.....
   第二组日期分组
       2019-10-06、2019-10-07、2019-10-08、2019-10-09、2019-10-10
       吉利、帝豪、Geely、博越、帝豪GL、帝豪GS、帝豪RS、帝豪吉利、帝豪汽车、远景X1
       2019-10-06、2019-10-07、2019-10-08、2019-10-09、2019-10-10
       远景SUV、远景X3、吉利自由舰、吉利熊猫、吉利金刚、吉利博瑞、吉利新博瑞、吉利英伦、英伦C5、英伦两厢
       2019-10-06、2019-10-07、2019-10-08、2019-10-09、2019-10-10
       吉利海景.......
       2019-10-06、2019-10-07、2019-10-08、2019-10-09、2019-10-10
       新美日汽车......
       2019-10-06、2019-10-07、2019-10-08、2019-10-09、2019-10-10
       吉利新美日.....
   第三组日期分组.......
       ......
       ......
   第四组日期分组.......
       ......
       ......
   第五组日期分组.......
       ......
       ......
   (30/5)(日期组) * 5(词组) * 12(索引) = 360(一个月区组成360次查询es)

使用fork/join 对这360条数据做分解，以30个为基准提交搜索到es，360/30=12，需要查询12次es，如果每一次需要花费2秒，最终执行时间不会超过3秒。

本地验证时，发现使用fork/join比单次慢，原因是因为本地cpu大多都是保持在60%使用，fork/join是cpu密集型，当cpu被暂满时就会出现线程之间的竞争等待，
所以移到支持cpu密集型的服务器测试，是会比较快。

未优化前

优化后

public BucketHit getBrand(SentimentParam param, QueryBuilder qb, String... fieldNames) {List<String> dateTimes = resolveDate(param.getStartDate(),param.getEndDate());String startDate = param.getStartDate();//搜索条件list，根据日期分解搜索次数List<SearchSourceBuilder> searchBuilders = Lists.newArrayListWithExpectedSize(dateTimes.size() * 4);for (String date : dateTimes){//每一个查询 日期间隔5天 且 查询 10个词searchBuilders.add(getSearchBuilder(startDate,date,TitleTemplateUtil.geelyOrTitle().subList(0,10),param));searchBuilders.add(getSearchBuilder(startDate,date,TitleTemplateUtil.geelyOrTitle().subList(10,20),param));searchBuilders.add(getSearchBuilder(startDate,date,TitleTemplateUtil.geelyOrTitle().subList(20,30),param));searchBuilders.add(getSearchBuilder(startDate,date,TitleTemplateUtil.geelyOrTitle().subList(30,40),param));searchBuilders.add(getSearchBuilder(startDate,date,TitleTemplateUtil.geelyOrTitle().subList(40,TitleTemplateUtil.geelyOrs.size()),param));startDate = date;}/*循环索引列表，每一个索引都对应 searchBuilders.size()次 搜索如 索引 12 个， 时间区间 36 条总共查询次数 12 * 36*/List<SearchRequest> srs = Lists.newArrayList();for (SentimentIndexEnum index : SentimentIndexEnum.values()) {for (SearchSourceBuilder ssb : searchBuilders) {SearchRequest searchRequest = new SearchRequest(index.getIndex());searchRequest.source(ssb);srs.add(searchRequest);}}ForkJoinPool forkJoinPool = new ForkJoinPool(20);long st3 = System.currentTimeMillis();   //获取开始时间//使用forkJoin分解任务List<SearchResponse> responses = forkJoinPool.invoke(new IndexForkRecursiveTask(srs));//同步调用/*MultiSearchRequest request = new MultiSearchRequest();for (SearchRequest sr : srs) {request.add(sr);}List<SearchResponse> responses = getMultiSearchResponse(request);*/forkJoinPool.shutdown();long et3 = System.currentTimeMillis(); //获取结束时间LOGGER.info("-------总耗时--------"+Thread.currentThread().getName() + "-耗时：" + (st3 - et3) + "ms");BucketHit bucketHit = doGetResponse(responses, fieldNames);return bucketHit;}/*** 分解日期，按 5天维度* 如 sd=2019-10-01，ed=2019-10-31*      返回 [2019-10-06, 2019-10-11, 2019-10-16, 2019-10-21, 2019-10-26, 2019-10-31]* @param sd* @param ed* @return*/private static List<String> resolveDate(String sd, String ed )  {Set<String> dateTimes = Sets.newHashSet();try {Date afterFiveDate = DateUtil.DAY.parse(sd);while (afterFiveDate.before(DateUtil.DAY.parse(ed))){afterFiveDate = DateUtil.getFetureDate(afterFiveDate,5);dateTimes.add(DateUtil.DAY.format(afterFiveDate));}dateTimes.add(ed);return dateTimes.stream().sorted().collect(Collectors.toList());}catch (Exception e){}return null;}

public class IndexForkRecursiveTask extends RecursiveTask<List<SearchResponse>> {private static final Logger LOGGER = LoggerFactory.getLogger(IndexForkRecursiveTask.class);private int num = 20;private List<SearchRequest> searchRequests;private static EsClient esClient;List<IndexForkRecursiveTask> tasks;public IndexForkRecursiveTask(List<SearchRequest> searchRequests) {this.searchRequests = searchRequests;}@Overrideprotected List<SearchResponse> compute() {tasks = Lists.newArrayList();if (searchRequests.size() <= num) {return responses(searchRequests);}// 将搜索条件分组，将大任务拆分小任务List<List<SearchRequest>> ssbs = getList();for (List<SearchRequest> l : ssbs) {IndexForkRecursiveTask frt = new IndexForkRecursiveTask(l);tasks.add(frt);}/* 执行所有任务 并汇总结果*/invokeAll(tasks);List<SearchResponse> responses = Lists.newCopyOnWriteArrayList();for (IndexForkRecursiveTask fr : tasks) {responses.addAll(fr.join());}return responses;}private List<SearchResponse> responses(List<SearchRequest> srs) {long st3 = System.currentTimeMillis();   //获取开始时间MultiSearchRequest request = new MultiSearchRequest();for (SearchRequest sr : srs) {request.add(sr);}List<SearchResponse> responses = Lists.newArrayList();MultiSearchResponse sr;RestHighLevelClient client = getEsClient().getRhlClient();try {LOGGER.info("client =========="+client);sr = client.multiSearch(request);for (MultiSearchResponse.Item item : sr.getResponses()) {SearchResponse response = item.getResponse();responses.add(response);}} catch (Exception e) {LOGGER.error("IndexRecursiveTask compute error ", e);} finally {esClient.close(client);}long et3 = System.currentTimeMillis(); //获取结束时间LOGGER.info(Thread.currentThread().getName() + "-耗时：" + (st3 - et3) + "ms");return responses;}//将多个查询拆小，每组num条private List<List<SearchRequest>> getList() {List<List<SearchRequest>> ret = Lists.newArrayList();int size = searchRequests.size() - 1;if (size <= num) {ret.add(searchRequests);return ret;}int start = 0;int end = num;for (int i = start; i < end; ) {if (end <= size) {ret.add(searchRequests.subList(start, end));start = end;end = end + num;} else {ret.add(searchRequests.subList(start, size));break;}}return ret;}private EsClient getEsClient() {if (esClient == null) {esClient = (EsClient) SpringBeanUtils.getApplicationContext().getBean("esClient");}return esClient;}
}

使用fork/join问题：
如果分解出A、B、C线程，A先执行完,去窃取C线程的任务，A先执行花费1ms、B、C执行也花费1ms，但是A先执行完去窃取C任务导致A现在多负担了一个任务，最后A执行花费2ms,
需要每个任务对应一个线程解决这问题？但是fork/join是不是失去意义。”任务太少拆解问题“

如果不分解则会查询出4秒。

对es搜索使用fork/join优化搜索相关推荐

Elasticsearch优化——搜索速度优化
Elasticsearch优化--搜索速度优化文章目录 Elasticsearch优化--搜索速度优化 1. 为文件系统cache预留足够的内存 2. 使用更快的硬件 3. 文档模型 4. 预索引数 ...
白话Elasticsearch04- 结构化搜索之使用terms query搜索多个值以及多值搜索结果优化
文章目录 terms概述准备数据小例子搜索articleID为KDKE-B-9947-#kL5或QQPX-R-3956-#aD8的帖子搜索tag中包含java的帖子优化搜索结果,仅仅搜索ta ...
阿里巴巴复杂搜索系统的可靠性优化之路
闲鱼技术-元茂 1. 背景搜索引擎是电商平台成交链路的核心环节,搜索引擎的高可用直接影响成交效率.闲鱼搜索引擎作为闲鱼关键系统,复杂度和系统体量都非常高,再加上闲鱼所有导购场景都依靠搜索赋能,搜索服 ...
阿里巴巴复杂搜索系统的可靠性优化之路 1
背景 \n 搜索引擎是电商平台成交链路的核心环节,搜索引擎的高可用直接影响成交效率.闲鱼搜索引擎作为闲鱼关键系统,复杂度和系统体量都非常高,再加上闲鱼所有导购场景都依靠搜索赋能,搜索服务的稳定可靠成为 ...
2010提高组-乌龟棋 [记忆优化搜索]
题目: 有n个格子,你有m张牌,每张牌上面有个数字,用了这张牌你就可以走这张牌上对应的数字了,每到达一个格子,你就会获得格子上相应的分数,问你如何合理的出牌,使分数最大化. 记忆优化搜索可能会好写一点 ...
记忆优化搜索（简单题）（洛谷P3183 [HAOI2016]食物链）（ P5635 【CSGRound1】天下第一）
昨天做了蓝桥杯的时候,发现自己对于记忆优化搜索甚是不熟悉,所以今天随便找了几个基础题做做,顺便写下两片题解,顺便用了一下devc++敲的代码,发现没有代码补全真的可以说是灰常难受了... 洛谷P318 ...
PyTorch机器学习自动化：自动框架搜索、超参优化
https://www.toutiao.com/a6643732469998699022/ [导读]AutoML-Freiburg在Github上发布了PyTorch机器学习自动化的资源,可以自动框架 ...
使用Fork/Join框架优化归并排序
Fork/Join框架是一个非常有意思的并发框架,它非常适合于处理类似归并排序这种将大的问题分解成多个小问题,并将结果进行合并的情况,这次我们就使用Fork/Join框架来优化我们的归并排序.查看更多 ...
京东的商品搜索功能是如何实现的_【干货小知识】京东商家如何优化搜索流量？...
在京东搜索流量是非常重要的一个入口,优化搜索流量时我们要从产品的标题先入手. 首要保证的是你要做的关键词都在标题里面,然后分析客户的搜索习惯. 通常会有以下几种搜索情况: 1.搜索商品核心关键词+ ...

对es搜索使用fork/join优化搜索

对es搜索使用fork/join优化搜索相关推荐

最新文章

热门文章