searchRecord调整

src/main/java/com/zhiwei/brandkbs2/es/EsClientDao.java

@@ -60,6 +60,8 @@ public class EsClientDao {
             "brandkbs_cache_maps", "brandkbs_mark_cache_maps", "channel_influence", "ind_full_text", "url", "question_url", "answer_url"};
 
     private static final Long ONE_HOUR = 60 * 60 * 1000L;
+
+    private static final Long HALF_HOUR = ONE_HOUR / 2;
     // 滚动查询超时时间
     private static final TimeValue TIME_VALUE = TimeValue.timeValueMinutes(8);
 
@@ -103,12 +105,34 @@ public class EsClientDao {
         long startTime = calendar.getTime().getTime();
         List<Long[]> cutTimes = Tools.cutTimeRange(startTime, endTime, ONE_HOUR);
         List<CompletableFuture<Pair<Long[], Map<ChannelIndex, ChannelIndex.Record>>>> futures = new ArrayList<>(cutTimes.size());
-        cutTimes.forEach(times -> futures.add(CompletableFuture.supplyAsync(() -> searchRecord(times[0], times[1]), executor)));
-        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).whenComplete((r, e) -> {
-            futures.forEach(f -> {
-                res.add(f.join());
-            });
-        }).join();
+        try {
+            cutTimes.forEach(times -> futures.add(CompletableFuture.supplyAsync(() -> {
+                try {
+                    return searchRecord(times[0], times[1]);
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                }
+            }, executor)));
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).whenComplete((r1, e1) -> futures.forEach(f -> res.add(f.join()))).join();
+        }catch (Exception e){
+            log.error("error-searchRecordRecentDay-oneHour,跨度调整为半小时再重试", e);
+            List<Long[]> halfCutTimes = Tools.cutTimeRange(startTime, endTime, HALF_HOUR);
+            futures.clear();
+            res.clear();
+            try {
+                halfCutTimes.forEach(times -> futures.add(CompletableFuture.supplyAsync(() -> {
+                    try {
+                        return searchRecord(times[0], times[1]);
+                    } catch (IOException e2) {
+                        throw new RuntimeException(e2);
+                    }
+                }, executor)));
+                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).whenComplete((r2, e2) -> futures.forEach(f -> res.add(f.join()))).join();
+                log.error("跨度半小时重试成功");
+            }catch (Exception exception){
+                log.error("error-searchRecordRecentDay-oneHour,跨度半小时重试失败", exception);
+            }
+        }
         return res;
     }
 
@@ -170,28 +194,24 @@ public class EsClientDao {
         return retryTemplate.execute(context -> searchScroll(sourceBuilder));
     }
 
-    private Pair<Long[], Map<ChannelIndex, ChannelIndex.Record>> searchRecord(long startTime, long endTime) {
+    private Pair<Long[], Map<ChannelIndex, ChannelIndex.Record>> searchRecord(long startTime, long endTime) throws IOException {
         Map<ChannelIndex, ChannelIndex.Record> res = new HashMap<>();
-        try {
-            QueryBuilder queryBuilder = QueryBuilders.rangeQuery("mtime").gte(startTime).lt(endTime);
-            List<JSONObject> results = searchScroll(queryBuilder, 10000, CHANNEL_RECORD_FETCH_SOURCE);
-            for (Map<String, Object> result : results) {
-                for (ChannelIndex channelIndex : ChannelIndex.createChannelIndexes(result)) {
-                    res.compute(channelIndex, (k, v) -> {
-                        if (null == v) {
-                            v = new ChannelIndex.Record();
-                        }
-                        try {
-                            return v.mergeRecord(new ChannelIndex.Record(result));
-                        } catch (Exception e) {
-                            log.error("searchRecord-error-id:{}", result.get("id"), e);
-                            return null;
-                        }
-                    });
-                }
+        QueryBuilder queryBuilder = QueryBuilders.rangeQuery("mtime").gte(startTime).lt(endTime);
+        List<JSONObject> results = searchScroll(queryBuilder, 10000, CHANNEL_RECORD_FETCH_SOURCE);
+        for (Map<String, Object> result : results) {
+            for (ChannelIndex channelIndex : ChannelIndex.createChannelIndexes(result)) {
+                res.compute(channelIndex, (k, v) -> {
+                    if (null == v) {
+                        v = new ChannelIndex.Record();
+                    }
+                    try {
+                        return v.mergeRecord(new ChannelIndex.Record(result));
+                    } catch (Exception e) {
+                        log.error("searchRecord-error-id:{}", result.get("id"), e);
+                        return null;
+                    }
+                });
             }
-        } catch (IOException e) {
-            log.error("searchRecord-", e);
         }
         log.info("startTime:{},endTime:{},size:{}", DF.format(startTime), DF.format(endTime), res.size());
         return Pair.of(new Long[]{startTime, endTime}, res);