[update]修改pom文件

.idea/workspace.xml

@@ -6,7 +6,12 @@
   <component name="ChangeListManager">
     <list default="true" id="3bf79822-fe13-41ea-bc8f-858fbea129e7" name="Changes" comment="[update]修改pom文件">
       <change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/tianao-analysis/target/classes/application.yml" beforeDir="false" afterPath="$PROJECT_DIR$/tianao-analysis/target/classes/application.yml" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/es-track-common/pom.xml" beforeDir="false" afterPath="$PROJECT_DIR$/es-track-common/pom.xml" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/es-track-common/src/main/java/cn/com/taiji/config/EsClientConfiguration.java" beforeDir="false" afterPath="$PROJECT_DIR$/es-track-common/src/main/java/cn/com/taiji/config/EsClientConfiguration.java" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/fusion-analysis/pom.xml" beforeDir="false" afterPath="$PROJECT_DIR$/fusion-analysis/pom.xml" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/fusion-analysis/src/main/java/cn/com/taiji/domain/FusionQuery.java" beforeDir="false" afterPath="$PROJECT_DIR$/fusion-analysis/src/main/java/cn/com/taiji/domain/FusionQuery.java" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/fusion-analysis/src/main/java/cn/com/taiji/service/impl/FusionAnalysisServiceImpl.java" beforeDir="false" afterPath="$PROJECT_DIR$/fusion-analysis/src/main/java/cn/com/taiji/service/impl/FusionAnalysisServiceImpl.java" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/fusion-analysis/src/main/resources/application.yml" beforeDir="false" afterPath="$PROJECT_DIR$/fusion-analysis/src/main/resources/application.yml" afterDir="false" />
     </list>
     <option name="SHOW_DIALOG" value="false" />
     <option name="HIGHLIGHT_CONFLICTS" value="true" />
@@ -226,6 +231,22 @@
     <MESSAGE value="[update]修改pom文件" />
     <option name="LAST_COMMIT_MESSAGE" value="[update]修改pom文件" />
   </component>
+  <component name="XDebuggerManager">
+    <breakpoint-manager>
+      <breakpoints>
+        <line-breakpoint enabled="true" type="java-line">
+          <url>file://$PROJECT_DIR$/fusion-analysis/src/main/java/cn/com/taiji/service/impl/FusionAnalysisServiceImpl.java</url>
+          <line>197</line>
+          <option name="timeStamp" value="3" />
+        </line-breakpoint>
+      </breakpoints>
+    </breakpoint-manager>
+    <pin-to-top-manager>
+      <pinned-members>
+        <PinnedItemInfo parentTag="org.elasticsearch.search.aggregations.Aggregations" memberName="aggregations" />
+      </pinned-members>
+    </pin-to-top-manager>
+  </component>
   <component name="XSLT-Support.FileAssociations.UIState">
     <expand />
     <select />

es-track-common/pom.xml

@@ -14,7 +14,7 @@
     <properties>
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
-        <elasticsearch.version>7.8.0</elasticsearch.version>
+        <elasticsearch.version>7.16.2</elasticsearch.version>
     </properties>
 
 

es-track-common/src/main/java/cn/com/taiji/config/EsClientConfiguration.java

@@ -6,6 +6,7 @@ import org.apache.http.auth.UsernamePasswordCredentials;
 import org.apache.http.client.CredentialsProvider;
 import org.apache.http.impl.client.BasicCredentialsProvider;
 import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
