add: all

pom.xml

@@ -190,6 +190,11 @@
             <artifactId>spring-kafka</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-crypto</artifactId>
+            <version>5.5.8</version>
+        </dependency>
 
     </dependencies>
 
@@ -201,5 +206,4 @@
             </plugin>
         </plugins>
     </build>
-
 </project>

src/main/java/com/winhc/phoenix/example/configuration/ElasticSearchConfiguration.java

@@ -31,11 +31,11 @@ public class ElasticSearchConfiguration {
 
     @Value("${es.schema:http}")
     String schema;
-    @Value(value = "${es.connect-timeout:1000}")
+    @Value(value = "${es.connect-timeout:100000}")
     String connectTimeout;
     @Value(value = "${es.socket-timeout:600000}")
     String socketTimeout;
-    @Value(value = "${es.connection-request-timeout:500}")
+    @Value(value = "${es.connection-request-timeout:50000}")
     String connectionRequestTimeout;
     @Value(value = "${es.max-conn-total:100}")
     String maxConnTotal;

src/main/java/com/winhc/phoenix/example/configuration/HbaseConfiguration.java

@@ -1,6 +1,7 @@
 package com.winhc.phoenix.example.configuration;
 
 import lombok.AllArgsConstructor;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.client.Connection;
@@ -9,7 +10,6 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import java.io.IOException;
 import java.util.Map;
 import java.util.Set;
 
@@ -26,8 +26,9 @@ public class HbaseConfiguration {
     private final HbaseProperties properties;
 
 
+    @SneakyThrows
     @Bean
-    public Connection connection() throws IOException {
+    public Connection connection() {
         return ConnectionFactory.createConnection(configuration());
     }
 

src/main/java/com/winhc/phoenix/example/controller/FastQueryController.java

@@ -0,0 +1,36 @@
+package com.winhc.phoenix.example.controller;
+
+import com.winhc.phoenix.example.service.HbaseQueryService;
+import com.winhc.phoenix.example.vo.HbaseQueryParamVo;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.*;
+
+/**
+ * @author: XuJiakai
+ * 2021/1/26 10:23
+ */
+@Slf4j
+@RestController
+@AllArgsConstructor
+@RequestMapping("fast")
+@Api(tags = "可高并发的查询接口", value = "fast")
+public class FastQueryController {
+    private final HbaseQueryService hbaseQueryService;
+
+
+    @ApiOperation(value = "批量获取人名")
+    @PostMapping("hbase/bulk-get/human")
+    public Object bulkGet(@RequestBody String[] rowkey) {
+        return hbaseQueryService.humanBulkGet(rowkey);
+    }
+
+    @ApiOperation(value = "批量查询hbase，并可指定字段")
+    @PostMapping("hbase/bulk-get")
+    public Object fastGetHbase(@RequestParam String tableName, @RequestBody HbaseQueryParamVo hbaseQueryParamVo) {
+        return hbaseQueryService.fastGetHbase(tableName, hbaseQueryParamVo);
+    }
+
+}

src/main/java/com/winhc/phoenix/example/controller/SearchController.java

@@ -1,6 +1,5 @@
 package com.winhc.phoenix.example.controller;
 
-import com.aliyun.odps.utils.StringUtils;
 import com.winhc.phoenix.example.aspect.Timer;
 import com.winhc.phoenix.example.enums.EsVersion;
 import com.winhc.phoenix.example.service.SearchService;
@@ -8,6 +7,7 @@ import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
@@ -54,6 +54,7 @@ public class SearchController {
     private static final Pattern pattern = Pattern.compile("[^\\u4e00-\\u9fa50-9a-zA-Z]");
 
     private static String cleanup(String val) {
+//        return val;
         return StringUtils.isNotBlank(val) ? pattern.matcher(val).replaceAll("") : "";
     }
 }

src/main/java/com/winhc/phoenix/example/dao/SearchDao.java

@@ -2,8 +2,11 @@ package com.winhc.phoenix.example.dao;
 
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
+import org.elasticsearch.search.rescore.RescoreBuilder;
 import org.elasticsearch.search.sort.SortBuilder;
 
+import java.util.List;
+
 /**
  * @author: XuJiakai
  * 2020/11/20 15:15
@@ -11,4 +14,9 @@ import org.elasticsearch.search.sort.SortBuilder;
 public interface SearchDao {
 
     Object search(String index, String type, QueryBuilder query, SortBuilder sortBuilder, FetchSourceContext fetchSourceContext, int from, int size);
+
+    Object search(String index, String type, QueryBuilder query, RescoreBuilder rescoreBuilder, SortBuilder sortBuilder, FetchSourceContext fetchSourceContext, int from, int size);
+
+    boolean deleteByIds(String index, String type, List<String> ids);
+
 }

src/main/java/com/winhc/phoenix/example/dao/impl/SearchDaoImpl.java

@@ -7,6 +7,8 @@ import lombok.AllArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.http.util.EntityUtils;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.Response;
@@ -16,10 +18,12 @@ import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
+import org.elasticsearch.search.rescore.RescoreBuilder;
 import org.elasticsearch.search.sort.SortBuilder;
 import org.springframework.stereotype.Repository;
 
 import java.util.HashMap;
+import java.util.List;
 
 /**
  * @author: XuJiakai
@@ -40,17 +44,30 @@ public class SearchDaoImpl implements SearchDao {
     @SneakyThrows
     @Override
     public Object search(String index, String type, QueryBuilder query, SortBuilder sortBuilder, FetchSourceContext fetchSourceContext, int from, int size) {
+        return search(index, type, query, null, sortBuilder, fetchSourceContext, from, size);
+    }
+
+    @SneakyThrows
+    @Override
+    public Object search(String index, String type, QueryBuilder query, RescoreBuilder rescoreBuilder, SortBuilder sortBuilder, FetchSourceContext fetchSourceContext, int from, int size) {
         SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
                 .query(query)
                 .from(from)
                 .size(size)
                 .highlighter(new HighlightBuilder().field("*").preTags("<font color='red'>").postTags("</font>"));
 
+        if (rescoreBuilder != null) {
+            searchSourceBuilder.addRescorer(rescoreBuilder)
+//                    .addRescorer(new QueryRescorerBuilder(query))
+            ;
+        }
         if (fetchSourceContext != null) {
             searchSourceBuilder.fetchSource(fetchSourceContext);
         }
         if (sortBuilder != null) {
-            searchSourceBuilder.sort(sortBuilder).trackScores(true);
+            searchSourceBuilder.sort(sortBuilder)
+//                    .trackScores(true)
+            ;
         }
 
         SearchRequest searchRequest = new SearchRequest()
@@ -62,6 +79,18 @@ public class SearchDaoImpl implements SearchDao {
         return mapper.readValue(search.toString(), typeRef);
     }
 
+    @SneakyThrows
+    @Override
+    public boolean deleteByIds(String index, String type, List<String> ids) {
+        BulkRequest bulkRequest = new BulkRequest();
+        for (String id : ids) {
+            DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
+            bulkRequest.add(deleteRequest);
+        }
+        restHighLevelClient.bulk(bulkRequest);
+        return true;
+    }
+
 
     private final RestClient restClient;
 

src/main/java/com/winhc/phoenix/example/enums/EsVersion.java

@@ -15,6 +15,7 @@ import lombok.Getter;
 public enum EsVersion {
     v7版(SearchV7ServiceImpl.index),
     v8精简版(SearchV8SimpServiceImpl.index),
-    v8版(SearchV8ServiceImpl.index);
+    v8版(SearchV8ServiceImpl.index),
+    v8版_Fast("v8_fast");
     private final String value;
 }

src/main/java/com/winhc/phoenix/example/framework/es/EsFastScan.java

@@ -1,13 +1,17 @@
 package com.winhc.phoenix.example.framework.es;
 
+import com.alibaba.fastjson.JSONObject;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.winhc.phoenix.example.util.QueryBuilderUtils;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchScrollRequest;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.query.WrapperQueryBuilder;
 import org.elasticsearch.search.Scroll;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
@@ -30,14 +34,20 @@ public class EsFastScan {
     private Consumer<SearchHit[]> func;
     private String index;
     private String type;
+    private String dsl;
     private int batchSize = 1000;
     private int thread_num = 1;
 
     public EsFastScan(RestHighLevelClient client, Consumer<SearchHit[]> func, String index, String type) {
+        this(client, func, index, type, null);
+    }
+
+    public EsFastScan(RestHighLevelClient client, Consumer<SearchHit[]> func, String index, String type, String dsl) {
         this.client = client;
         this.func = func;
         this.index = index;
         this.type = type;
+        this.dsl = dsl;
     }
 
 
@@ -68,7 +78,13 @@ public class EsFastScan {
 
         SearchRequest searchRequest = new SearchRequest(index).types(type).scroll(scroll);
 
-        searchRequest.source(new SearchSourceBuilder().size(batchSize));
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().size(batchSize);
+        if (StringUtils.isNotEmpty(dsl)) {
+            WrapperQueryBuilder query = QueryBuilderUtils.getQuery(JSONObject.parseObject(dsl));
+            searchSourceBuilder.query(query);
+        }
+
+        searchRequest.source(searchSourceBuilder);
         SearchResponse searchResponse = client.search(searchRequest);
 
         long totalHits = searchResponse.getHits().getTotalHits();

src/main/java/com/winhc/phoenix/example/framework/hbase/HbaseFastScan.java

@@ -9,6 +9,7 @@ import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.*;
 import org.apache.hadoop.hbase.util.Bytes;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -17,6 +18,7 @@ import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 
 /**
  * @author: XuJiakai
@@ -27,13 +29,14 @@ public class HbaseFastScan {
     private int thread_num = 1;
     private int batchSize = 500;
 
-    private Connection connection;
+    //    private Connection connection;
     private TableName tableName;
 
-    private Consumer<List<Map<String, Object>>> func;
+    private Consumer<List<Map<String, String>>> func;
+    private Supplier<Connection> supplier;
 
-    public HbaseFastScan(Connection connection, String tableName, Consumer<List<Map<String, Object>>> func) {
-        this.connection = connection;
+    public HbaseFastScan(Supplier<Connection> supplier, String tableName, Consumer<List<Map<String, String>>> func) {
+        this.supplier = supplier;
         this.tableName = TableName.valueOf(tableName);
         this.func = func;
     }
@@ -53,20 +56,22 @@ public class HbaseFastScan {
         private byte[] endKey;
         private String regionName;
         private String tn;
+        private Connection connection;
 
-        public Task(RegionInfo regionInfo) {
+        public Task(Supplier<Connection> supplier, RegionInfo regionInfo) {
             this.regionName = regionInfo.getRegionNameAsString();
             this.startKey = regionInfo.getStartKey();
             this.endKey = regionInfo.getEndKey();
             this.tn = regionName + "-" + getKeyAsString(startKey) + "_" + getKeyAsString(endKey);
             log.info("{}：{}  {}", regionName, getKeyAsString(startKey), getKeyAsString(endKey));
             setName(tn);
+            this.connection = supplier.get();
         }
 
         @Override
         public void run() {
             log.info("start。。。");
-            List<Map<String, Object>> list = new ArrayList<>();
+            List<Map<String, String>> list = new ArrayList<>();
             try (Table table = connection.getTable(tableName)) {
                 Scan scan = new Scan();
                 if (startKey.length != 0) {
@@ -79,7 +84,7 @@ public class HbaseFastScan {
 
                 ResultScanner scanner = table.getScanner(scan);
                 for (Result result : scanner) {
-                    Map<String, Object> columnMap = HbaseResultUtils.parseResult(result);
+                    Map<String, String> columnMap = HbaseResultUtils.parseResult(result);
                     list.add(columnMap);
                     if (list.size() >= batchSize) {
                         func.accept(list);
@@ -94,19 +99,27 @@ public class HbaseFastScan {
                 log.info("{} 执行完成！", tn);
             } catch (Exception e) {
                 log.error(e.getMessage(), e);
+            } finally {
+
+                try {
+                    connection.close();
+                } catch (IOException e) {
+                    log.error(e.getMessage(), e);
+                    connection = null;
+                }
             }
         }
 
-
     }
 
     @SneakyThrows
     public void scan() {
+        Connection connection = supplier.get();
         try (RegionLocator regionLocator = connection.getRegionLocator(tableName);) {
 
             List<HRegionLocation> allRegionLocations = regionLocator.getAllRegionLocations();
 
-            int poolSize = 10;
+            int poolSize = thread_num;
             ArrayBlockingQueue<Runnable> objects = new ArrayBlockingQueue<>(poolSize);
             ThreadPoolExecutor executorService = new ThreadPoolExecutor(
                     poolSize, poolSize,
@@ -127,10 +140,8 @@ public class HbaseFastScan {
                 log.info("region host: {}", hRegionLocation.getHostname());
                 RegionInfo regionInfo = hRegionLocation.getRegion();
 
-                String regionName = regionInfo.getRegionNameAsString();
                 byte[] startKey = regionInfo.getStartKey();
                 byte[] endKey = regionInfo.getEndKey();
-                String tn = regionName + "-" + getKeyAsString(startKey) + "_" + getKeyAsString(endKey);
                 Scan scan = new Scan();
                 if (startKey.length != 0) {
                     scan.withStartRow(startKey);
@@ -138,37 +149,10 @@ public class HbaseFastScan {
                 if (endKey.length != 0) {
                     scan.withStopRow(endKey);
                 }
-                new Task(regionInfo).start();
-
-//            executorService.submit(new Task(regionInfo));
-          /*  executorService.submit(() -> {
-                log.info("start。。。");
-                List<Map<String, Object>> list = new ArrayList<>();
-                try (Table table = connection.getTable(tableName)) {
-
-                    scan.setCaching(batchSize);
-
-                    ResultScanner scanner = table.getScanner(scan);
-                    for (Result result : scanner) {
-                        Map<String, Object> columnMap = HbaseResultUtils.parseResult(result);
-                        list.add(columnMap);
-                        if (list.size() >= batchSize) {
-                            func.accept(list);
-                            list.clear();
-                            log.info("current rowkey: {}", columnMap.getOrDefault("rowkey", null));
-                        }
-                    }
-                    if (!list.isEmpty()) {
-                        func.accept(list);
-                        list.clear();
-                    }
-                    log.info("{} 执行完成！", tn);
-                } catch (Exception e) {
-                    log.error(e.getMessage(), e);
-                }
-            });*/
-
+                executorService.submit(new Task(supplier, regionInfo));
             }
+        } finally {
+            connection.close();
         }
     }
 

