RxJava怎么处理线程

在 RxJava 里，“处理线程”主要靠 Scheduler（调度器） 和 操作符 来控制上游在哪执行、下游在哪执行。

下面按「核心概念 → 常用方式 → 常见场景」给你讲清楚。

一、核心概念：Scheduler（调度器）

RxJava 不会自动帮你切线程，必须显式指定。

常用 Scheduler

Schedulers.io()          // IO 密集型（网络、数据库）
Schedulers.computation() // CPU 密集型计算
Schedulers.newThread()   // 每次新建线程（不推荐频繁用）
AndroidSchedulers.mainThread() // Android 主线程（RxAndroid）
Schedulers.single()      // 单线程，顺序执行
Schedulers.trampoline()  // 当前线程（测试常用）

二、两个最关键的线程控制操作符

1️⃣ subscribeOn() —— 决定上游（被观察者）在哪执行

Observable.create(emitter -> {
    // 这里在 io 线程
})
.subscribeOn(Schedulers.io())

✅ 只生效一次（第一个 subscribeOn 有效）

2️⃣ observeOn() —— 决定下游（观察者）在哪执行

.observeOn(AndroidSchedulers.mainThread())
.subscribe(data -> {
    // 主线程更新 UI
});

✅ 可多次使用，每次切换后续线程

三、经典线程模型（Android 最常见）

网络请求 → 主线程更新 UI

Observable.fromCallable(() -> {
    return api.request(); // IO
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
    result -> updateUI(result),
    error -> showError(error)
);

四、多次切换线程示例

Observable.just(1)
    .subscribeOn(Schedulers.io())          // 上游 IO
    .map(i -> {
        // IO 线程
        return i * 2;
    })
    .observeOn(Schedulers.computation())   // 切到计算线程
    .map(i -> {
        // 计算线程
        return i + 1;
    })
    .observeOn(AndroidSchedulers.mainThread()) // 切回主线程
    .subscribe(i -> {
        // 主线程
    });

五、Flowable / Single / Maybe 也一样

Single.fromCallable(() -> download())
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> {}, throwable -> {});

六、容易踩的坑 ⚠️

❌ 1. 忘记切线程

.subscribeOn(Schedulers.io()) // 只在创建时有效

subscribeOn 放太晚没用

❌ 2. 在主线程做耗时操作

Observable.just(1)
    .map(i -> heavyWork()) // 主线程阻塞

❌ 3. 频繁 newThread

Schedulers.newThread() // 性能差

✅ 用 io() 或 computation()

七、一句话总结

RxJava 线程控制 = subscribeOn 定上游 + observeOn 切下游

如果你愿意，我可以：

• 给你画一张 线程切换流程图
• 讲 RxJava 在 Android / 后端的最佳实践
• 对比 Coroutine / Reactor 的线程模型

你现在是 Android 还是后端场景？