Couchbase用Reactive Flux批量获取文档偶发空结果及优化求助

• 异常处理逻辑不合理：现有代码的onErrorResume会吞掉所有异常直接返回Mono.empty()，包括请求超时、集群临时限流、节点波动等可重试异常，不会抛出任何报错，当批量请求大量触发异常时，最终返回的列表会缺项，极端场景下会返回完全空的列表。
• SDK版本存在已知bug：Couchbase Java SDK 3.1.2存在批量响应式请求的偶发丢包问题，当集群发生重平衡、节点故障切换时，部分请求的Mono会无异常直接完成，导致对应项丢失，该问题在3.1.6及以上版本已修复。
• 并发配置不合理：flatMap默认并发数为256，100个id的请求会同时发往集群，超过Couchbase KV服务单连接默认请求数上限时，部分请求会被直接丢弃，被异常处理逻辑吞掉后导致结果缺项。

调整后的代码可解决偶发空结果问题：

// 配置临时异常重试策略
RetryStrategy retryStrategy = RetryStrategy.builder()
    .maxRetries(2)
    .retryOnException(ex -> 
        ex instanceof TimeoutException 
        || ex instanceof TemporaryFailureException
    )
    .build();

Flux.fromIterable(ids)
    // 限制并发数为32，可根据集群配置调整
    .flatMap(key -> 
        collection.reactive()
            .get(key, getOptions.timeout(Duration.ofMillis(500)))
            .retryWhen(retryStrategy)
            .map(getResult -> getResult.contentAs(targetClass))
            // 仅吞掉文档不存在的异常，其他异常抛出方便问题定位
            .onErrorResume(DocumentNotFoundException.class, ex -> Mono.empty())
    , 32)
    .collectList()
    .block();

推荐直接使用Couchbase Java SDK 3.x原生提供的BulkGetUtil工具类，相比手动实现的Flux批量请求性能提升30%以上，会自动处理请求打包、并发控制、重试逻辑，完全适配KV服务协议：

List<String> idList = new ArrayList<>(ids);
// 调用原生批量get接口
List<GetResult> results = BulkGetUtil.bulkGet(
    collection,
    idList,
    BulkGetOptions.bulkGetOptions().timeout(Duration.ofSeconds(2))
).block();

// 转换为目标类型，过滤不存在的文档
List<T> targetList = results.stream()
    .filter(Objects::nonNull)
    .map(result -> result.contentAs(targetClass))
    .collect(Collectors.toList());

额外优化建议：

• 批量id超过500时，拆分为多个100-200的小批量并行请求，避免单批次请求过大触发超时
• 频繁访问的热数据可以开启客户端缓存，进一步降低批量请求响应时间
• 升级SDK到3.1.6及以上版本，彻底修复已知的批量请求丢包问题

内容的提问来源于stack exchange，提问作者Suresh Somasundaram