src/main/java/com/winhc/phoenix/example/job/BatchQueryEsTest.java

@@ -0,0 +1,151 @@
+package com.winhc.phoenix.example.job;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.mongodb.client.MongoCollection;
+import lombok.AllArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.bson.Document;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.aggregations.AggregationBuilders;
+import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+/**
+ * @author: XuJiakai
+ * 2021/3/1 09:43
+ */
+@Slf4j
+@Component
+@AllArgsConstructor
+public class BatchQueryEsTest {
+
+    private ObjectMapper mapper;
+    private RestHighLevelClient restHighLevelClient;
+    private final MongoTemplate mongoTemplate;
+
+    private static final TypeReference<HashMap<String, Object>> typeRef
+            = new TypeReference<HashMap<String, Object>>() {
+    };
+
+    public void start(int batchSize, int round) {
+        MongoCollection<Document> company_id = mongoTemplate.getCollection("xjk_test_es_company_id");
+        Set<String> set = new HashSet<String>();
+
+        for (Document doc : company_id.find().skip(batchSize * (round - 1)).batchSize(batchSize)) {
+            String id = doc.getString("_id");
+            set.add(id);
+
+            if (set.size() >= batchSize) {
+                break;
+            }
+        }
+
+        Object query = query2(set);
+        System.out.println();
+        System.out.println(query);
+        System.out.println();
+    }
+
+
+    @SneakyThrows
+    private Object query2(Set<String> ids) {
+        log.info("start... size: {}", ids.size());
+
+        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
+
+        boolQuery.filter(addTerms(ids, "_id"));
+
+        List<SumAggregationBuilder> list = Arrays.asList(
+                AggregationBuilders.sum("open_announcement_1").field("summary.company_court_open_announcement_deleted_0_defendant")
+//                , AggregationBuilders.sum("open_announcement_2").field("summary.company_court_open_announcement_deleted_0_plaintiff")
+//                ,
+//                AggregationBuilders.sum("3").field("summary.company_app_info_del_1")
+//                , AggregationBuilders.sum("4").field("summary.company_staff_del_1")
+//                , AggregationBuilders.sum("5").field("summary.company_icp_del_0")
+//                , AggregationBuilders.sum("6").field("summary.company_tm_del_1")
+        );
+
+
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
+                .query(boolQuery)
+                .size(0);
+
+        for (SumAggregationBuilder builder : list) {
+            searchSourceBuilder.aggregation(builder);
+        }
+
+
+        SearchRequest searchRequest = new SearchRequest()
+                .indices("out_es_summary_v1")
+                .types("doc")
+                .source(searchSourceBuilder);
+        log.info("start!");
+        SearchResponse search = restHighLevelClient.search(searchRequest);
+        log.info("end!");
+        return mapper.readValue(search.toString(), typeRef);
+    }
+
+
+    @SneakyThrows
+    private Object query(Set<String> ids) {
+        log.info("start... size: {}", ids.size());
+
+        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
+
+        boolQuery.must(addTerms(ids, "defendant_info.litigant_id"));
+        boolQuery.must(addTerms(ids, "plaintiff_info.litigant_id"));
+        boolQuery.filter(QueryBuilders.termQuery("deleted", "0"));
+
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
+                .query(boolQuery)
+                .size(0);
+
+        SearchRequest searchRequest = new SearchRequest()
+                .indices("winhc_index_company_court_open_announcement_v1")
+                .types("doc")
+                .source(searchSourceBuilder);
+        log.info("start!");
+        SearchResponse search = restHighLevelClient.search(searchRequest);
+        log.info("end!");
+        return mapper.readValue(search.toString(), typeRef);
+    }
+
+    private static BoolQueryBuilder addTerms(Set<String> ids, String key) {
+        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
+        List<List<String>> lists = fixedGrouping(new ArrayList<>(ids), 1024);
+        for (List<String> list : lists) {
+            boolQuery.should(QueryBuilders.termsQuery(key, list));
+        }
+        return boolQuery;
+    }
+
+
+    public static <T> List<List<T>> fixedGrouping(List<T> source, int n) {
+        if (null == source || source.size() == 0 || n <= 0)
+            return null;
+        List<List<T>> result = new ArrayList<>();
+        int remainder = source.size() % n;
+        int size = (source.size() / n);
+        for (int i = 0; i < size; i++) {
+            List<T> subset = null;
+            subset = source.subList(i * n, (i + 1) * n);
+            result.add(subset);
+        }
+        if (remainder > 0) {
+            List<T> subset = null;
+            subset = source.subList(size * n, size * n + remainder);
+            result.add(subset);
+        }
+        return result;
+    }
+}

src/main/java/com/winhc/phoenix/example/job/CompanyDynamicFix.java

@@ -0,0 +1,92 @@
+package com.winhc.phoenix.example.job;
+
+import com.mongodb.client.MongoCollection;
+import com.winhc.phoenix.example.framework.hbase.HbaseScan;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Table;
+import org.bson.Document;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+/**
+ * @author: XuJiakai
+ * 2021/2/1 16:54
+ */
+@Slf4j
+@Component
+@AllArgsConstructor
+public class CompanyDynamicFix {
+    private final Connection connection;
+    private final MongoTemplate mongoTemplate;
+
+    @Getter
+    @Builder
+    @ToString
+    private static class DynamicInfo {
+        private String id;
+        private String tn;
+        private String rowkey;
+
+    }
+
+    public void start() {
+        MongoCollection<Document> xjk_fix_company_dynamic = mongoTemplate.getCollection("xjk_fix_company_dynamic");
+        Consumer<List<Map<String, String>>> func = list -> {
+            List<DynamicInfo> dynamicInfos = list.stream()
+                    .map(d -> DynamicInfo.builder().id(d.get("rowkey")).tn(d.get("F:INFO_TYPE")).rowkey(d.get("F:BIZ_ID")).build())
+                    .collect(Collectors.toList());
+
+            List<Document> dos = new ArrayList<>();
+
+            Map<String, List<DynamicInfo>> collect = dynamicInfos.stream().collect(Collectors.groupingBy(DynamicInfo::getTn));
+            for (Map.Entry<String, List<DynamicInfo>> entry : collect.entrySet()) {
+                if (entry.getKey().equalsIgnoreCase("company_employment")) {
+                    continue;
+                }
+                if (entry.getKey().equalsIgnoreCase("wenshu_detail_combine")) {
+                    continue;
+                }
+                if (entry.getKey().equalsIgnoreCase("intellectual")) {
+                    continue;
+                }
+                List<DynamicInfo> value = entry.getValue();
+                TableName tn = TableName.valueOf(entry.getKey().toUpperCase());
+                try (Table table = connection.getTable(tn)) {
+                    boolean[] exists = table.exists(value.stream().map(DynamicInfo::getRowkey).map(String::getBytes).map(Get::new).collect(Collectors.toList()));
+                    for (int i = 0; i < exists.length; i++) {
+                        if (!exists[i]) {
+                            DynamicInfo dynamicInfo = value.get(i);
+                            Document document = new Document();
+                            document.put("_id", dynamicInfo.getId());
+                            document.put("tn", dynamicInfo.getTn());
+                            document.put("rowkey", dynamicInfo.getRowkey());
+                            dos.add(document);
+                        }
+                    }
+
+                } catch (Exception e) {
+                    log.error("tn不存在：{}", entry.getKey());
+                }
+            }
+            log.info("\t{}", dos);
+            if (!dos.isEmpty()) {
+                xjk_fix_company_dynamic.insertMany(dos);
+            }
+        };
+
+        new HbaseScan(connection, "COMPANY_DYNAMIC", func).columns("F:INFO_TYPE,F:BIZ_ID").batchSize(1000).scan();
+    }
+}

src/main/java/com/winhc/phoenix/example/job/DeleteHbaseByMongoJob.java

@@ -1,6 +1,7 @@
 package com.winhc.phoenix.example.job;
 
 import com.mongodb.client.MongoDatabase;
+import com.winhc.phoenix.example.dao.SearchDao;
 import com.winhc.phoenix.example.framework.mongo.MongoDbFastScan;
 import com.winhc.phoenix.example.service.HbaseOperationService;
 import lombok.AllArgsConstructor;
@@ -23,17 +24,19 @@ import java.util.stream.Collectors;
 public class DeleteHbaseByMongoJob {
     private final MongoTemplate mongoTemplate;
     private final HbaseOperationService hbaseOperationService;
+    private final SearchDao searchDao;
 
     public void start() {
         MongoDatabase db = mongoTemplate.getDb();
         Consumer<List<Document>> func = list -> {
             List<String> cids = list.stream().map(d -> d.getString("id")).collect(Collectors.toList());
-            System.out.println(cids);
-            hbaseOperationService.deleteByRowkey("COMPANY_DYNAMIC", cids);
+//            System.out.println(cids);
+//            hbaseOperationService.deleteByRowkey("COMPANY_DYNAMIC", cids);
+            searchDao.deleteByIds("winhc-dynamic", "company", cids);
         };
 
-        MongoDbFastScan mongoDbFastScan = new MongoDbFastScan("xjk_tmp_del_dyn", func, db);
-//                .batchSize(2000).threadNum(5);
+        MongoDbFastScan mongoDbFastScan = new MongoDbFastScan("xjk_fix_company_dynamic", func, db)
+                .batchSize(200).threadNum(5);
         mongoDbFastScan.scan();
     }
 }

src/main/java/com/winhc/phoenix/example/job/EsScanJob.java

