java如何在并行线程中执行observable

Question

我有一个可观察的消息，它包含几个消息。我想同时处理这些消息。如何使用rxJava实现这一点

信息。（并行执行可观察项的代码）。订阅（消息->；处理（消息））

（如果observable包含五条不同的消息，那么我需要在五条不同的线程中处理这五条消息）

Answer 1

如果您想留在Observable世界中，您可以flatMap使用subscribeOn和您想要并行计算的每个元素：

Observable.range(1, 10)
.flatMap(v -> 
    Observable.fromCallable(() -> compute(v))
    .subscribeOn(Schedulers.computation)
)
.subscribe(e -> { }, Throwable::printStackTrace);

Answer 2

运行“观察”消息的单线程，并将它包含的每个消息分派到新的消息处理任务（例如简单的Runnable），该任务将提交给某种类型的workers thread pool

在这里，您可以找到简单的操作方法：https://docs.oracle.com/javase/tutorial/essential/concurrency/executors.html