如何像 RxJava1 那样在 RxJava2 中使用 Subject (AsyncSubject, BehaviorSubject...)？

【问题标题】：How to use Subject (AsyncSubject, BehaviorSubject...) in RxJava2 like the way RxJava1 did?如何像 RxJava1 那样在 RxJava2 中使用 Subject (AsyncSubject, BehaviorSubject...)？
【发布时间】：2019-07-22 16:16:35
【问题描述】：

在 Rxjava1 中，我们可以像这样订阅 Subject 到 Observable：

val asyncSubject = AsyncSubject<T>.create()
Observable<T>.subscribe(asyncSubject);
asyncSubject.subscribe(...)

不知道如何在 Rxjava2 Flowable 中实现同样的功能？

我正在考虑这样的事情

Flowable<T>
.doOnComplete { t -> asyncSubject.OnComplete() }
.subscribe { t -> asyncSubject.onNext(t) }

有更好/更简洁的方法吗？

【问题讨论】：

您的第一个示例应该可以正常工作...是否出现错误？
@AlexHart 是的，它在 RxJava1 中运行良好，但我不能在 Rxjava2 中为 Flowable 做到这一点。
啊，对于 flowables，您可以使用处理器。就像一个行为处理器
@AlexHart 太棒了。您能否将其发布为答案？

标签： android rx-java2 behaviorsubject

【解决方案1】：

RxJava2 将事物分成Flowable 和Observable。当您需要背压时使用 Flowables。

由于这次突破，我们有两种不同类型的 Subject 对象，您的常规旧 BehaviorSubject 等与 Observable 一起使用，以及一组名为 Processors 的新类，例如BehaviorProcessor等

这些类的工作方式与主题基本相同，但可以使用Flowables 而不是Observables。

在您的情况下，您可能希望使用 Processor 而不是主题：

val proc = AsyncProcessor.create<Int>()
val flowable = Flowable.just(1)
flowable.subscribe(proc)

【讨论】：