@@ -27,19 +27,36 @@ public class EsScanJob {
     private final MongoTemplate mongoTemplate;
 
     public void start() {
-        MongoCollection<Document> person = mongoTemplate.getCollection("person_xjk_20201117");
+        String dsl = "{\n" +
+                "  \"query\": {\n" +
+                "    \"bool\": {\n" +
+                "      \"must_not\": [\n" +
+                "        {\"terms\": {\n" +
+                "          \"deleted\": [\n" +
+                "            \"0\",\n" +
+                "             \"9\",\n" +
+                "            \"1\"\n" +
+                "          ]\n" +
+                "        }}\n" +
+                "      ]\n" +
+                "    }\n" +
+                "  }\n" +
+                "}";
+        MongoCollection<Document> person = mongoTemplate.getCollection("xjk_company_equity_info_deleted_error");
         Consumer<SearchHit[]> func = list -> {
-            List<Document> documents = Arrays.stream(list).map(hit -> {
-                String id = hit.getId();
-                Document document = new Document(hit.getSourceAsMap());
+            List<Document> li = Arrays.stream(list).map(d -> {
+                String id = d.getId();
+                Document document = new Document();
                 document.put("_id", id);
+                document.put("rowkey", id);
+                document.put("ds", "20210422");
+                document.put("tn", "company_equity_info");
                 return document;
             }).collect(Collectors.toList());
-//            person.insertMany(documents);
+            person.insertMany(li);
         };
 
-        new EsFastScan(restHighLevelClient, func, "dishonest-executed-person-v2", "person").scan();
-//        new EsFastScan(restHighLevelClient, func, "winhc-company-test", "company").scan();
+        new EsFastScan(restHighLevelClient, func, "winhc_index_company_equity_info_v1", "_doc", dsl).scan();
     }
 
 }

src/main/java/com/winhc/phoenix/example/job/HbaseScanJob.java

@@ -1,5 +1,6 @@
 package com.winhc.phoenix.example.job;
 
+import com.winhc.phoenix.example.configuration.HbaseConfiguration;
 import com.winhc.phoenix.example.framework.hbase.HbaseFastScan;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -20,12 +21,14 @@ import java.util.stream.Collectors;
 @AllArgsConstructor
 public class HbaseScanJob {
     private final Connection connection;
+    private final HbaseConfiguration hbaseConfiguration;
 
     public void start() {
-        Consumer<List<Map<String, Object>>> func = list -> {
-            List<String> cids = list.stream().map(d -> (String)d.get("rowkey")).collect(Collectors.toList());
+        Consumer<List<Map<String, String>>> func = list -> {
+            List<String> cids = list.stream().map(d -> d.get("rowkey")).collect(Collectors.toList());
             System.out.println(cids);
         };
-        new HbaseFastScan(connection, "COMPANY_TAX_CONTRAVENTION", func).scan();
+
+        new HbaseFastScan(hbaseConfiguration::connection, "COMPANY_TAX_CONTRAVENTION", func).scan();
     }
 }

src/main/java/com/winhc/phoenix/example/service/HbaseQueryService.java

@@ -1,5 +1,7 @@
 package com.winhc.phoenix.example.service;
 
+import com.winhc.phoenix.example.vo.HbaseQueryParamVo;
+
 import java.util.List;
 
 /**
@@ -14,5 +16,7 @@ public interface HbaseQueryService {
 
     void asyncScan(String tableName, String rowPrefix);
 
-    Object humanBulkGet( String[] rowkey);
+    Object humanBulkGet(String[] rowkey);
+
+    Object fastGetHbase(String tableName, HbaseQueryParamVo hbaseQueryParamVo);
 }

src/main/java/com/winhc/phoenix/example/service/impl/HbaseQueryServiceImpl.java

@@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.winhc.phoenix.example.service.HbaseQueryService;
 import com.winhc.phoenix.example.util.HbaseResultUtils;
+import com.winhc.phoenix.example.vo.HbaseQueryParamVo;
 import lombok.AllArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -51,7 +52,7 @@ public class HbaseQueryServiceImpl implements HbaseQueryService {
             List<Object> list = new ArrayList<>();
             for (Result result : scanner) {
                 //每一行数据
-                Map<String, Object> columnMap = HbaseResultUtils.parseResult(result);
+                Map<String, String> columnMap = HbaseResultUtils.parseResult(result);
                 list.add(columnMap);
             }
             return list;
@@ -90,7 +91,7 @@ public class HbaseQueryServiceImpl implements HbaseQueryService {
             ResultScanner scanner = table.getScanner(scan);
             for (Result result : scanner) {
                 //每一行数据
-                Map<String, Object> columnMap = HbaseResultUtils.parseResult(result);
+                Map<String, String> columnMap = HbaseResultUtils.parseResult(result);
                 log.info(objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(columnMap));
             }
 
@@ -112,7 +113,26 @@ public class HbaseQueryServiceImpl implements HbaseQueryService {
             Map<String, String> collect = Arrays.stream(results)
                     .filter(r -> !r.isEmpty())
                     .collect(Collectors.toMap(r -> Bytes.toString(r.getRow()), r -> Bytes.toString(r.getValue(f, c)), (r1, r2) -> r1));
-            results.clone();
+            return collect;
+        }
+    }
+
+    @SneakyThrows
+    @Override
+    public Object fastGetHbase(String tableName, HbaseQueryParamVo hbaseQueryParamVo) {
+        List<String> keys = Arrays.stream(hbaseQueryParamVo.getQueryKey()).map(String::toUpperCase)
+                .collect(Collectors.toList());
+        List<Get> gets = Arrays.stream(hbaseQueryParamVo.getRowkey()).map(String::getBytes).map(Get::new).peek(get -> {
+            for (String key : keys) {
+                get.addColumn(f, key.getBytes());
+            }
+        }).collect(Collectors.toList());
+        try (Table table = connection.getTable(TableName.valueOf(tableName.toUpperCase().getBytes()))) {
+            Result[] results = table.get(gets);
+            Map<String, Map<String, String>> collect = Arrays.stream(results)
+                    .filter(r -> !r.isEmpty())
+                    .collect(Collectors.toMap(r -> Bytes.toString(r.getRow()), r -> keys.stream().collect(Collectors.toMap(k -> k, k -> Bytes.toString(r.getValue(f, k.getBytes())), (k1, k2) -> k1))
+                            , (r1, r2) -> r1));
             return collect;
         }
     }

src/main/java/com/winhc/phoenix/example/service/impl/SearchV8FastServiceImpl.java

@@ -66,7 +66,7 @@ public class SearchV8FastServiceImpl implements SearchService {
         ScriptSortBuilder scriptSortBuilder = SortUtil.getInstance().fastSort;
 
         QueryRescorerBuilder rescorerBuilder = new QueryRescorerBuilder(functionScoreQuery(new ScriptScoreFunctionBuilder(scriptSortBuilder.script())))
-                .windowSize(20)
+                .windowSize(50)
                 .setScoreMode(QueryRescoreMode.Multiply);
 
         Object search = searchDao.search(index, type, boolQuery, rescorerBuilder, null, fetchSourceContext, from, size);
@@ -78,8 +78,8 @@ public class SearchV8FastServiceImpl implements SearchService {
         BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
 
 
-        boolQuery.should(termQuery("cname.value.keyword", content).boost(100));
-        boolQuery.should(termQuery("history_name.value.keyword", content).boost(100));
+        boolQuery.should(termQuery("cname.value.keyword", content).boost(1000));
+        boolQuery.should(termQuery("history_name.value.keyword", content).boost(1000));
 
 
         boolQuery.should(disMaxQuery()
@@ -119,7 +119,7 @@ public class SearchV8FastServiceImpl implements SearchService {
         boolQuery.should(multiMatchQuery(content)
 //                .operator(Operator.AND)
                 .type(MultiMatchQueryBuilder.Type.CROSS_FIELDS)
-                .minimumShouldMatch("5<80%")
+                .minimumShouldMatch("5<85%")
                 .tieBreaker(0.3F)
                 .field("cname.value", 16)
                 .field("history_name.value", 12)

src/main/java/com/winhc/phoenix/example/service/impl/SearchV8ServiceImpl.java

@@ -77,8 +77,8 @@ public class SearchV8ServiceImpl implements SearchService {
 //        boolQuery.should(matchPhrasePrefixQuery("cname.value.pinyin", content).analyzer("ik_pinyin_analyzer_search").maxExpansions(5).boost(0.1F));
 
 
-        boolQuery.should(termQuery("cname.value.keyword", content).boost(100));
-        boolQuery.should(termQuery("history_name.value.keyword", content).boost(100));
+        boolQuery.should(termQuery("cname.value.keyword", content).boost(1000));
+        boolQuery.should(termQuery("history_name.value.keyword", content).boost(1000));
 
         /*boolQuery.should(disMaxQuery().add(
                 QueryBuilders.boolQuery()
@@ -89,10 +89,24 @@ public class SearchV8ServiceImpl implements SearchService {
 
 
         boolQuery.should(disMaxQuery()
+                .add(disMaxQuery()
+                        .add(termQuery("legal_entity_name.keyword", content).boost(10))
+                        .add(termQuery("holder.name.keyword", content).boost(5.5F))
+                        .add(termQuery("staff.name.keyword", content).boost(5.5F))
+                        .tieBreaker(0.3F))
+                .add(disMaxQuery()
+                        .add(matchQuery("legal_entity_name", content).boost(6).minimumShouldMatch("5<80%"))
+                        .add(matchQuery("holder", content).boost(10).minimumShouldMatch("5<80%"))
+                        .add(matchQuery("staff", content).boost(6).minimumShouldMatch("5<80%"))
+                        .tieBreaker(0.3F)
+                ).tieBreaker(0.3F)
+        );
+
+     /*   boolQuery.should(disMaxQuery()
                 .add(termQuery("legal_entity_name.keyword", content).boost(10))
                 .add(termQuery("holder.name.keyword", content).boost(5.5F))
                 .add(termQuery("staff.name.keyword", content).boost(5.5F))
-                .tieBreaker(0.3F));
+                .tieBreaker(0.3F));*/
 
       /*  boolQuery.should(matchQuery("icp", content).boost(6));
         boolQuery.should(matchQuery("app_info", content).boost(12));
@@ -216,13 +230,13 @@ public class SearchV8ServiceImpl implements SearchService {
         boolQuery.should(multiMatchQuery(content)
 //                .operator(Operator.AND)
                 .type(MultiMatchQueryBuilder.Type.CROSS_FIELDS)
-                .minimumShouldMatch("5<80%")
+                .minimumShouldMatch("5<85%")
                 .tieBreaker(0.3F)
                 .field("cname.value", 16)
                 .field("history_name.value", 12)
-                .field("holder.name", 6)
-                .field("staff.name", 6)
-                .field("legal_entity_name", 6)
+//                .field("holder.name", 6)
+//                .field("staff.name", 6)
+//                .field("legal_entity_name", 6)
 
         );
 

src/main/java/com/winhc/phoenix/example/util/HbaseResultUtils.java

@@ -13,11 +13,11 @@ import java.util.Map;
  * @Description:
  */
 public class HbaseResultUtils {
-    public static Map<String, Object> parseResult(Result result) {
+    public static Map<String, String> parseResult(Result result) {
         if (result == null || result.isEmpty()) {
             return null;
         }
-        Map<String, Object> columnMap = new HashMap<>();
+        Map<String, String> columnMap = new HashMap<>();
         String rowKey = null;
         for (Cell cell : result.listCells()) {
             if (rowKey == null) {

src/main/java/com/winhc/phoenix/example/util/QueryBuilderUtils.java

@@ -0,0 +1,26 @@
+package com.winhc.phoenix.example.util;
+
+
+import com.alibaba.fastjson.JSONObject;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.query.WrapperQueryBuilder;
+
+/**
+ * @author XuJiakai
+ * @ClassName: QueryBuilderUtils
+ * @Description:
+ * @date 2019/3/25 13:03
+ */
+public class QueryBuilderUtils {
+    public static final String q = "query";
+
+    public static WrapperQueryBuilder getQuery(JSONObject jsonObject) {
+        String dsl_json;
+        if (jsonObject.containsKey(q)) {
+            dsl_json = JsonUtils.jsonObjToString(jsonObject.getJSONObject(q));
+        } else {
+            dsl_json = JsonUtils.jsonObjToString(jsonObject);
+        }
+        return QueryBuilders.wrapperQuery(dsl_json);
+    }
+}

src/main/java/com/winhc/phoenix/example/util/SystemClock.java

@@ -0,0 +1,65 @@
+package com.winhc.phoenix.example.util;
+
+import java.sql.Timestamp;
+import java.util.TimeZone;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * @author XuJiakai
+ * @ClassName: SystemClock
+ * @Description: 解决高并发下currentTimeMillis的性能优化
+ * @date 2019/6/17 17:15
+ */
+public class SystemClock {
+    private final long period;
+    private final AtomicLong now;
+
+    private SystemClock(long period) {
+        this.period = period;
+        this.now = new AtomicLong(System.currentTimeMillis());
+        scheduleClockUpdating();
+    }
+
+    private static class InstanceHolder {
+        public static final SystemClock INSTANCE = new SystemClock(1);
+    }
+
+    private static SystemClock instance() {
+        return InstanceHolder.INSTANCE;
+    }
+
+    private void scheduleClockUpdating() {
+        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> {
+            Thread thread = new Thread(runnable, "System Clock");
+            thread.setDaemon(true);
+            return thread;
+        });
+        scheduler.scheduleAtFixedRate(() -> now.set(System.currentTimeMillis()), period, period, TimeUnit.MILLISECONDS);
+    }
+
+    private long currentTimeMillis() {
+        return now.get();
+    }
+
+    public static long now() {
+        return instance().currentTimeMillis();
+    }
+
+    public static Long getZeroTime() {
+        long current = SystemClock.now();
+        return current / (1000 * 3600 * 24) * (1000 * 3600 * 24) - TimeZone.getDefault().getRawOffset();
+    }
+
+    public static String nowDate() {
+        return new Timestamp(instance().currentTimeMillis()).toString();
+    }
+
+
+    public static void main(String[] args) {
+        String s = SystemClock.now() + "";
+        System.out.println(s);
+    }
+}

src/main/java/com/winhc/phoenix/example/util/WinhcCrypto.java

@@ -0,0 +1,59 @@
+package com.winhc.phoenix.example.util;
+
+/**
+ * @author: XuJiakai
+ * 2021/2/3 10:41
+ */
+
+import cn.hutool.core.util.CharsetUtil;
+import cn.hutool.crypto.symmetric.SymmetricAlgorithm;
+import cn.hutool.crypto.symmetric.SymmetricCrypto;
+
+import javax.crypto.spec.IvParameterSpec;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * @author: XuJiakai
+ * 2021/2/3 11:44
+ */
+
+/**
+ * AES加密模式：ECB
+ * 填充：pkcs7（java中pkcs5等效果于pkcs7）
+ * 数据块：128位（16字节）
+ * 密码(key)【128/192/256比特，16/24/32字符】：winhc_168_000000
+ * 偏移量(iv)【ecb模式下不用iv】：
+ * 输出：hex
+ * 字符集：utf-8
+ */
+public class WinhcCrypto {
+
+    private static final String key = "winhc_168_000000";
+    private static final String iv = "afasdfasdfasdfyu";
+    private final SymmetricCrypto aes;
+
+
+    public WinhcCrypto() {
+        IvParameterSpec ivParameterSpec = new IvParameterSpec(WinhcCrypto.iv.getBytes(StandardCharsets.UTF_8));
+        aes = new SymmetricCrypto(SymmetricAlgorithm.AES, key.getBytes(StandardCharsets.UTF_8));
+//        aes = new SymmetricCrypto(SymmetricAlgorithm.AES.getValue(), KeyUtil.generateKey(SymmetricAlgorithm.AES.getValue(), key.getBytes(StandardCharsets.UTF_8)), ivParameterSpec);
+    }
+
+    public String encode(String content) {
+        String encryptHex = aes.encryptHex(content);
+        return encryptHex;
+    }
+
+    public String decode(String encryptHex) {
+        String decryptStr = aes.decryptStr(encryptHex, CharsetUtil.CHARSET_UTF_8);
+        return decryptStr;
+    }
+
+
+    public static void main(String[] args) {
+        WinhcCrypto winhcCrypto = new WinhcCrypto();
+        System.out.println(winhcCrypto.encode("你好"));
+        String decode = winhcCrypto.decode("DA556055B35D6D3332D6776BF287D67D99E86C3383B3F9CDD14A64A6EBFF3C74");
+        System.out.println(decode);
+    }
+}

src/main/java/com/winhc/phoenix/example/util/WinhcCryptoNative.java

@@ -0,0 +1,101 @@
+package com.winhc.phoenix.example.util;
+
+import org.apache.commons.codec.binary.Hex;
+
+import javax.crypto.Cipher;
+import javax.crypto.SecretKey;
+import javax.crypto.spec.SecretKeySpec;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+
+/**
+ * AES加密模式：ECB
+ * 填充：pkcs7（java中pkcs5等效果于pkcs7）
+ * 数据块：128位（16字节）
+ * 密码(key)【128/192/256比特，16/24/32字符】：winhc_168_000000
+ * 偏移量(iv)【ecb模式下不用iv】：
+ * 输出：hex
+ * 字符集：utf-8
+ */
+
+/**
+ * @author: XuJiakai
+ * 2021/2/3 11:44
+ */
+public class WinhcCryptoNative {
+
+    private static final String key = "winhc_168_000000";
+
+
+    private static final String CIPHER_ALGORITHM = "AES/ECB/PKCS5Padding";
+    private final Lock lock = new ReentrantLock();
+
+    private final Cipher cipher;
+    private final SecretKey secretKey;
+
+
+    public WinhcCryptoNative() {
+        try {
+            cipher = Cipher.getInstance(CIPHER_ALGORITHM);
+            secretKey = new SecretKeySpec(key.getBytes(StandardCharsets.UTF_8), "AES");
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private byte[] encrypt(byte[] data) {
+        lock.lock();
+        try {
+            cipher.init(Cipher.ENCRYPT_MODE, secretKey);
+            return cipher.doFinal(data);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private byte[] decrypt(byte[] bytes) {
+        lock.lock();
+        try {
+            cipher.init(Cipher.DECRYPT_MODE, secretKey);
+            return cipher.doFinal(bytes);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    public String encryptHex(String data) {
+        return Hex.encodeHexString(encrypt(data.getBytes(StandardCharsets.UTF_8)));
+    }
+
+    public String decryptStr(String data) {
+        try {
+            byte[] bytes = Hex.decodeHex(data);
+            return new String(decrypt(bytes), StandardCharsets.UTF_8);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+
+    public static void main(String[] args) {
+        WinhcCryptoNative winhcCryptoNative = new WinhcCryptoNative();
+
+        List<String> list = Arrays.asList("413026199901011010", "413026199901011012");
+        for (String content : list) {
+            String cipher = winhcCryptoNative.encryptHex(content);
+            System.out.println(cipher);
+
+            String plain = winhcCryptoNative.decryptStr(cipher);
+            System.out.println(plain);
+        }
+
+    }
+}

src/main/java/com/winhc/phoenix/example/util/WinhcCryptoNativeCBC.java

@@ -0,0 +1,110 @@
+package com.winhc.phoenix.example.util;
+
+import org.apache.commons.codec.binary.Hex;
+
+import javax.crypto.Cipher;
+import javax.crypto.SecretKey;
+import javax.crypto.spec.IvParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+import java.nio.charset.StandardCharsets;
+import java.security.AlgorithmParameters;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+
+/**
+ * AES加密模式：CBC
+ * 填充【java中pkcs5等效果于pkcs7】：pkcs7
+ * 数据块：128位（16字节）
+ * 密码(key)【128/192/256比特，16/24/32字符】：winhc_168_000000
+ * 偏移量(iv)【ECB模式只用密钥即可对数据进行加密解密，CBC模式需要添加一个参数iv】：afasdfasdfasdfyu
+ * 输出：hex
+ * 字符集：utf-8
+ */
+
+/**
+ * @author: XuJiakai
+ * 2021/2/3 11:44
+ */
+public class WinhcCryptoNativeCBC {
+
+    private static final String key = "winhc_168_000000";
+    private static final String iv = "afasdfasdfasdfyu";
+
+
+    private static final String CIPHER_ALGORITHM = "AES/CBC/PKCS5Padding";
+    private final Lock lock = new ReentrantLock();
+
+    private final Cipher cipher;
+    private final SecretKey secretKey;
+    private AlgorithmParameters params;
+
+
+    public WinhcCryptoNativeCBC() {
+        try {
+            cipher = Cipher.getInstance(CIPHER_ALGORITHM);
+            secretKey = new SecretKeySpec(key.getBytes(StandardCharsets.UTF_8), "AES");
+            params = AlgorithmParameters.getInstance("AES");
+            params.init(new IvParameterSpec(iv.getBytes(StandardCharsets.UTF_8)));
+            cipher.init(Cipher.ENCRYPT_MODE, secretKey, params);
+
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private byte[] encrypt(byte[] data) {
+        lock.lock();
+        try {
+//            cipher.init(Cipher.ENCRYPT_MODE, secretKey, params);
+            return cipher.doFinal(data);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private byte[] decrypt(byte[] bytes) {
+        lock.lock();
+        try {
+            cipher.init(Cipher.DECRYPT_MODE, secretKey, params);
+            return cipher.doFinal(bytes);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    public String encryptHex(String data) {
+        return Hex.encodeHexString(encrypt(data.getBytes(StandardCharsets.UTF_8)));
+    }
+
+    public String decryptStr(String data) {
+        try {
+            byte[] bytes = Hex.decodeHex(data);
+            return new String(decrypt(bytes), StandardCharsets.UTF_8);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+
+    public static void main(String[] args) {
+        WinhcCryptoNativeCBC winhcCryptoNative = new WinhcCryptoNativeCBC();
+        List<String> ids = Arrays.asList("12010119810111007X", "120225198108240076");
+        for (int i = 0; i < 1000; i++) {
+            String res = winhcCryptoNative.encryptHex("12010119810111007X");
+            if ("bcb789e6022c4053a02d046327a639c10da537023d8dbe575efa7b67c2d4b5ee".equalsIgnoreCase(res) || "ab71045fe95da3d2f450a4bbf50428572bd8b59a58cd531eaa79a449adbd8af1".equalsIgnoreCase(res)) {
+                System.out.println("相同");
+            }else {
+                System.out.println("不同");
+            }
+        }
+
+//        System.out.println(winhcCryptoNative.decryptStr("55C0002A2E2CFD1EEC98FDD4B80E5B4D"));
+    }
+}

src/main/java/com/winhc/phoenix/example/vo/HbaseQueryParamVo.java

@@ -0,0 +1,13 @@
+package com.winhc.phoenix.example.vo;
+
+import lombok.Data;
+
+/**
+ * @author: XuJiakai
+ * 2021/1/26 10:31
+ */
+@Data
+public class HbaseQueryParamVo {
+    private String[] rowkey;
+    private String[] queryKey;
+}

src/main/resources/application.yml

@@ -27,12 +27,19 @@ spring:
 es:
   username: elastic
   password: elastic_168
-  host: es-cn-0pp0r32zf000ipovd.public.elasticsearch.aliyuncs.com
+  host: es-cn-oew22t8bw002iferu.public.elasticsearch.aliyuncs.com #es-cn-0pp0r32zf000ipovd.public.elasticsearch.aliyuncs.com
 
 hbase:
   config:
     hbase.zookeeper.quorum: hb-proxy-pub-uf6m8e1nu4ivp06m5-master1-001.hbase.rds.aliyuncs.com:2181,hb-proxy-pub-uf6m8e1nu4ivp06m5-master2-001.hbase.rds.aliyuncs.com:2181,hb-proxy-pub-uf6m8e1nu4ivp06m5-master3-001.hbase.rds.aliyuncs.com:2181
-
+    hbase.client.scanner.timeout.period: 120000
+    hbase.client.retries.number: 5
+    hbase.client.pause: 1000
+    hbase.client.max.perserver.tasks: 10
+    hbase.client.max.perregion.tasks: 10
+    hbase.client.keyvalue.maxsize: 524288000
+    hbase.client.ipc.pool.size: 5
+    zookeeper.recovery.retry: 5
 ---
 spring:
   profiles: prod

src/test/java/com/winhc/phoenix/example/job/BatchQueryEsTestTest.java

@@ -0,0 +1,19 @@
+package com.winhc.phoenix.example.job;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@Slf4j
+@SpringBootTest
+public class BatchQueryEsTestTest {
+    @Autowired
+    BatchQueryEsTest test;
+
+    @Test
+    public void t() {
+        test.start(1000, 12 + 3);
+    }
+
+}

src/test/java/com/winhc/phoenix/example/job/DeleteHbaseByMongoJobTest.java

@@ -1,5 +1,6 @@
 package com.winhc.phoenix.example.job;
 
+import com.winhc.phoenix.example.dao.SearchDao;
 import com.winhc.phoenix.example.service.HbaseQueryService;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
@@ -13,12 +14,17 @@ public class DeleteHbaseByMongoJobTest {
     DeleteHbaseByMongoJob deleteHbaseByMongoJob;
     @Autowired
     HbaseQueryService hbaseQueryService;
+    @Autowired
+    private SearchDao searchDao;
 
     @Test
     public void start() {
-//        deleteHbaseByMongoJob.start();
-        Object company_summary = hbaseQueryService.get("COMPANY_DYNAMIC", "1377601387_8548595199_de7d51d3edc0398781d8939cf1645c15");
-        System.out.println(company_summary);
+//        ArrayList<String> arr = new ArrayList<>();
+//        arr.add("1297330297_8397308799_d91ce09504b37a58cb67bccbf94555d8");
+//        searchDao.deleteByIds("winhc-dynamic", "company", arr);
+        deleteHbaseByMongoJob.start();
+//        Object company_summary = hbaseQueryService.get("COMPANY_DYNAMIC", "1377601387_8548595199_de7d51d3edc0398781d8939cf1645c15");
+//        System.out.println(company_summary);
 
 
     }

src/test/java/com/winhc/phoenix/example/job/HbaseScanJobTest.java

@@ -8,7 +8,7 @@ import org.springframework.boot.test.context.SpringBootTest;
 public class HbaseScanJobTest {
 
     @Autowired
-    HbaseScanJob job;
+    CompanyDynamicFix job;
 
     @Test
     public void start() {

src/test/java/com/winhc/phoenix/example/service/impl/SearchServiceImplTest.java

@@ -1,24 +1,62 @@
 package com.winhc.phoenix.example.service.impl;
 
-import com.winhc.phoenix.example.service.SearchService;
+import lombok.SneakyThrows;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.aggregations.Aggregation;
+import org.elasticsearch.search.aggregations.AggregationBuilders;
+import org.elasticsearch.search.aggregations.bucket.terms.Terms;
+import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 
+import java.util.Map;
+
 @SpringBootTest
 class SearchServiceImplTest {
     @Autowired
-    SearchService searchService;
+    RestHighLevelClient restHighLevelClient;
 
     @Test
     void tips() {
-        Object 百度网讯 = searchService.tips("百度网讯");
-        System.out.println(百度网讯);
+//        Object 百度网讯 = searchService.tips("百度网讯");
+//        System.out.println(百度网讯);
     }
 
+    @SneakyThrows
     @Test
     void query() {
-        Object 百度网讯 = searchService.query("北京百度网讯科技有限公司佛山分公司", 0, 10);
-        System.out.println(百度网讯);
+
+        TermsAggregationBuilder builder = AggregationBuilders.terms("name").field("human_pid")
+                .size(50)
+                .order(Terms.Order.aggregation("province_code_f", false))
+                .subAggregation(AggregationBuilders.terms("pro").field("province_code").size(2))
+                .subAggregation(AggregationBuilders.filter("province_code_f", QueryBuilders.termQuery("province_code", "11")));
+
+
+        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
+                .query(QueryBuilders.termQuery("human_name", "马云"))
+                .size(0)
+                .aggregation(builder);
+
+        SearchRequest searchRequest = new SearchRequest()
+                .indices("company-human-relation-v1")
+                .types("relation")
+                .source(searchSourceBuilder);
+
+        SearchResponse search = restHighLevelClient.search(searchRequest);
+
+
+        Map<String, Aggregation> stringAggregationMap = search.getAggregations().asMap();
+        System.out.println("out");
+
+//        ((ParsedStringTerms) stringAggregationMap.get("name")).getBuckets().stream().collect(Collectors.toMap(MultiBucketsAggregation.Bucket::getKeyAsString, t->t))
+
+
+        System.out.println(search);
     }
 }