subscribeOn 主线程和observerOn 调用者线程RxJava2

Question

我正在为一个非线程安全的库包装遗留代码。 当库的客户端从主线程调用 API 时，我需要切换到主线程，然后切换回调用者线程以返回结果

我认为我可以使用（这是针对 Android，但问题更笼统）

internal object TransformCompletableTemporarilySwitchToMainThread : CompletableTransformer {
    override fun apply(upstream: Completable): CompletableSource {
        return upstream
                .observeOn(Schedulers.trampoline())
                .subscribeOn(AndroidSchedulers.mainThread())
    }
}

有没有像 RxJava1 的 Schedulers.immediate() 这样的东西？我知道对于测试，您可以将Schedulers.immediate() 替换为Schedulers.trampoline()，但是从文档和我运行的测试来看，Schedulers.trampoline() 与Schedulers.immediate() 没有太大关系 有其他方法吗？

添加

   /**
 * Returns a default, shared {@link Scheduler} instance whose {@link io.reactivex.Scheduler.Worker}
 * instances queue work and execute them in a FIFO manner on one of the participating threads.
 * <p>
 * The default implementation's {@link Scheduler#scheduleDirect(Runnable)} methods execute the tasks on the current thread
 * without any queueing and the timed overloads use blocking sleep as well.
 * <p>
 * Note that this scheduler can't be reliably used to return the execution of
 * tasks to the "main" thread. Such behavior requires a blocking-queueing scheduler currently not provided
 * by RxJava itself but may be found in external libraries.
 * <p>
 * This scheduler can't be overridden via an {@link RxJavaPlugins} method.
 * @return a {@link Scheduler} that queues work on the current thread
 */

这两部分是什么意思？

• 返回一个默认的共享 {@link Scheduler} 实例，其 {@link io.reactivex.Scheduler.Worker} * 实例排队工作并执行 它们在其中一个参与线程上以 FIFO 方式进行。

和

• @return 一个 {@link Scheduler} 将当前线程上的工作排入队列

Answer 1

immediate 或 trampoline 调度程序不适合返回特定线程。您需要一个单线程调度程序，您可以从 ExecutorService 创建它：

ExecutorService exec = Executors.newSingleThreadedExecutor();
Scheduler singleThreaded = Schedulers.from(exec);

Observable.fromCallable(() -> api.init())
.subscribeOn(singleThreaded)
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(v -> System.out.println("Initialized"))
.observeOn(singleThreaded)
.map(v -> api.getData(v))
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(v -> System.out.println("Some data: "))
.observeOn(singleThreaded)
.doOnNext(v -> api.close())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(v -> System.out.println("Done"))
;

// later
exec.shutdown();

编辑：：

不可能返回到任意线程。但是，如果一个线程有一个循环器/处理程序，那么您可以使用 AndroidScheduler 将其转换为调度程序，并通过observeOn 定位它。