RxJava调度器-线程行为和饥饿？

Question

我对observeon()和subscribeon()使用了RXJava。我知道他们不是在一条流上并行化排放。换句话说，单个排放流只会放在一个线程上，对吗？我下面的测试似乎表明了这一点。我的理解是，您还必须flatmap()调度器，如.flatmap(v->observable.just(v).subscribeon(Schedulers.Computation())，以在单个流上并行化发射。

同样，如果是这种情况，那么调度器会发生线程饥饿吗？如果我的计算调度程序有5个线程，但我有超过5个长时间运行的异步流正在处理，有没有可能出现饥饿？或者这不太可能仅仅是因为RXJava的性质？

public class Test {
    public static void main(String[] args) {


        Observable<String> airports = Observable.just("ABQ", "HOU", 
            "PHX", "DAL", "DFW", "AUS","SAN","LAX","JFK");


        airports.subscribeOn(Schedulers.io()).map(Test::stall)
        .subscribe(s -> System.out.println("Sub1 " + s +
                " " + Thread.currentThread().getName()));

        airports.subscribeOn(Schedulers.io()).map(Test::stall)
        .subscribe(s -> System.out.println("Sub2 " + s +
                " " + Thread.currentThread().getName()));

        sleep();
    }

    private static String stall(String str) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return str;
    }

    private static void sleep() {
        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

Answer 1

使用flatmap时，可能一个异步源被其他异步源淹没，无法在自己的源上取得进展。然而，在实践中，我没有看到这种情况发生，这是由于OS和JVM的打嗝给了足够的喘息空间，也是因为flatmap本身的背压和仲裁。如果您担心这种情况太多，可以使用maxconcurrent参数和flatmap重载，并限制并发订阅的数量。

RxJava主要是以非阻塞的方式编写的，因此当需要合并或组合源代码时，它们不会真正等待对方。

计算调度程序是一个单线程执行程序池，并以循环方式分配给调用方。不知道标准执行者的公平性。