RxJava 执行顺序混乱 [英] RxJava order of execution confusion

问题描述

我有一个非常简单的 RxJava 示例

I have this very simple RxJava example

List<Integer> arrayIntegers = new ArrayList<>(Arrays.asList(1, 2, 3, 4, 5));

Observable.fromIterable(arrayIntegers).map(i -> {
    Log.d("RxJava", "map i = " + i);
    return i;
}).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).
subscribe(new DisposableObserver<Integer>() {
    @Override
    public void onNext(Integer i) {
        Log.d("RxJava", "next i = " + i);
    }

    @Override
    public void onError(Throwable e) {}

    @Override
    public void onComplete() {
        Log.d("RxJava", "Completed");
    }
});

这给出了这个结果..

D/RxJava: map i = 1
D/RxJava: map i = 2
D/RxJava: map i = 3
D/RxJava: map i = 4
D/RxJava: map i = 5
D/RxJava: next i = 1
D/RxJava: next i = 2
D/RxJava: next i = 3
D/RxJava: next i = 4
D/RxJava: next i = 5

我所期待的是更像这样的东西

What I was expecting though is something more like this

D/RxJava: map i = 1
D/RxJava: next i = 1
D/RxJava: map i = 2
D/RxJava: next i = 2
D/RxJava: map i = 3
D/RxJava: next i = 3
D/RxJava: map i = 4
D/RxJava: next i = 4
D/RxJava: map i = 5
D/RxJava: next i = 5

有人能解释一下我做错了什么导致我的订单不正确吗?

Could someone explain what I am doing wrong which is causing my order to be incorrect?

推荐答案

您的 Rx 管道在两个不同的线程中运行.第一个线程(Schedulers.newThread())执行fromIterable 和map.第二个调用 DisposableObserver.

Your Rx pipeline operates in two different threads. The first thread (Schedulers.newThread()), does the fromIterable and map. The second calls DisposableObserver.

在两个线程之间，observeOn() 操作符中有(一个不可见的)缓冲区.所以发生的事情是第一个线程获得一些 cpu 时间，快速完成所有工作并将结果放入 observeOn 缓冲区.这可能发生在该线程在 CPU 上的单次旋转期间.

Between the two threads there's (an invisible) buffer inside the observeOn() operator. So what happens is that the first thread gets some cpu time, quickly does all of its work and puts the results in the observeOn buffer. This probably happens during a single spin of that thread on the CPU.

然后，第二个线程启动，它从缓冲区中取出项目并通过调用您的 DisposableObserver 继续处理.

Then, the second thread is launched, it takes the items from the buffer and continues the processing by calling your DisposableObserver.

我认为您在这里寻找的是使用单线程.删除 observeOn 操作符，看看它是否做你想要它做的事情.

I think that what you're looking for here is using a single thread. Remove the observeOn operator, see if it does what you want it to do.

刚刚尝试删除 observeOn 操作符并在我的机器上运行它.结果如下:

Just tried removing the observeOn operator and running it on my machine. Here's the result:

RxJava map i = 1
RxJava next i = 1
RxJava map i = 2
RxJava next i = 2
RxJava map i = 3
RxJava next i = 3
RxJava map i = 4
RxJava next i = 4
RxJava map i = 5
RxJava next i = 5

为了将来参考，打印到达各种 Rx 操作符的线程的名称可能对调试有用.您可以使用 Thread.currentThread().getName() 获取名称.

For future reference, it might be useful for debugging purposes to print the name of the thread that reaches the various Rx operators. You can get the name using Thread.currentThread().getName().

这篇关于RxJava 执行顺序混乱的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！