RxJava2 - 使用 PublishSubject 发射项目答案

【问题标题】：RxJava2 - Emitting items using PublishSubjectRxJava2 - 使用 PublishSubject 发射项目
【发布时间】：2018-05-31 14:32:58
【问题描述】：

我有一个场景

subject1: PublishSubject 和 subject2:BehaviorSubject。

首先，我为subject1 发出单个项目，然后我为subject2 发出项目，但之后我还想向subject1 发出不同的项目。

fun emittingItems() {
    subject1.onNext(functionA1)
    subject2.onNext(functionB)
    if (something) subject1.onNext(functionA2)
}

发生的情况是，我按以下顺序收到一个项目：functionA1、functionA2、functionB。

为什么会出现这种行为？如何按以下顺序发出项目：functionA1、functionB、functionA2。

订阅主题：

val disposable = viewModel.subject1
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::someFunction())
disposables.add(disposable)

【问题讨论】：

标签： android rx-java2 subject

【解决方案1】：

使用observeOn(AndroidSchedulers.mainThread())，您可以在主线程上安排事件传播。调度本身是顺序的，而每个调度的Runnable 可能会处理多个添加到用于它的队列中的元素。

这是一种竞争条件，在主线程本身调用emittingItems() 时肯定会出现，并且在从任何其他线程调用它时可能会出现。

但由于您正在处理两个不同的异步流，因此您不能期望在两个不同的观察者中进行任何顺序观察。

您可以通过将两个源合并为一个流来实现给定的目标：

Observable.merge(subject1, subject2)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(subject);

【讨论】：

使用merge 我不会得到这个序列：functionA1, functionA2, functionB?
不，不是您的设置。合并后的 observable 将发出下一个事件，因为这些事件是由主题发出的。
谢谢。顺便说一句，如果我愿意subject1.onNext(A), subject1.onNext(B), subject1.onNext(C)，我会收到A,B,C，因为它是同一个主题，对吧？
视情况而定。和上面一样，是的。通常，您将收到带有观察者的事件，具体取决于您设置的流。它通常与来源无关。