如何根据条件为RxJava的Observable设置subscribeOn调度器

核心思路 #

subscribeOn 用于指定Observable上游任务的执行线程，要实现条件触发调度器切换，只需根据判断逻辑动态选择是否添加subscribeOn，或切换不同的调度器即可。

方案一：直接在flatMap内做条件判断 #

这是最直观的实现方式，适合简单场景：

// 自定义判断条件，可根据业务动态设置
boolean useNewThreadScheduler = true;

Observable.just("One", "Two", "Three")
    .flatMap(v -> {
        // 先构建基础操作链
        Observable<String> operation = performLongOperation(v)
                .doOnNext(s -> System.out.println("processing item on thread: " + Thread.currentThread().getName()));
        
        // 根据条件选择调度器
        if (useNewThreadScheduler) {
            return operation.subscribeOn(Schedulers.newThread());
        } else {
            // Android环境用AndroidSchedulers.mainThread()；普通Java环境用Schedulers.trampoline()（当前线程）
            return operation.subscribeOn(AndroidSchedulers.mainThread());
        }
    })
    .subscribe(item -> System.out.println("Received item: " + item));

方案二：封装工具方法复用逻辑 #

如果多处需要这种条件调度逻辑，建议封装成ObservableTransformer工具类，提高复用性：

// 封装条件调度的Transformer
public static <T> ObservableTransformer<T, T> conditionalSubscribeOn(
        boolean condition, 
        Scheduler targetScheduler, 
        Scheduler fallbackScheduler
) {
    return upstream -> condition ? upstream.subscribeOn(targetScheduler) : upstream.subscribeOn(fallbackScheduler);
}

// 使用示例
Observable.just("One", "Two", "Three")
    .flatMap(v -> performLongOperation(v)
            .doOnNext(s -> System.out.println("processing item on thread: " + Thread.currentThread().getName()))
            // 直接通过compose复用工具方法
            .compose(conditionalSubscribeOn(useNewThreadScheduler, Schedulers.newThread(), AndroidSchedulers.mainThread()))
    )
    .subscribe(item -> System.out.println("Received item: " + item));

注意事项 #

• subscribeOn的调用位置不影响最终线程（它仅决定Observable上游的执行线程），习惯上放在操作链的最前端
• 普通Java环境没有AndroidSchedulers时，若要指定主线程，可使用Schedulers.from(Executors.newSingleThreadExecutor())创建专属线程池，或用Schedulers.trampoline()在当前订阅线程执行
• 不要混淆subscribeOn和observeOn：前者控制上游任务执行线程，后者控制下游订阅回调的线程，本场景需用subscribeOn

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