Reactive Spring Boot：如何不重复Redis调用从Flux分离数据？

你的问题根源在于Reactor的Flux默认是冷流：每个订阅者触发时，都会重新执行整个上游数据流的逻辑。所以四个过滤后的Flux各自被订阅时，getDataFromRedisForKeys(keys)会被执行四次，导致四次Redis查询。

要实现单次非阻塞的数据获取并共享给多个下游订阅者，只需将冷流转为热流，让所有下游订阅者复用同一次上游执行结果。以下是两种常用方案：

方案1：使用publish().autoConnect()（实时共享，无缓存） #

适合所有订阅者在数据流启动前或启动过程中订阅的场景，实时推送数据，不会缓存结果：

Flux<String> keys; // 从代码其他部分获取的keys流

// 将Redis查询的冷流转为可共享的热流
Flux<RedisData> sharedResponse = getDataFromRedisForKeys(keys)
    .publish()
    .autoConnect(); // 第一个订阅者出现时自动触发Redis查询

// 基于共享热流做过滤，所有下游复用同一次查询数据
Flux<RedisData> responseTypeA = filterATypeResponseFromRedisDataFlux(sharedResponse);
Flux<RedisData> responseTypeB = filterBTypeResponseFromRedisDataFlux(sharedResponse);
Flux<RedisData> responseTypeC = filterCTypeResponseFromRedisDataFlux(sharedResponse);
Flux<RedisData> responseTypeD = filterDTypeResponseFromRedisDataFlux(sharedResponse);

publish()创建ConnectableFlux，它不会自动执行上游逻辑；autoConnect()会在第一个订阅者订阅时自动启动上游的Redis查询，后续所有订阅者都共享这个数据流的输出。

方案2：使用cache()（缓存结果，支持延迟订阅） #

如果存在订阅者在数据流完成后才订阅的场景，cache()会缓存所有元素，后续订阅者直接获取缓存数据：

Flux<String> keys; // 从代码其他部分获取的keys流

// 缓存Redis查询结果，供所有订阅者复用
Flux<RedisData> sharedResponse = getDataFromRedisForKeys(keys)
    .cache();

// 基于缓存的数据流做过滤
Flux<RedisData> responseTypeA = filterATypeResponseFromRedisDataFlux(sharedResponse);
Flux<RedisData> responseTypeB = filterBTypeResponseFromRedisDataFlux(sharedResponse);
Flux<RedisData> responseTypeC = filterCTypeResponseFromRedisDataFlux(sharedResponse);
Flux<RedisData> responseTypeD = filterDTypeResponseFromRedisDataFlux(sharedResponse);

注意事项 #

确保getDataFromRedisForKeys使用的是真正的响应式Redis客户端（比如Spring Data Redis的ReactiveRedisTemplate），这样整个流程全程非阻塞，不会因为转热流引入阻塞操作。

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