java - RxJava : How to handle events in multiple threads

Question

例如，我有以下使用 RxJava 库的代码:

public class MultithreadingExample {
public static void main(String[] args) throws InterruptedException {
    Observable.from(Lists.newArrayList(1, 2, 3, 4, 5))
            .observeOn(Schedulers.computation())
            .map(numberToString())
            .subscribe(printResult());
    Thread.sleep(10000);
}

private static Func1<Integer, String> numberToString() {
    return number -> {
        System.out.println("Operator thread: " + Thread.currentThread().getName());
        return String.valueOf(number);
    };
}

private static Action1<String> printResult() {
    return result -> {
        System.out.println("Subscriber thread: " + Thread.currentThread().getName());
        System.out.println("Result: " + result);
    };
}

}

我希望事件在观察者中由多个线程处理，例如，项目“1”由线程 1 处理，项目“2”由线程 2 处理，依此类推。

使用 RxJava 实现此目的的最佳方法是什么？

Answer 1

您可以使用flatMap()运算符。

Observable.from(Lists.newArrayList(1, 2, 3, 4, 5))
        .flatMap( number -> Observable.defer( numberToString() )
                              .subscribeOn( Schedulers.computation() ) )
        .observeOn(Schedulers.computation())
        .map(numberToString())
        .subscribe(printResult());

flatMap() 运算符将在(可能是新的)线程上订阅新的 observable，将结果合并回最终 observeOn() 所在的线程完成了。