RxJava单一后台线程调度程序

Question

我对RxJava相当新，所以这可能是一个愚蠢的问题。我将描述我的情况。

我有一些在UI线程上运行的代码，它们将更新一些图像，但这些图像并不是非常重要，并且在生成它们时会消耗一些资源，因此我想在单个线程上（当然不是UI线程）逐个生成它们。我猜测跳板调度程序是我想要的，但我的问题是如果我使用它，它会在UI线程上执行工作，而我希望在另一个线程上执行工作。

显然，我可以编写自己的线程，在其中可以排队项目，然后逐个处理它们，但我想知道RxJava是否有简单的解决方案？

我的当前代码如下：

Observable<Bitmap> getImage = Observable.create(new Observable.OnSubscribe<Bitmap>() {
    @Override public void call(Subscriber<? super Bitmap> subscriber) {
        Log.w(TAG,"On ui thread? "+ UIUtils.isRunningOnUIThread());
        subscriber.onNext(doComplexTaskToGetImage());
        subscriber.onCompleted();
    }
});

getImage.subscribeOn(Schedulers.trampoline()).subscribe(new Action1<Bitmap>() {
    @Override public void call(Bitmap bitmap) {
        codeToSetTheBitmap(bitmap);
    }
});

我的日志一直显示"On ui thread?"为true。那么，我如何使该代码以及所有后续尝试在一个单独的线程上运行（而不是UI线程），并且无需编写大量排队工作的代码来实现顺序执行？

编辑：

我相信现在可以使用Schedulers.single()来完成这个任务，或者如果您想要自己创建，可以使用new SingleScheduler()。我仍在测试，但我认为它可以实现我之前发布问题时所期望的功能。

- casolorz

---

你需要在后台线程Scheduler上进行订阅（Schedulers类包含用于创建它们的静态工厂方法）。然后你需要在主线程Scheduler上进行观察，可以通过RxAndroid中的AndroidSchedulers.mainThread()方法获取。 - corsair992

我明白，但是哪个“调度程序”会给我一个后台线程，它将继续被重复使用，就像“trampoline”调度程序一样工作？ - casolorz

你可以创建一个线程池执行器(ThreadPoolExecuter)，并将其池大小设置为1 (Executors.newSingleThreadExecutor())，然后使用它通过Schedulers.from()方法生成调度器。或者，你可以创建一个带有运行Looper(例如HandlerThread)的自定义线程(Thread)，并将在该Looper上运行的Handler传递给AndroidSchedulers.handlerThread()方法，以获取将在该线程上操作的Scheduler。 - corsair992

有趣。谢谢，我会尝试一下这些。 - casolorz

Executors.newSingleThreadExecutor() 看起来正好符合我的需求。谢谢。如果您想将其放入答案中，我很乐意将其标记为已回答。 - casolorz

Answer 1

你可以通过以下方式之一创建一个可重复使用的线程来为Observable创建一个Scheduler：

• 创建一个线程池执行器ThreadPoolExecuter并将其池大小设置为1（Executors.newSingleThreadExecutor()是一个方便的静态工厂方法），然后使用它通过Schedulers.from()方法生成调度程序。
• RxAndroid提供了一个自定义的Scheduler实现，它使用一个Handler来调度操作，因此可以与任何运行有Looper的线程一起使用，只需将其Handler传递给AndroidSchedulers.handlerThread()工厂方法即可。

请注意，如果在这些任务结束时与UI进行交互，您需要观察主线程Scheduler。

- corsair992

---

我希望你能为我翻译一句话：「I miss a Schedulers.singleThread() method in rxjava, just like its newThread() one.」我在 RxJava 中想要一个 Schedulers.singleThread() 方法，就像它的 newThread() 方法一样。 - Alvaro Gutierrez Perez

RxJava 2有一个新的Schedulers.single()。 - Albert Vila Calvo

Answer 2

在RxJava 2中，你可以使用Schedulers.single()，它:

返回一个默认的、共享的、单线程支持的Scheduler实例，用于在同一后台线程上需要强顺序执行的工作。

请查看文档获取更多详细信息。

我在Schedulers文档中没有看到它在RxJava 1中可用。

- Albert Vila Calvo

---

但是 single() 不总是返回相同的调度程序吗？因此，如果您有两个使用 single() 的不同项，它们将共享相同的线程，对吧？因此，这并不能回答问题，因为线程不会专门用于一个作业。 - FrankMonza

当然，single() 方法始终返回相同的线程，因为你希望工作是按顺序的（这就是它的全部意义）。如果你想要两个独立的单线程调度程序，请使用Schedulers.from(Executor executor)，如在其他答案中所解释的。 - Albert Vila Calvo

Answer 3

您正在使用trampoline调度程序，这意味着您的源可观察对象将在当前线程（即主线程）上运行。

subscribeOn将对上游和下游都起作用。这就是为什么您的日志显示您正在运行main thread

为了解决这个问题，您可以在subscribeOn中使用Schedulers.single()，然后再在主线程中使用observeOn。

- Trieu Tran