java - RxJava Rx 默认是单线程的。真的吗？

Question

嗨，我发现很少有手册指出:Rx 默认情况下是单线程的。

您可以在这里找到这个短语https://github.com/Froussios/Intro-To-RxJava/blob/master/Part%204%20-%20Concurrency/1.%20Scheduling%20and%20threading.md但这不仅仅是一个获取它的地方。

我试着用它玩了一下，实际上它是多线程的。这是我尝试过的简单代码:

    Observable<Long> values1 = Observable.interval(1000, TimeUnit.MILLISECONDS);
    values1.subscribe(
        v -> {
            System.out.println("Received 1: " + v + ", Thread: " + Thread.currentThread().getName());
        },
        e -> System.out.println("Error: " + e),
        () -> System.out.println("Completed")
    );

    Observable<Long> values2 = Observable.interval(1000, TimeUnit.MILLISECONDS);
    values2.subscribe(
        v -> {
            System.out.println("Received 2: " + v + ", Thread: " + Thread.currentThread().getName());
        },
        e -> System.out.println("Error: " + e),
        () -> System.out.println("Completed")
    );

    values1.subscribe(
            v -> {
                System.out.println("Received 3: " + v + ", Thread: " + Thread.currentThread().getName());
            },
            e -> System.out.println("Error: " + e),
            () -> System.out.println("Completed")
        );

    System.in.read();

这里只有两个 observable 和 3 个 reader。此代码输出:

Received 1: 0, Thread: RxComputationThreadPool-1
Received 3: 0, Thread: RxComputationThreadPool-3
Received 2: 0, Thread: RxComputationThreadPool-2
Received 1: 1, Thread: RxComputationThreadPool-1
Received 2: 1, Thread: RxComputationThreadPool-2
Received 3: 1, Thread: RxComputationThreadPool-3
...

正如我从输出中看到的那样，每个读者都会使用新线程，并且我从未明确提到为每个新观察者创建新线程。那么这是单线程的默认行为吗？我是不是错过了什么地方？

Answer 1

RxJava 是单线程的，它有两个部分:

1. 大多数操作都是按顺序执行的。这意味着当源产生一个值时，它会立即由​​链中的下一步处理，除非:
   • 该运算符是异步运算符之一，需要延迟和线程才能工作。示例包括observeOn和interval 。由于您的示例使用 interval这需要真正的单线程阻塞，因此使用更高效的基于调度程序的解决方案。
   • 从内存中来看，唯一使用阻塞的运算符是从使用阻塞的其他格式转换而来的运算符，例如 Callable<T>和Future<T> .
2. 大多数 RxJava 标准和运算符都假设输入是顺序的，这意味着当您实现自己的可观察对象时，您需要确保调用 onNext不要同时发生。这大概就是“RxJava 是单线程的”规则所基于的，因为它本身不支持并行操作。 RxJava2添加 ParallelObservable支持真正并行性的操作系列。