Reactor中使用Sinks.Many实现航班状态推送时如何过滤历史数据和管理订阅

方案选型与问题解答 #

1. 为什么不推荐用Sinks.many().multicast().replay()实现

这个类别的Sink缓存是全局通用的，仅支持配置固定缓存大小、TTL过期时间，不支持按业务维度（航班号）去重保留最新记录，也无法主动删除已到达航班的历史数据，完全匹配不了你要求新订阅者仅获取当前有效航班最新状态的需求，不用再纠结如何清理它的内部缓存，直接换方案即可。

2. 单订阅者对应单播Sink的方案问题解答

你的这个思路是可落地的，两个问题的解法如下：

• 管理所有活跃Sink：用线程安全的集合存储所有正常工作的单播Sink即可，比如CopyOnWriteArrayList<Sinks.Many<Flight>>。每次有航班新增、ETA变更、到达的事件触发时，遍历该集合逐个调用emit方法推送事件即可。
• 感知订阅销毁：你把单播Sink转为Flux给订阅者的时候，追加doOnCancel、doOnTerminate回调，触发时将当前Sink从活跃集合中移除即可。另外每次emit的时候可以判断返回的Sinks.EmitResult，如果是失败状态（比如订阅端已经主动取消），也直接将对应Sink从集合中清理，避免内存泄漏。

3. Flux.create与Sinks的核心区别

你用Flux.create配合onRequest、onDispose追踪订阅者的逻辑，本质是手动实现自定义多播逻辑，和unicast没有关系，unicast的定义是仅允许一个订阅者订阅的Sink。
两者的核心差异：

• FluxSink是和单次订阅绑定的：每次有新订阅者订阅Flux.create生成的Flux时，都会执行一次你传入的消费者函数，生成一个全新的FluxSink实例，生命周期完全和对应订阅者绑定，没有订阅者就没法主动发数据。
• Sinks是独立的发射器：不管有没有订阅者都可以主动推送数据，你可以自由控制事件分发规则，是发给一个订阅者还是多个订阅者，不需要每次新订阅都重写事件生产逻辑，灵活度更高。

更简化的实现方案 #

不用自己维护一堆单播Sink，你可以直接用以下实现，成本更低：

1. 先创建一个全局无缓存的多播Sink，只负责推送实时变更事件：

// 全局多播Sink，无历史缓存，仅推实时事件
private final Sinks.Many<Flight> globalChangeSink = Sinks.many().multicast().onBackpressureBuffer();
// 存储航班最新状态的线程安全Map
private final ConcurrentHashMap<String, Flight> flightMap = new ConcurrentHashMap<>();

2. 对外暴露的Flux这么实现，自动满足新订阅者先拿全量最新状态再收实时变更的需求：

public Flux<Flight> getFlightFlux() {
    return Flux.defer(() -> {
        // 先发射当前所有未到达的航班最新状态
        List<Flight> currentValidFlights = new ArrayList<>(flightMap.values());
        return Flux.fromIterable(currentValidFlights)
                // 拼接实时变更事件流
                .concatWith(globalChangeSink.asFlux())
                // 可选：按航班号+ETA+到达状态去重，避免给订阅者推送重复数据
                .distinctUntilChanged(flight -> Tuples.of(flight.getCode(), flight.getEta(), flight.isArrived()));
    });
}

3. 你业务侧有事件的时候，同时更新Map和往全局Sink推事件即可：

• 新增/变更航班：更新flightMap后调用globalChangeSink.tryEmitNext(最新Flight对象)
• 航班到达：更新flightMap放入arrived=true的对象，推送到Sink后，再将该航班号从Map中移除即可

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