RxJava：新Observer订阅时如何返回Flowable的最后2个元素

我有一个返回Flowable<String>的方法，该Flowable使用BackpressureStrategy.LATEST策略，属于一个库类。现在需求是：当新的Observer订阅这个Flowable时，能获取到最后2个历史元素。我尝试过将Flowable改为BUFFER策略，但这样会向Observer发送大量元素，不符合需求。另外，当前使用buffer(2)时，如果Observer延迟订阅，该方法不会触发，无法拿到历史元素。

代码示例如下：

private fun observeFlowable(): Flowable<String> {
  observer().toFlowable(BackpressureStrategy.LATEST)
}

// 订阅代码
private fun subscribeFlowable() {
    observeFlowable().buffer(2).doSomething()
}

解决方案 #

要实现“新订阅者获取最近2个历史元素+后续新元素”的需求，最直接的方式是使用**重播缓存（Replay）**相关的操作符，它能缓存指定数量的历史元素，并在新订阅时自动发送这些缓存内容。

方法1：使用replay(2) + autoConnect() #

通过replay(2)操作符缓存最近的2个元素，再用autoConnect()让多个订阅者共享同一个流实例，避免重复订阅原Flowable（如果原流是冷流的话）。修改后的代码如下：

// 改造原Flowable，添加重播缓存
private fun observeFlowableWithReplay(): Flowable<String> {
    return observeFlowable()
        .replay(2) // 缓存最近2个元素
        .autoConnect() // 自动连接，当有第一个订阅者时触发原流订阅
}

// 订阅时直接使用改造后的Flowable
private fun subscribeFlowable() {
    observeFlowableWithReplay()
        .subscribe { string -> 
            // 处理收到的元素，新订阅会先收到缓存的2个历史元素，再接收新元素
        }
}

方法2：使用ReplaySubject作为中间层 #

如果需要更灵活的控制（比如手动管理缓存生命周期），可以用ReplaySubject.createWithSize(2)作为中间代理：

// 创建一个缓存2个元素的ReplaySubject
private val replaySubject = ReplaySubject.createWithSize<String>(2)

// 初始化时将原Flowable的元素转发到Subject
init {
    observeFlowable().subscribe(replaySubject)
}

// 订阅时直接订阅这个Subject
private fun subscribeFlowable() {
    replaySubject.subscribe { string ->
        // 新订阅会收到最近2个历史元素，以及后续新元素
    }
}

注意事项 #

• 两种方法都能保证新订阅者获取到最近2个历史元素，同时不会像BUFFER策略那样发送大量旧元素。
• 如果原Flowable是热流（比如基于事件回调的流），replay操作符能完美适配；如果是冷流，autoConnect()可以避免每次订阅都重新触发原流的逻辑。
• buffer(2)的作用是将连续的2个元素打包成列表发送，它本身不具备缓存历史元素的能力，所以延迟订阅时无法拿到之前的元素，这也是它不适用的原因。

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