如何基于Spring Data Scroll API实现通用多租户Elasticsearch检索服务

针对多租户Elasticsearch环境下不同索引的全量批量检索需求，我们可以通过动态数据载体+Scroll API的方式实现通用服务，以下是具体实现方案：

1. 依赖与基础配置 #

首先确保项目引入Spring Data Elasticsearch依赖（以Spring Boot为例）：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

配置ES连接信息（application.yml）：

spring:
  elasticsearch:
    rest:
      uris: http://your-es-host:9200
      username: your-username
      password: your-password

2. 定义通用数据载体 #

由于不同索引字段差异大，我们采用「固定公共字段+动态扩展字段」的DTO来适配所有索引：

public class GenericDocument {
    private String id;
    private String projectId;
    private String sampleId;
    // 存储索引自定义字段
    private Map<String, Object> extraFields;

    // Getter、Setter、构造方法省略
}

3. 实现通用Scroll检索服务 #

核心思路是利用Spring Data的ElasticsearchOperations操作ES，结合Scroll API分批拉取全量数据，同时根据索引名动态适配：

3.1 索引映射缓存（可选） #

提前加载并缓存所有索引的映射信息，用于字段类型校验或转换，避免类型异常：

@Component
public class IndexMappingCache {
    private final RestHighLevelClient restHighLevelClient;
    private final Map<String, Map<String, Object>> mappingCache = new ConcurrentHashMap<>();

    public IndexMappingCache(RestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
    }

    // 服务启动时加载所有索引映射
    @PostConstruct
    public void loadAllMappings() throws IOException {
        GetMappingsRequest request = new GetMappingsRequest();
        request.indices("*"); // 按需指定索引前缀或全部索引
        GetMappingsResponse response = restHighLevelClient.indices().getMapping(request, RequestOptions.DEFAULT);
        response.mappings().forEach((index, mapping) -> 
            mappingCache.put(index, mapping.getSourceAsMap())
        );
    }

    // 根据索引名获取映射
    public Map<String, Object> getMapping(String indexName) {
        return mappingCache.get(indexName);
    }
}

3.2 核心检索服务实现 #

@Service
public class GenericEsScrollService {
    private final ElasticsearchOperations elasticsearchOperations;
    private final IndexMappingCache indexMappingCache;

    public GenericEsScrollService(ElasticsearchOperations elasticsearchOperations, IndexMappingCache indexMappingCache) {
        this.elasticsearchOperations = elasticsearchOperations;
        this.indexMappingCache = indexMappingCache;
    }

    /**
     * 批量拉取指定索引的全量文档
     * @param indexName 目标索引名
     * @param batchSize 每批次拉取数量
     * @return 文档流，可按需处理
     */
    public Stream<GenericDocument> scrollAllDocuments(String indexName, int batchSize) {
        // 构建全量查询
        NativeSearchQuery query = new NativeSearchQueryBuilder()
                .withQuery(QueryBuilders.matchAllQuery())
                .withPageable(PageRequest.of(0, batchSize))
                .build();

        // 初始化Scroll，设置上下文超时时间
        Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1));
        SearchHits<Map<String, Object>> initialHits = elasticsearchOperations.searchScrollStart(
                scroll.getKeepAlive().getSeconds(),
                query,
                Map.class,
                IndexCoordinates.of(indexName)
        );

        String scrollId = initialHits.getScrollId();
        List<SearchHit<Map<String, Object>>> initialHitList = initialHits.getSearchHits();

        // 处理初始批次+循环拉取剩余数据
        return Stream.concat(
                initialHitList.stream().map(this::convertToGenericDoc),
                Stream.generate(() -> {
                    SearchHits<Map<String, Object>> nextHits = elasticsearchOperations.searchScrollContinue(
                            scrollId,
                            scroll.getKeepAlive().getSeconds(),
                            Map.class,
                            IndexCoordinates.of(indexName)
                    );
                    scrollId = nextHits.getScrollId();
                    List<SearchHit<Map<String, Object>>> nextHitList = nextHits.getSearchHits();

                    if (nextHitList.isEmpty()) {
                        // 必须清理Scroll上下文，释放ES资源
                        elasticsearchOperations.searchScrollClear(Collections.singletonList(scrollId));
                        return null;
                    }
                    return nextHitList.stream().map(this::convertToGenericDoc);
                }).takeWhile(Objects::nonNull).flatMap(Function.identity())
        );
    }

    /**
     * 将ES返回的Map结构转换为通用DTO
     */
    private GenericDocument convertToGenericDoc(SearchHit<Map<String, Object>> hit) {
        Map<String, Object> source = hit.getContent();
        GenericDocument doc = new GenericDocument();
        
        // 填充公共字段
        doc.setId(hit.getId());
        doc.setProjectId((String) source.get("project_id"));
        doc.setSampleId((String) source.get("sample_id"));

        // 提取自定义字段（排除公共字段）
        Map<String, Object> extraFields = new HashMap<>(source);
        extraFields.remove("id");
        extraFields.remove("project_id");
        extraFields.remove("sample_id");
        
        // 可选：根据映射信息做类型转换，避免Object类型操作异常
        Map<String, Object> mapping = indexMappingCache.get(hit.getIndex());
        if (mapping != null) {
            Map<String, Object> fields = (Map<String, Object>) mapping.get("properties");
            extraFields.forEach((key, value) -> {
                if (fields.containsKey(key)) {
                    String type = (String) ((Map<String, Object>) fields.get(key)).get("type");
                    // 示例：将字符串转为整数（根据实际映射调整）
                    if ("integer".equals(type) && value instanceof String) {
                        extraFields.put(key, Integer.parseInt((String) value));
                    }
                }
            });
        }

        doc.setExtraFields(extraFields);
        return doc;
    }
}

4. 关键注意事项 #

• Scroll资源清理：必须在检索完成后调用searchScrollClear释放ES服务器上的Scroll上下文，否则会持续占用内存。
• 批次大小与超时：batchSize建议设置为1000-5000（根据ES内存调整），Scroll超时时间避免过长或过短，一般1-5分钟即可。
• 多租户隔离：如果租户索引采用前缀/后缀区分，需在传入indexName时自动拼接租户标识，确保数据隔离。
• 性能优化：对于超大规模索引，可结合Slice Scroll实现并行检索，提升拉取速度。

内容的提问来源于stack exchange，提问作者Miguel Angel Salinas Gancedo