RxJava：订阅后阻塞可观察对象？答案

【问题标题】：RxJava: Blocking on an observable after subscription?RxJava：订阅后阻塞可观察对象？
【发布时间】：2015-08-07 16:15:23
【问题描述】：

我特别需要一个 observable。

通常，我的 observables 在不同的线程中运行。但是，有时他们需要在订阅过程中阻塞另一个线程。未来的行为方式。

一个例子：

val o = Observable.create(/* computation */)
  .subscribeOn(Schedulers.newThread())
  .observeOn(/* current thread */);

val s = o.subscribe(/* subscriber */);`

突然，另一个线程上发生了一个事件，通知当前线程它应该等待订阅s 的执行。（例如 Android 的 onPause。）

我该怎么做？如何等待订阅s 甚至可能检索所有结果？

（主题？）

【问题讨论】：

o.toBlocking() 相当于future.get()
@VladimirMironov 是的，但那是之前我订阅了 observable。我需要的是订阅后。所以基本上，没有双重计算。
也许序列化的 ReplaySubject 可以解决问题。本质上，您将主题转换为可观察对象，然后将其转换为阻塞对象...或在直接线程上订阅该可观察对象。
... 或者注意将结果缓存在原始 observable 中，然后在直接线程上订阅。
我看到有一个Observable#cache() 方法。这可能是最简单的解决方案。

标签： multithreading system.reactive rx-java rx-android rx-scala

【解决方案1】：

对此最简单的解决方案似乎是：

使用Observable#cache() 使可观察对象成为缓存对象。（如果您的 observable 是一个值流，请使用适当的 Subject 并将其用作 observable。）
订阅缓存的可观察对象/主题。
当事件发生时，只需在您需要的线程上订阅缓存的 observable / subject。酌情取消订阅。

【讨论】：