实现RxJava2 API小子集时，用ForkJoinPool.commonPool()替代计算调度器问询

针对你封装基于监听器的API到RxJava2 Flowable的场景，我们可以分两步解决问题：完善订阅取消时的监听器移除逻辑，以及显式指定使用ForkJoinPool.commonPool()作为调度器。

完整实现代码 #

import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;
import java.util.concurrent.ForkJoinPool;

public class EventBus {
    private final Flowable<Event> events;

    public EventBus(MyApi api) {
        // 1. 封装监听器到Flowable，同时处理取消订阅逻辑
        Flowable<Event> rawEvents = Flowable.create(emitter -> {
            Callback listener = new Callback() {
                @Override
                public void onEvent(Event event) {
                    emitter.onNext(event);
                }
            };
            api.addListener(listener);

            // 注册取消回调：当订阅被dispose时自动移除监听器
            emitter.setCancellable(() -> api.removeListener(listener));
        }, BackpressureStrategy.BUFFER); // 显式指定背压策略，根据业务场景调整

        // 2. 显式指定使用ForkJoinPool.commonPool()作为下游处理的调度器
        this.events = rawEvents.observeOn(Schedulers.from(ForkJoinPool.commonPool()));
    }

    // 对外提供订阅事件的入口
    public Flowable<Event> getEvents() {
        return events;
    }
}

关键细节解释 #

1. 监听器的自动移除
   通过emitter.setCancellable()注册的回调，会在订阅被dispose()时被RxJava自动触发，这样就能确保不再接收事件时及时从MyApi移除监听器，避免内存泄漏和无效的事件回调。

2. 背压策略的选择
   Flowable.create()必须指定背压策略，这里用BUFFER作为示例——它会缓存下游来不及处理的事件。如果你的业务对事件丢失不敏感，也可以选择DROP或LATEST，具体根据需求调整。

3. 用ForkJoinPool.commonPool()替代计算调度器

   • RxJava默认的Schedulers.computation()在Java 8+环境下本身就基于ForkJoinPool.commonPool()，但如果你想显式控制调度器（比如避免RxJava默认配置的变更），可以用Schedulers.from(ForkJoinPool.commonPool())把通用池包装成RxJava兼容的Scheduler。
   • observeOn()操作符的作用是让下游观察者的onNext/onComplete等回调在指定的调度器线程上执行，这是最常见的场景——把事件处理逻辑切换到计算线程池。
   • 如果你的api.addListener(listener)是阻塞操作，需要在后台线程执行，可以在Flowable.create()之后添加subscribeOn(Schedulers.from(ForkJoinPool.commonPool()))，注意subscribeOn()只会影响订阅阶段的线程（也就是addListener的执行线程）。

额外注意事项 #

• 确保MyApi的addListener和removeListener方法是线程安全的，因为dispose()可能在任意线程触发removeListener。
• 如果MyApi本身就在后台线程触发onEvent回调，评估是否需要额外的线程切换，避免不必要的性能开销。

内容的提问来源于stack exchange，提问作者Martín Coll