es开启数据压缩，es增加游标，searchAfter，范围查询方法

2026-08-13 05:12:07 +08:00 · 2025-05-09 15:21:45 +08:00
parent 8d544d10d0
commit 6cecff9e26
6 changed files with 324 additions and 196 deletions
@@ -0,0 +1,43 @@
+package com.cf.imes.module.executor.enums;
+
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+
+/**
+ * @author ES 索引类型
+ */
+
+@RequiredArgsConstructor
+@Getter
+public enum EsIndexEnum {
+
+
+    ORDER_PLATE_MODEL("生产单板材造型", "imes_order_plate_model"),
+
+    ORDER_OPTIMIZE_PLATE_MODEL("排单优化数据", "imes_order_optimize_plate_model"),
+
+    ORDER_PARTS_REMARK_MODEL("生产单配件备注", "imes_order_parts_remark_model"),
+
+    PLAN_PROCESS_SCHEME_OPTIMIZE_MODEL("排单的加工方案组对应的优化信息", "imes_plan_process_scheme_optimize_model"),
+
+    PLAN_ACTUAL_GOODS_MODEL("排单混单时选择的实际生产的大板信息", "imes_plan_actual_goods_model"),
+
+    PLAN_PROCESS_SCHEME_CONFIG("排单的加工方案组对应的配置信息", "imes_plan_process_scheme_config_model"),
+
+    ORDER_PLATE_COMPRESS_MODEL("生产单板材造型的压缩数据","imes_order_plate_compress_model");
+
+    /**
+     * 索引名称
+     */
+    private final String name;
+
+    /**
+     * 索引
+     */
+    private final String index;
+
+
+
+}
@@ -80,6 +80,7 @@ import static com.cf.imes.framework.common.exception.util.ServiceExceptionUtil.e
 import static com.cf.imes.framework.common.util.json.JsonUtils.*;
 import static com.cf.imes.framework.security.core.util.SecurityFrameworkUtils.getUserOrganId;
 import static com.cf.imes.module.executor.enums.ErrorCodeConstants.*;
+import static com.cf.imes.module.executor.enums.EsIndexEnum.*;
 import static com.cf.imes.module.system.enums.ErrorCodeConstants.*;

 /**
@@ -966,14 +967,7 @@ public class OptimizePlanServiceImpl implements OptimizePlanService {
    @Override
    public List<OrderModelDO> getPlateModel(Long orderId) {

-        List<PlateDO> plateDOS = plateMapper.selectPlateNum(Collections.singletonList(orderId), getUserOrganId());
-
-        if(plateDOS.isEmpty()){
-            return new ArrayList<>();
-        }
-
-        return esUtils.getEsDocument(FIELD_ORDER_ID, orderId, plateDOS.size(), OptimizePlanService.ORDER_PLATE_MODEL, OrderModelDO.class);
-
+        return esUtils.getEsDocumentByScroll(FIELD_ORDER_ID,orderId,1000,ORDER_PLATE_MODEL.getIndex(),OrderModelDO.class);

    }

@@ -5,17 +5,21 @@ import cn.hutool.core.collection.CollUtil;
 import co.elastic.clients.elasticsearch.ElasticsearchClient;
 import co.elastic.clients.elasticsearch._types.ElasticsearchException;
 import co.elastic.clients.elasticsearch._types.FieldValue;
+import co.elastic.clients.elasticsearch._types.SortOrder;
+import co.elastic.clients.elasticsearch._types.Time;
 import co.elastic.clients.elasticsearch._types.query_dsl.Query;
-import co.elastic.clients.elasticsearch.core.DeleteByQueryRequest;
-import co.elastic.clients.elasticsearch.core.SearchRequest;
-import co.elastic.clients.elasticsearch.core.SearchResponse;
+import co.elastic.clients.elasticsearch.core.*;
 import co.elastic.clients.elasticsearch.core.search.Hit;
 import co.elastic.clients.elasticsearch.indices.RefreshRequest;
+import co.elastic.clients.json.JsonData;
 import co.elastic.clients.json.JsonpMappingException;
 import com.cf.imes.framework.common.exception.ServiceException;
 import com.cf.imes.framework.common.util.json.JsonUtils;
 import com.cf.imes.framework.es.core.dal.ESDocument;
 import com.cf.imes.framework.es.core.service.ESDocumentService;
+import com.cf.imes.framework.security.core.LoginUser;
+import com.cf.imes.framework.security.core.util.SecurityFrameworkUtils;
+import com.cf.imes.module.executor.dal.dataobject.ordermodel.OrderModelDataTest;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;

@@ -118,7 +122,7 @@ public class EsUtils {
        SearchRequest.Builder builder = new SearchRequest.Builder();
        builder.index(index);

-        builder.query(q -> q.terms(b -> b.field(filedName).terms(t->t.value(fieldValues))));
+        builder.query(q -> q.bool(b->b.filter(f->f.terms(t->t.field(filedName).terms(te->te.value(fieldValues))))));
        try {
            SearchResponse<T> search = elasticsearchClient.search(builder.build(), targetClass);
            List<Hit<T>> hits = search.hits().hits();
@@ -143,7 +147,8 @@ public class EsUtils {
        builder.index(index);
        builder.size(size);

-        builder.query(q -> q.terms(b -> b.field(filedName).terms(t->t.value(fieldValues))));
+        builder.query(q-> q.bool(b-> b.filter(f-> f.terms(t-> t.field(filedName).terms(v->v.value(fieldValues))))));
+
        try {
            SearchResponse<T> search = elasticsearchClient.search(builder.build(), targetClass);
            List<Hit<T>> hits = search.hits().hits();
@@ -288,13 +293,190 @@ public class EsUtils {



+
+
+    // 范围查询（包括最大值和最小值），仅限于造型数据查询使用
+    public <T> List<T> getEsDocumentByRange(String filedName, Long value, String index, int minSortId, int maxSortId, Class<T> targetClass) {
+
+        SearchRequest.Builder builder = new SearchRequest.Builder();
+        builder.index(index);
+        int size = maxSortId - minSortId;
+        builder.size( size <= 0 ? 100 : size+2 );
+
+        builder.query(q->q.bool(b->b.filter(f->{
+            f.term(t->t.field(filedName).value(value));
+            f.range(r -> r.field("sortId").gte(JsonData.of(minSortId)).lte(JsonData.of(maxSortId)));
+            return f;
+        })));
+
+
+        try {
+            SearchResponse<T> search = elasticsearchClient.search(builder.build(), targetClass);
+            List<Hit<T>> hits = search.hits().hits();
+            if (CollUtil.isNotEmpty(hits)) {
+                return hits.stream().map(Hit::source).toList();
+            }
+            return new ArrayList<>();
+        }catch (JsonpMappingException e){
+            log.error(e.getMessage());
+            throw new ServiceException(ORDER_DATA_ERROR);
+        }catch (IOException | ElasticsearchException e) {
+            log.error(e.getMessage());
+            throw new ServiceException(DATA_DATA_ERROR);
+        }
+    }
+
+
+
+
+
+
+
+
+    // searchAfter 查询造型数据
+    public <T> List<T> getEsDocumentBySearchAfter(String filedName,
+                                                  Long value,
+                                                  String index,
+                                                  int size,
+                                                  int querySortId,
+                                                  Class<T> targetClass){
+
+        List<T> results = new ArrayList<>();
+
+        try {
+
+
+            SearchRequest.Builder searchBuilder = new SearchRequest.Builder()
+                    .index(index)
+                    .size(size)
+                    .sort(s -> s.field(f -> f.field("sortId").order(SortOrder.Asc)))
+                    .query(q -> q.bool(b -> b
+                            .filter(m -> m.term(t -> t.field(filedName).value(value)))
+                    ));
+
+            while (true) {
+
+                searchBuilder.searchAfter(FieldValue.of(querySortId));
+
+                SearchResponse<T> response = elasticsearchClient.search(searchBuilder.build(), targetClass);
+
+                List<Hit<T>> hits = response.hits().hits();
+                if (hits.isEmpty()) {
+                    break;
+                }
+
+                for (Hit<T> hit : hits) {
+                    results.add(hit.source());
+                }
+
+                querySortId += size;
+
+            }
+
+            return results;
+
+        } catch (JsonpMappingException e){
+            log.error(e.getMessage());
+            throw new ServiceException(ORDER_DATA_ERROR);
+        }catch (IOException | ElasticsearchException e) {
+            log.error(e.getMessage());
+            throw new ServiceException(DATA_DATA_ERROR);
+        }
+
+    }
+
+
+
+
+
+
+
+    // scroll 游标查询，目前仅用于查询造型数据
+    public <T> List<T> getEsDocumentByScroll(String filedName, Long value,Integer size, String index, Class<T> targetClass){
+
+        // scroll 保持游标有效时间
+        final String scrollTime = "1m";
+
+        List<T> result = new ArrayList<>();
+
+
+        String scrollId = null;
+
+        try {
+
+            // 执行第一次搜索，开启 scroll
+            SearchResponse<T> searchResponse = elasticsearchClient.search(s -> s
+                            .index(index)
+                            .scroll(Time.of(t -> t.time(scrollTime)))
+                            .size(size)
+                            .query(q->q.bool(b->b.filter(f->f.term(t->t.field(filedName).value(value))))),
+//                            .source(sou->sou.filter(f->f.includes("orderId","plateId","plateModelData","sortId"))),
+                    targetClass
+            );
+
+
+            scrollId = searchResponse.scrollId();
+            List<Hit<T>> hits = searchResponse.hits().hits();
+            hits.forEach(hit -> result.add(hit.source()));
+
+            // 循环 scroll 拉取剩余数据
+            while (!hits.isEmpty()) {
+                String finalScrollId = scrollId;
+                ScrollResponse<T> scrollResponse = elasticsearchClient.scroll(s -> s
+                        .scrollId(finalScrollId)
+                        .scroll(Time.of(t -> t.time(scrollTime))), targetClass);
+
+                scrollId = scrollResponse.scrollId();
+                hits = scrollResponse.hits().hits();
+
+                if (!hits.isEmpty()) {
+                    hits.forEach(hit -> result.add(hit.source()));
+                }
+            }
+
+            return result;
+
+        }catch (IOException | ElasticsearchException e){
+            log.error(e.getMessage());
+            throw new ServiceException(DATA_DATA_ERROR);
+        }finally {
+            // 清理 scroll
+            clearScroll(scrollId);
+        }
+
+    }
+
+
+
+
+
+
+    private void clearScroll(String scrollId) {
+        try {
+            if (scrollId != null && !scrollId.isEmpty()) {
+                elasticsearchClient.clearScroll(c -> c.scrollId(scrollId));
+            }
+        } catch (Exception e) {
+            log.warn("Failed to clear scroll: {}", e.getMessage());
+        }
+    }
+
+
+
+
+
+
+
+
+
+
    // ES 文档删除
    public void deleteEsDocument(String filedName,Long value, String index) {

        DeleteByQueryRequest.Builder builder = new DeleteByQueryRequest.Builder();
        builder.index(index);

-        builder.query(q -> q.term(b -> b.field(filedName).value(value)));
+        builder.query(q -> q.bool(b->b.filter(f->f.term(t->t.field(filedName).value(value)))));

        try {
            elasticsearchClient.deleteByQuery(builder.build());
@@ -351,7 +533,7 @@ public class EsUtils {
        DeleteByQueryRequest.Builder builder = new DeleteByQueryRequest.Builder();
        builder.index(index);

-        builder.query(q -> q.terms(b -> b.field(filedName).terms(e -> e.value(fieldValues))));
+        builder.query(q -> q.bool(b->b.filter(f->f.terms(t->t.field(filedName).terms(te->te.value(fieldValues))))));

        try {
            elasticsearchClient.deleteByQuery(builder.build());
@@ -384,6 +566,30 @@ public class EsUtils {



+    public void savePlateModelEs(String index,List<OrderModelDataTest > targetClass) {
+        try {
+
+            BulkRequest.Builder br = new BulkRequest.Builder();
+            LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
+            targetClass.forEach(esDocument -> {
+                esDocument.setOrganId(loginUser.getOrganId());
+                br.operations(op -> op.index(idx -> idx
+                        .index(index)
+                        .document(esDocument)));
+            });
+            elasticsearchClient.bulk(br.build());
+
+        } catch (IOException | ElasticsearchException e) {
+            e.printStackTrace();
+            log.error(e.getMessage());
+            throw new ServiceException(DATA_DATA_ERROR);
+        }
+
+    }
+
+
+
+


    // ES 文档数据刷新