RxJava新手求助：将Observable结果传递至另一个无限Observable

嘿，作为RxJava新手碰到这种需求很正常，其实核心思路就是用Subject来做中间桥梁——它既是Observer（可以接收每次Retrofit请求的结果），又是Observable（可以让其他订阅者持续接收数据）。下面给你一步步拆解实现方案：

1. 选择合适的Subject #

根据你的需求，推荐用PublishSubject：它会把所有后续发射的事件发送给订阅者，刚好匹配你“多次请求、持续传递结果”的场景。如果需要新订阅者能拿到最近一次的请求结果，可以换成BehaviorSubject（需要给它一个初始值，比如空列表）。

先定义全局的Subject：

// 全局或者在合适的生命周期类中定义
val pojoListSubject = PublishSubject.create<ArrayList<POJO>>()

2. 把Retrofit请求结果发送到Subject #

每次发起Retrofit请求后，在订阅回调里把拿到的POJO列表发射到Subject中：

// 你的Retrofit请求返回的Observable<ArrayList<POJO>>
result.subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        { pojoArray ->
            // 将本次请求的列表发送给Subject
            pojoListSubject.onNext(pojoArray)
        },
        { error ->
            // 这里可以选择处理错误，或者把错误传递给Subject（注意：Subject收到onError后会终止）
            // 如果不想因为一次请求失败导致整个Subject失效，建议在这里单独处理错误
            Log.e("RetrofitError", "请求失败", error)
        }
    )

3. 订阅Subject接收所有结果 #

在需要接收数据的地方（比如UI层），订阅这个Subject就能持续获取每次请求的POJO列表：

// 订阅Subject，接收所有请求的结果
val disposable = pojoListSubject.subscribe(
    { receivedPojoList ->
        // 在这里处理拿到的列表，比如合并到全局数据集合、更新RecyclerView等
        updateUI(receivedPojoList)
    },
    { error ->
        // 处理Subject传递过来的错误（如果之前选择传递的话）
    }
)

// 记得在生命周期销毁时取消订阅，避免内存泄漏
// 比如在Activity的onDestroy中：
// disposable.dispose()

可选：拆分列表为单个POJO发射 #

如果你的下游Observable需要的是单个POJO对象，而不是整个列表，可以用flatMapIterable把列表拆分成单个对象后再发送：

val singlePojoSubject = PublishSubject.create<POJO>()

// 处理请求结果时拆分列表
result.subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .flatMapIterable { pojoArray -> pojoArray } // 将列表转为单个POJO的Observable
    .subscribe(
        { pojo ->
            singlePojoSubject.onNext(pojo)
        },
        { error ->
            Log.e("RetrofitError", "请求失败", error)
        }
    )

// 订阅单个POJO
singlePojoSubject.subscribe { pojo ->
    // 处理单个POJO对象
}

注意事项 #

• 生命周期管理：一定要记得在组件（比如Activity/Fragment）销毁时取消订阅Subject的Disposable，否则会导致内存泄漏。
• 错误处理：如果Subject收到onError事件，它会立即终止，不再发射任何事件。所以如果某次请求失败不影响后续请求，建议在请求的订阅回调里单独处理错误，不要传递给Subject。
• 线程调度：如果下游订阅者需要在主线程处理UI，记得给Subject的订阅加上observeOn(AndroidSchedulers.mainThread())。

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