RxJava—一个生产者，多个并发消费者，在单个订阅中

Question

我正在尝试了解RxJava并发的一些细节，但我不确定我的想法是否正确。我对SubscribeOn/观察的工作原理有很好的了解，但我正在尝试确定池调度程序的一些细节。为此，我正在考虑尽可能简单地实现一个1-N生产者-消费者链，其中消费者的数量与CPU一样多。

根据文档，Schedulers.computation（）由与内核一样多的线程池支持。但是，根据Reactive合约，运算符只能获得顺序调用。

因此，像这样的设置

Observable.range(1, 1000) // Whatever has to be processed
            .observeOn(Schedulers.computation())
            .doOnNext(/* heavy computation */)
            .doOnCompleted(() -> System.out.println("COMPLETED"))
            .forEach(System.out::println);

尽管使用了线程池，但只会收到对doOnNext的并发调用。对睡眠和检查操作观察On.java的实验似乎证实了这一点，因为每个观察调用都会获得一个工作器。此外，如果不是这样，应该会有一个复杂的OnComplated管理，必须等待任何挂起的OnNext完成，我发现不存在。

假设我在这里走对了（也就是说，只涉及一个线程，尽管你可以用observeOn跳过其中几个线程），那么正确的模式是什么？我可以找到相反情况的示例（将多个异步事件生成器同步到一个使用者中），但这不是一个简单的示例。

我想flatMap也参与其中，可能是使用了限制并发订阅数量的beta版（1.x版）。也许可以像这样简单地使用窗口/平面图？

Observable
.range(1, 1000) // Whatever has to be processed
.window(1) // Emit one observable per item, for example 
.flatMap(/* Processing */, 4) // For 4-concurrent processing
.subscribe()

在这种方法中，我仍然缺少一种以Rx通用方式最大化CPU的简单方法（即，指定计算调度器，而不是flatMap的最大订阅数）。所以，也许……：

Observable
.range(1, 1000) // Whatever has to be processed
.window(1) // Emit one observable per item, for example 
.flatMap(v -> Observable.just(v)
                        .observeOn(Schedulers.computation())
                        .map(/* heavy parallel computation */))
.subscribe()

最后，在一些使用flatMap的示例中，我看到在flatMap之后有一个调用，我不知道为什么需要这个调用，因为flatMap不应该为下游执行序列化吗？（例如，在本例中：http://akarnokd.blogspot.com.es/2016/02/flatmap-part-1.html)

Answer 1

Thomas Nield有篇很好的文章就是关于那个案子的

RxJava-最大化并行化

我个人在这种情况下所做的，我只是在具有最大并发调用参数的平面地图中订阅Schedulers.io。

    Observable.range(1, 1000) // Whatever has to be processed
            .flatMap(v -> Observable.fromCallable(() -> { /* io calls */}) 
                    .subscribeOn(Schedulers.io()), Runtime.getRuntime().availableProcessors() + 1)
            .subscribe();

根据评论中的建议进行编辑。最好使用调度程序。计算（）用于CPU限制的工作

    Observable.range(1, 1000) // Whatever has to be processed
            .flatMap(v -> Observable.fromCallable(() -> { /* intense calculation */}) 
                    .subscribeOn(Schedulers.computation()))
            .subscribe();