@@ -47,6 +48,13 @@ public class EsClientConfiguration {
     @Bean
     public RestHighLevelClient client() {
 
+        RestClientBuilder.RequestConfigCallback requestConfigCallback = requstConfigBuilder -> {
+            requstConfigBuilder.setConnectTimeout(300000);
+            requstConfigBuilder.setSocketTimeout(400000);
+            requstConfigBuilder.setConnectionRequestTimeout(0);
+            return requstConfigBuilder;
+        };
+
         HttpHost[] httpHosts = new HttpHost[esUris.length];
         //将地址转换为http主机数组，未配置端口则采用默认9200端口，配置了端口则用配置的端口
         for (int i = 0; i < httpHosts.length; i++) {

fusion-analysis/pom.xml

@@ -14,7 +14,7 @@
     <properties>
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
-        <elasticsearch.version>7.8.0</elasticsearch.version>
+        <elasticsearch.version>7.16.2</elasticsearch.version>
     </properties>
 
     <dependencies>

fusion-analysis/src/main/java/cn/com/taiji/domain/FusionQuery.java

@@ -12,6 +12,8 @@ public class FusionQuery {
 
     private String endTime;
 
+    private Integer pageSize;
+
     public String getStartTime() {
         return startTime;
     }
@@ -27,4 +29,12 @@ public class FusionQuery {
     public void setEndTime(String endTime) {
         this.endTime = endTime;
     }
+
+    public Integer getPageSize() {
+        return pageSize;
+    }
+
+    public void setPageSize(Integer pageSize) {
+        this.pageSize = pageSize;
+    }
 }

fusion-analysis/src/main/java/cn/com/taiji/service/impl/FusionAnalysisServiceImpl.java

@@ -7,13 +7,18 @@ import cn.com.taiji.domain.StaticOrDynamicData;
 import cn.com.taiji.domain.dto.LeaceOutDto;
 import cn.com.taiji.service.FusionAnalysisService;
 import cn.com.taiji.utils.DateUtils;
+import cn.hutool.core.io.LineHandler;
 import com.alibaba.excel.EasyExcel;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.http.client.config.RequestConfig;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
 import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.SearchHit;
@@ -34,6 +39,7 @@ import org.springframework.stereotype.Service;
 import javax.servlet.http.HttpServletResponse;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.net.URLEncoder;
 import java.text.ParseException;
 import java.util.ArrayList;
@@ -51,6 +57,21 @@ import static cn.com.taiji.constants.EsIndexConstants.*;
 @Service
 public class FusionAnalysisServiceImpl implements FusionAnalysisService {
 
+    private static final RequestOptions COMMON_OPTIONS;
+
+    static {
+        RequestConfig requestConfig = RequestConfig.custom()
+                .setConnectTimeout(5000)
+                .setSocketTimeout(60000)
+                .build();
+        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
+        builder.setRequestConfig(requestConfig).setHttpAsyncResponseConsumerFactory(
+                // 设置查询内容大小限制,默认100 * 1024 * 1024
+                new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(5*1024 * 1024 * 1024)
+        );
+        COMMON_OPTIONS = builder.build();
+    }
+
     @Autowired
     private RestHighLevelClient client;
 
@@ -67,7 +88,7 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
         searchSourceBuilder.query(boolQueryBuilder);
         TermsAggregationBuilder aggBuilder = AggregationBuilders.terms("ship_field_agg")
                 .field("mergeTarget")
-                .size(1000000)
+                .size(query.getPageSize())
                 .subAggregation(
                         AggregationBuilders.topHits("ship_field_agg_top1")
                                 .size(100)
@@ -78,9 +99,12 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
         searchSourceBuilder
                 .trackTotalHits(true)
                 .aggregation(aggBuilder)
-                .size(1000000);
+                .size(query.getPageSize())
+                .timeout(TimeValue.timeValueHours(1L))
+                .timeout(TimeValue.timeValueMinutes(30L))
+                .timeout(TimeValue.timeValueSeconds(500L));
         request.source(searchSourceBuilder);
-        SearchResponse search = client.search(request, RequestOptions.DEFAULT);
+        SearchResponse search = client.search(request, COMMON_OPTIONS);
         Aggregations aggregations = search.getAggregations();
         ParsedStringTerms aggregation = aggregations.get("ship_field_agg");
         for (Terms.Bucket bucket : aggregation.getBuckets()) {
@@ -89,12 +113,6 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
             SearchHits searchHits = topHits.getHits();
             Long value = searchHits.getTotalHits().value;
             SearchHit[] hits = searchHits.getHits();
-            JSONObject sourceItem = JSONObject.parseObject(hits[0].getSourceAsString());
-            JSONObject sourceItem2 = JSONObject.parseObject(hits[value.intValue()-1].getSourceAsString());
-            data.setMergeTarget(sourceItem.getString("mergeTarget"));
-            data.setTrackCount(value);
-            data.setStartTime(sourceItem.getString("mergeTime"));
-            data.setEndTime(sourceItem2.getString("mergeTime"));
             Integer num = 0;
             for (int i = 0; i < hits.length; i++) {
                 if (i != hits.length-1){
@@ -108,8 +126,16 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
                     }
                 }
             }
-            data.setFluctuateNumber(num);
-            list.add(data);
+            if (num>0){
+                JSONObject sourceItem = JSONObject.parseObject(hits[0].getSourceAsString());
+                JSONObject sourceItem2 = JSONObject.parseObject(hits[value.intValue()-1].getSourceAsString());
+                data.setMergeTarget(sourceItem.getString("mergeTarget"));
+                data.setTrackCount(value);
+                data.setStartTime(sourceItem.getString("mergeTime"));
+                data.setEndTime(sourceItem2.getString("mergeTime"));
+                data.setFluctuateNumber(num);
+                list.add(data);
+            }
         }
         try {
             response.setContentType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet");
@@ -126,6 +152,7 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
 
     @Override
     public void fusionAnalysisExcelTwo(FusionQuery query, HttpServletResponse response) throws IOException, ParseException {
+
         List<FusionData> list = new ArrayList<>();
         Date startTime = DateUtils.parse(query.getStartTime(),"yyyy-MM-dd");
         String time = DateUtils.format(startTime,"yyyy-MM-dd");
@@ -137,15 +164,15 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
         searchSourceBuilder.query(boolQueryBuilder);
         TermsAggregationBuilder aggBuilder = AggregationBuilders.terms("ship_field_agg")
                 .field("mergeTarget")
-                .size(1000000)
+                .size(query.getPageSize())
                 .subAggregation(
                         AggregationBuilders.terms("ship_field_agg_one")
                                 .field("targetCourse")
-                                .size(1000000)
+                                .size(query.getPageSize())
                                 .subAggregation(
                                         AggregationBuilders.terms("ship_field_agg_two")
                                                 .field("targetSpeed")
-                                                .size(1000000)
+                                                .size(query.getPageSize())
                                                 .subAggregation(AggregationBuilders.topHits("ship_field_agg_top1")
                                                         .size(100)
                                                         .sort(
@@ -156,9 +183,12 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
         searchSourceBuilder
                 .trackTotalHits(true)
                 .aggregation(aggBuilder)
-                .size(1000000);
+                .size(query.getPageSize())
+                .timeout(TimeValue.timeValueHours(1L))
+                .timeout(TimeValue.timeValueMinutes(30L))
+                .timeout(TimeValue.timeValueSeconds(500L));
         request.source(searchSourceBuilder);
-        SearchResponse search = client.search(request, RequestOptions.DEFAULT);
+        SearchResponse search = client.search(request, COMMON_OPTIONS);
         Aggregations aggregations = search.getAggregations();
         ParsedStringTerms aggregation = aggregations.get("ship_field_agg");
         for (Terms.Bucket bucket : aggregation.getBuckets()) {
@@ -174,6 +204,11 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
                     for (int i = 0; i < hit.length; i++) {
                         if (num>2){
                             data.setFluctuateNumber(data.getFluctuateNumber()+1);
+                            data.setMergeTarget(bucket.getKeyAsString());
+                            data.setTrackCount(bucket.getDocCount());
+                            data.setStartTime(query.getStartTime());
+                            data.setEndTime(query.getEndTime());
+                            list.add(data);
                             num =0;
                         }
                         if (i != hit.length-1){
@@ -189,11 +224,6 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
                     }
                 }
             }
-            data.setMergeTarget(bucket.getKeyAsString());
-            data.setTrackCount(Long.valueOf(bucket.getDocCount()));
-            data.setStartTime(query.getStartTime());
-            data.setEndTime(query.getEndTime());
-            list.add(data);
         }
         try {
             response.setContentType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet");
@@ -222,19 +252,22 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
         searchSourceBuilder.query(boolQueryBuilder);
         TermsAggregationBuilder aggBuilder = AggregationBuilders.terms("ship_field_agg")
                 .field("fusionBatchNum")
-                .size(1000000);
+                .size(query.getPageSize());
         searchSourceBuilder
                 .trackTotalHits(true)
                 .aggregation(aggBuilder)
-                .size(1000000);
+                .size(query.getPageSize())
+                .timeout(TimeValue.timeValueHours(1L))
+                .timeout(TimeValue.timeValueMinutes(30L))
+                .timeout(TimeValue.timeValueSeconds(500L));
         request.source(searchSourceBuilder);
-        SearchResponse search = client.search(request, RequestOptions.DEFAULT);
+        SearchResponse search = client.search(request, COMMON_OPTIONS);
         Aggregations aggregations = search.getAggregations();
         ParsedStringTerms aggregation = aggregations.get("ship_field_agg");
         for (Terms.Bucket bucket : aggregation.getBuckets()) {
             LeaceOutDto dto = new LeaceOutDto();
             dto.setBatchNumber(bucket.getKeyAsString());
-            dto.setCount(Long.valueOf(bucket.getDocCount()));
+            dto.setCount(bucket.getDocCount());
             dtos.add(dto);
         }
         dtos.forEach(item->{
@@ -248,10 +281,13 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
             boolQueryBuilder2.filter(QueryBuilders.termsQuery("mergeType","RADAR","AIS_RADAR","BEIDOU_RADAR","AIS_BEIDOU_RADAR"));
             searchSourceBuilder2.query(boolQueryBuilder2)
                     .trackTotalHits(true)
-                    .size(1000000);
+                    .size(query.getPageSize())
+                    .timeout(TimeValue.timeValueHours(1L))
+                    .timeout(TimeValue.timeValueMinutes(30L))
+                    .timeout(TimeValue.timeValueSeconds(500L));
             request2.source(searchSourceBuilder2);
             try {
-                SearchResponse search2 = client.search(request, RequestOptions.DEFAULT);
+                SearchResponse search2 = client.search(request, COMMON_OPTIONS);
                 SearchHits searchHits = search2.getHits();
                 Long totalHits = searchHits.getTotalHits().value;
                 if (totalHits<5) {
@@ -298,13 +334,16 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
         searchSourceBuilder.query(boolQueryBuilder);
         TermsAggregationBuilder aggBuilder = AggregationBuilders.terms("ship_field_agg")
                 .field("devideNo")
-                .size(1000000);
+                .size(query.getPageSize());
         searchSourceBuilder
                 .trackTotalHits(true)
                 .aggregation(aggBuilder)
-                .size(1000000);
+                .size(query.getPageSize())
+                .timeout(TimeValue.timeValueHours(1L))
+                .timeout(TimeValue.timeValueMinutes(30L))
+                .timeout(TimeValue.timeValueSeconds(500L));
         request.source(searchSourceBuilder);
-        SearchResponse search = client.search(request, RequestOptions.DEFAULT);
+        SearchResponse search = client.search(request, COMMON_OPTIONS);
         Aggregations aggregations = search.getAggregations();
         ParsedStringTerms aggregation = aggregations.get("ship_field_agg");
         for (Terms.Bucket bucket : aggregation.getBuckets()) {
@@ -324,10 +363,13 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
             boolQueryBuilder2.filter(QueryBuilders.termsQuery("mergeType","BEIDOU","AIS_BEIDOU","BEIDOU_RADAR","AIS_BEIDOU_RADAR"));
             searchSourceBuilder2.query(boolQueryBuilder2)
                     .trackTotalHits(true)
-                    .size(1000000);
+                    .size(query.getPageSize())
+                    .timeout(TimeValue.timeValueHours(1L))
+                    .timeout(TimeValue.timeValueMinutes(30L))
+                    .timeout(TimeValue.timeValueSeconds(500L));
             request2.source(searchSourceBuilder2);
             try {
-                SearchResponse search2 = client.search(request, RequestOptions.DEFAULT);
+                SearchResponse search2 = client.search(request, COMMON_OPTIONS);
                 SearchHits searchHits = search2.getHits();
                 Long totalHits = searchHits.getTotalHits().value;
                 if (totalHits<5) {
@@ -374,13 +416,16 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
         searchSourceBuilder.query(boolQueryBuilder);
         TermsAggregationBuilder aggBuilder = AggregationBuilders.terms("ship_field_agg")
                 .field("targetID")
-                .size(1000000);
+                .size(query.getPageSize());
         searchSourceBuilder
                 .trackTotalHits(true)
                 .aggregation(aggBuilder)
-                .size(1000000);
+                .size(query.getPageSize())
+                .timeout(TimeValue.timeValueHours(1L))
+                .timeout(TimeValue.timeValueMinutes(30L))
+                .timeout(TimeValue.timeValueSeconds(500L));
         request.source(searchSourceBuilder);
-        SearchResponse search = client.search(request, RequestOptions.DEFAULT);
+        SearchResponse search = client.search(request, COMMON_OPTIONS);
         Aggregations aggregations = search.getAggregations();
         ParsedStringTerms aggregation = aggregations.get("ship_field_agg");
         for (Terms.Bucket bucket : aggregation.getBuckets()) {
@@ -400,10 +445,13 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
             boolQueryBuilder2.filter(QueryBuilders.termsQuery("mergeType","AIS","RADAR","AIS_BEIDOU","AIS_RADAR","BEIDOU_RADAR","AIS_BEIDOU_RADAR"));
             searchSourceBuilder2.query(boolQueryBuilder2)
                     .trackTotalHits(true)
-                    .size(1000000);
+                    .size(query.getPageSize())
+                    .timeout(TimeValue.timeValueHours(1L))
+                    .timeout(TimeValue.timeValueMinutes(30L))
+                    .timeout(TimeValue.timeValueSeconds(500L));
             request2.source(searchSourceBuilder2);
             try {
-                SearchResponse search2 = client.search(request, RequestOptions.DEFAULT);
+                SearchResponse search2 = client.search(request, COMMON_OPTIONS);
                 SearchHits searchHits = search2.getHits();
                 Long totalHits = searchHits.getTotalHits().value;
                 if (totalHits<5) {
@@ -451,13 +499,16 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
         searchSourceBuilder.query(boolQueryBuilder);
         TermsAggregationBuilder aggBuilder = AggregationBuilders.terms("ship_field_agg")
                 .field("fusionBatchNum")
-                .size(1000000);
+                .size(query.getPageSize());
         searchSourceBuilder
                 .trackTotalHits(true)
                 .aggregation(aggBuilder)
-                .size(1000000);
+                .size(query.getPageSize())
+                .timeout(TimeValue.timeValueHours(1L))
+                .timeout(TimeValue.timeValueMinutes(30L))
+                .timeout(TimeValue.timeValueSeconds(500L));
         request.source(searchSourceBuilder);
-        SearchResponse search = client.search(request, RequestOptions.DEFAULT);
+        SearchResponse search = client.search(request, COMMON_OPTIONS);
         Aggregations aggregations = search.getAggregations();
         ParsedStringTerms aggregation = aggregations.get("ship_field_agg");
         for (Terms.Bucket bucket : aggregation.getBuckets()) {
@@ -478,10 +529,13 @@ public class FusionAnalysisServiceImpl implements FusionAnalysisService {
             boolQueryBuilder2.filter(QueryBuilders.termsQuery("mergeType","RADAR","AIS_RADAR","BEIDOU_RADAR","AIS_BEIDOU_RADAR"));
             searchSourceBuilder2.query(boolQueryBuilder2)
                     .trackTotalHits(true)
-                    .size(1000000);
+                    .size(query.getPageSize())
+                    .timeout(TimeValue.timeValueHours(1L))
+                    .timeout(TimeValue.timeValueMinutes(30L))
+                    .timeout(TimeValue.timeValueSeconds(500L));
             request2.source(searchSourceBuilder2);
             try {
-                SearchResponse search2 = client.search(request, RequestOptions.DEFAULT);
+                SearchResponse search2 = client.search(request, COMMON_OPTIONS);
                 SearchHits searchHits = search2.getHits();
                 SearchHit[] hits = searchHits.getHits();
                 if (hits.length > 0) {

fusion-analysis/src/main/resources/application.yml

@@ -3,7 +3,7 @@ server:
 
 spring:
   application.name: fusion-analysis
-
+#  profiles.active: prod
 
 taiji:
   elasticsearch.rest: