java - 如何观察java rx中的调用线程？

Question

有没有办法告诉 java rx 在 observeOn 函数中使用当前线程？我正在为 android 同步适配器编写代码，我希望在同步适配器线程中而不是在主线程中观察结果。

使用 Retrofit + RX Java 的示例网络调用如下所示：

MyRetrofitApi.getInstance().getObjects()
.subscribeOn(Schedulers.io())
.observeOn(<current_thread>)
.subscribe(new Subscriber<Object>() {
    //do stuff on the sync adapter thread

}

我尝试使用

...
.observeOn(AndroidSchedulers.handlerThread(new Handler(Looper.myLooper())))
...

这与 android rx 为主线程创建调度程序的方式相同，但一旦我将 Looper.myLooper() 替换为 Looper.getMainLooper()，它就不再工作了。

我可以使用 Schedulers.newThread()，但由于它与大量服务器调用的复杂同步代码，我会不断创建一个新线程来触发新的网络调用，该调用再次创建新线程以启动更多网络调用。有没有办法做到这一点？还是我的方法本身完全错误？

Answer 1

尝试使用Schedulers.immediate()

MyRetrofitApi.getInstance().getObjects()
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.immediate())
.subscribe(new Subscriber<Object>() {
    //do stuff on the sync adapter thread

}

它的描述是：Creates and returns a Scheduler that executes work immediately on the current thread.

注意：
我认为将所有工作保留在 SyncAdapter 的线程上是可以的，因为它已经在使用不同的线程

Answer 2

哦，我刚刚在 wiki 上找到了这个：https://github.com/ReactiveX/RxAndroid#observing-on-arbitrary-threads

new Thread(new Runnable() {
    @Override
    public void run() {
        final Handler handler = new Handler(); // bound to this thread
        Observable.just("one", "two", "three", "four", "five")
                .subscribeOn(Schedulers.newThread())
                .observeOn(HandlerScheduler.from(handler))
                .subscribe(/* an Observer */)

        // perform work, ...
    }
}, "custom-thread-1").start();

我认为这也适用于您的情况 - 当然，除了创建一个新线程......所以只是：

final Handler handler = new Handler(); // bound to this thread
MyRetrofitApi.getInstance().getObjects()
    .subscribeOn(Schedulers.io())
    .observeOn(HandlerScheduler.from(handler))
    .subscribe(new Subscriber<Object>() {
        //do stuff on the sync adapter thread

    }