RxJava(Kotlin)、Observable.amb 和 PublishSubject 没有触发答案

【问题标题】：RxJava(Kotlin), Observable.amb and PublishSubject is not firingRxJava(Kotlin)、Observable.amb 和 PublishSubject 没有触发
【发布时间】：2016-11-16 11:14:02
【问题描述】：

我们正在尝试观察 15 秒的间隔，或者每当我们针对我们的主题 refreshEventsSubject 触发 onNext 时，但没有成功。

主题是这样开始的

private val refreshEventsSubject = PublishSubject<Long>()

然后我们试着像这样观察它

Observable.merge(Observable.interval(0, 15, TimeUnit.SECONDS), refreshEventsSubject) .subscribe { ... }

我们每 15 秒从间隔中获取事件，但主题在运行后没有触发

refreshEventsSubject.onNext(0)

任何想法表示赞赏。

（一切都是用 Kotlin 编写的）

【问题讨论】：

在 RxJava 中，您通过 PublishSubject.create() 创建一个 PublishSubject 静态方法，因为通过无参数构造函数创建一个方法不起作用。我不知道 RxKotlin 是否对此进行了补偿。
是的，PublishSubject 的 Kotlin rx 绑定看起来像这样 fun <T> PublishSubject() : PublishSubject<T> = PublishSubject.create()

标签： android rx-java kotlin rx-kotlin

【解决方案1】：

确保不会从主线程调用refreshEventsSubject.onNext(0)，因为它可能会导致死锁！

还可以使用http://reactivex.io/documentation/operators/amb.html 而不是合并，因为合并会在您的主题上调用 onNext 时发出两个事件。

【讨论】：

【解决方案2】：

阅读 AMB 的文档：http://reactivex.io/documentation/operators/amb.html

特别是emit all of the items from only the first of these Observables to emit an item or notification。

你要找的运营商大概是Observable.merge:http://reactivex.io/documentation/operators/merge.html

【讨论】：

当然，它仍然无法正常工作。如果我们直接订阅主题，我们会得到事件，但是当与间隔流结合/合并时，什么都不会发生。