如何在並行線程中執行observable

Question

我有Observable調用消息，其中包含幾條消息。我想同時處理這些消息。我可以如何使用rxJava？

消息。（代碼以執行觀測的物品平行）.subscribe（MSG->處理（MSG））

（如果可觀察到的包含五個不同的消息然後我需要在五個單獨的線程來處理這五個消息）

Answer 1

如果你想留在Observable世界裏，你可以flatMap與subscribeOn和計算每個元素要在並行：

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

Answer 2

運行單個線程「觀察」您的消息，並將其中包含的每條消息分配給提交給某種workers thread pool的新消息處理任務（例如，簡單的Runnable）。

在這裏，你會發現簡單的操作方法：https://docs.oracle.com/javase/tutorial/essential/concurrency/executors.html