RxJava操作符怎么用

RxJava 是 ReactiveX 在 Java 平台上的实现，核心就是用操作符（Operators）把数据流（Observable / Flowable 等）进行转换、过滤、组合、调度。下面从基础概念 → 常用操作符 → 使用示例 → 实战建议来讲。

一、RxJava 基本模型（先理解这个）

Observable<T>
    .create / just / fromXxx   // 创建
    .map / filter / flatMap    // 操作符
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        data -> {},            // onNext
        error -> {},           // onError
        () -> {}               // onComplete
    );

三大角色

• 数据源（Observable / Flowable / Single / Maybe / Completable）
• 操作符（链式中处理数据）
• 观察者（subscribe）

二、常用操作符分类 + 示例

1️⃣ 创建类（数据从哪来）

just：直接发数据

Observable.just("A", "B", "C")
    .subscribe(s -> System.out.println(s));

fromIterable：集合转流

Observable.fromIterable(Arrays.asList(1, 2, 3))
    .subscribe(System.out::println);

create：手动发射

Observable.create(emitter -> {
    emitter.onNext(1);
    emitter.onNext(2);
    emitter.onComplete();
});

2️⃣ 转换类（数据变一变）

map：1 → 1

Observable.just("1", "2", "3")
    .map(Integer::parseInt)
    .subscribe(i -> System.out.println(i + 1));

flatMap：1 → 多（无序）

Observable.just("user1", "user2")
    .flatMap(user ->
        Observable.just(user + "_data1", user + "_data2")
    )
    .subscribe(System.out::println);

concatMap：1 → 多（保序）

.flatMap → 可能乱序
.concatMap → 按顺序

3️⃣ 过滤类（只留想要的）

filter

Observable.range(1, 10)
    .filter(x -> x % 2 == 0)
    .subscribe(System.out::println);

take / skip

.take(3)   // 只取前3个
.skip(2)   // 跳过前2个

distinct

.distinct() // 去重

4️⃣ 组合类（多个流）

merge：合并（并发）

Observable.merge(
    Observable.just("A"),
    Observable.just("B")
)
.subscribe(System.out::println);

zip：一对一组合

Observable.zip(
    Observable.just(1, 2),
    Observable.just("A", "B"),
    (num, str) -> num + str
).subscribe(System.out::println);

5️⃣ 线程调度（非常重要）

Observable.just(1)
    .subscribeOn(Schedulers.io())        // 上游线程
    .map(i -> i + 1)
    .observeOn(AndroidSchedulers.mainThread()) // 下游线程
    .subscribe(i -> updateUI(i));

常用 Scheduler

• Schedulers.io()：IO 操作
• Schedulers.computation()：计算
• AndroidSchedulers.mainThread()：UI（Android）

三、完整示例（网络 + UI）

api.getUser()
    .subscribeOn(Schedulers.io())
    .map(User::getName)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(
        name -> textView.setText(name),
        throwable -> toast("error")
    );

四、新手常见坑

❌ 不切线程 → 主线程卡死
❌ flatMap 乱序误用
❌ 忘记 dispose()（内存泄漏）

✅ 推荐用 CompositeDisposable

CompositeDisposable disposable = new CompositeDisposable();
disposable.add(observable.subscribe());
disposable.clear();

五、学习建议

1. 先掌握：map / flatMap / filter / just / subscribeOn
2. 再学：zip / merge / switchMap
3. 实战中用 Retrofit + RxJava 最佳

如果你愿意，可以告诉我：

• 你是 Android / 后端 / 入门？
• 想看 RxJava2 还是 RxJava3？
• 有没有具体场景（如登录、列表、轮询）？

我可以直接给你对应场景的 RxJava 写法。