Spring WebFlux中Flux偶尔返回空或部分响应问题求助

问题背景 #

基于Spring WebFlux实现了从Elasticsearch获取数据的API，返回ResponseEntity<Flux<Product>>，但偶尔会出现空响应或部分响应的情况，响应大小约100KB。

相关代码片段 #

Controller #

@PostMapping(value = "/test")
public ResponseEntity<Flux<Product>> getresult(@Valid @RequestBody RequestDTO requestDTO, ServerWebExchange serverWebExchange) {
    return new ResponseEntity<>(service.get(requestDTO), HttpStatus.OK);
}

ServiceImpl #

return elasticSearchRepository.search(searchRequest)
        .doOnError(err -> log.error(Constants.Log.ES_ERROR_MESSAGE, err.getMessage()))
        .map(pp -> serviceImplHelper.makeProductDTO(pp, ma, intermediateObject.getMapping()))
        .log();

ElasticSearchRepositoryImpl #

public Flux<ProductPage> search(SearchRequest searchRequest) {
    return reactiveElasticsearchClient.search(searchRequest)
            .map(r -> mapper.convertValue(r.getSourceAsMap(), ProductPage.class))
            .groupBy(p -> p.getMetaData().getCId())
            .flatMap(g -> g.reduce((a, b) -> a.getMetaData().getCDate().compareTo(b.getMetaData().getCDate()) > 0 ? a : b))
            .switchIfEmpty(Flux.defer(() -> Flux.error(new NoDataException(Constants.Error._NOT_FOUND))));
}

排查方向与解决方案 #

1. 分组聚合逻辑的潜在问题 #

groupBy属于有状态操作，若上游流提前终止（比如ES查询异常），分组后的Flux可能无法完成聚合，导致结果丢失或不完整。建议替换为更可靠的collectMap聚合方式：

public Flux<ProductPage> search(SearchRequest searchRequest) {
    return reactiveElasticsearchClient.search(searchRequest)
            .map(r -> mapper.convertValue(r.getSourceAsMap(), ProductPage.class))
            // 按CId分组，直接保留每组中CDate最新的元素
            .collectMap(
                    p -> p.getMetaData().getCId(),
                    Function.identity(),
                    (existing, incoming) -> existing.getMetaData().getCDate().compareTo(incoming.getMetaData().getCDate()) > 0 ? existing : incoming
            )
            .flatMapMany(map -> Flux.fromIterable(map.values()))
            .switchIfEmpty(Flux.defer(() -> Flux.error(new NoDataException(Constants.Error._NOT_FOUND))));
}

2. 流异常处理不彻底 #

doOnError仅记录错误日志，不会干预流的错误传递；若map转换（makeProductDTO）中抛出未捕获异常，会直接终止流，导致已发送部分响应后中断。建议将map改为flatMap，内部捕获转换异常，避免中断整个流：

return elasticSearchRepository.search(searchRequest)
        .flatMap(pp -> {
            try {
                return Mono.just(serviceImplHelper.makeProductDTO(pp, ma, intermediateObject.getMapping()));
            } catch (Exception e) {
                log.error("转换ProductDTO失败，CId: {}", pp.getMetaData().getCId(), e);
                return Mono.empty(); // 丢弃异常元素，继续处理后续数据
            }
        })
        .doOnError(err -> log.error(Constants.Log.ES_ERROR_MESSAGE, err.getMessage()))
        .log();

3. Elasticsearch查询超时或结果截断 #

检查ES查询的超时配置，避免因查询超时导致流提前终止：

// 在构造SearchRequest时添加超时设置
searchRequest.source().timeout(new TimeValue(15, TimeUnit.SECONDS));

同时在reactiveElasticsearchClient.search()后添加log()，查看返回的SearchHit数量是否符合预期，确认ES是否返回了完整结果。

4. WebFlux响应处理简化 #

返回ResponseEntity<Flux<Product>>时，若流未正常触发complete信号，WebFlux可能无法发送完整响应。建议简化返回类型，直接返回Flux<Product>，让WebFlux原生处理流响应：

@PostMapping(value = "/test")
public Flux<Product> getresult(@Valid @RequestBody RequestDTO requestDTO) {
    return service.get(requestDTO);
}

5. 验证流完整性 #

查看log()操作的输出，确认是否有onComplete信号。如果日志只显示onNext无onComplete，说明上游流未正常结束，需排查ES客户端是否存在连接中断、静默异常等情况。

内容的提问来源于stack exchange，提问作者Amulya M