RxJava多线程事件处理：如何让观察者用不同线程处理各事件？

嘿，这个需求很明确！你当前的代码用了Schedulers.computation()调度器，它的线程池是固定大小（默认等于CPU核心数）的，线程会被复用，所以没法做到每个事件分配一个全新的线程。要实现每个事件由不同线程处理，我们可以通过**flatMap结合Schedulers.newThread()**来实现，这是最直接也最符合RxJava设计思路的方式。

为什么原来的代码做不到？ #

先理清楚核心问题：observeOn(Schedulers.computation())只是指定了下游操作（map和subscribe）的执行线程，但这个线程是从computation线程池里复用的——也就是说，多个事件会轮流使用池子里的线程，不会为每个事件新建线程。

修改后的代码实现 #

下面是调整后的代码，能让每个事件的处理都在独立线程中执行：

public class MultithreadingExample {
    public static void main(String[] args) throws InterruptedException {
        Observable.from(Lists.newArrayList(1, 2, 3, 4, 5))
                // 用flatMap把每个事件转换成独立的Observable，每个都在新线程执行
                .flatMap(number -> 
                    Observable.just(number)
                            .subscribeOn(Schedulers.newThread()) // 为每个事件分配新线程
                            .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);
        };
    }
}

代码解释 #

• flatMap的作用：它会把原始Observable的每个事件，转换成一个单独的新Observable。这样我们就能为每个新Observable单独指定执行线程。
• subscribeOn(Schedulers.newThread())：这个操作符会让当前Observable的上游操作（这里就是map）在新创建的线程中执行。Schedulers.newThread()会为每个订阅创建一个全新的线程，执行完就销毁，正好满足你每个事件用不同线程的需求。
• 如果你希望订阅者（subscribe中的逻辑）也在不同线程执行，可以在flatMap的内部Observable后面再加一个observeOn(Schedulers.newThread())，不过一般订阅者的逻辑如果比较轻量，复用线程也没问题。

注意事项 #

• Schedulers.newThread()会频繁创建和销毁线程，如果你的事件量非常大（比如成百上千个），可能会带来线程创建的开销，这种情况下可以考虑自定义一个有界的线程池调度器，平衡线程数量和处理效率。
• 一定要确保主线程等待足够长的时间（比如代码里的Thread.sleep(10000)），否则主线程提前退出，所有子线程都会被终止，看不到完整的输出。

内容的提问来源于stack exchange，提问作者evgeniy44