RxJava的核心概念
RxJava是一种基于响应式编程思想的异步编程库,其核心是“观察者模式”和“链式调用”，要理解RxJava，首先要掌握三个基本角色：被观察者（Observable）、观察者（Observer）和订阅（Subscribe），被观察者负责发出事件，观察者负责接收并处理事件，订阅则是两者建立连接的过程，通过这种模式，RxJava将异步操作封装成事件流，使代码逻辑更清晰、易于维护。
创建与订阅：事件流的起点
在RxJava中,创建事件流通常从Observable开始。Observable可以创建多种类型的事件流：普通事件（just）、数组/集合事件（fromArray）、异步事件（create）等。Observable.just("Hello", "RxJava")会发出两个字符串事件。
订阅操作通过subscribe()方法实现，该方法可以传入观察者的不同回调函数：onNext()处理正常事件，onError()处理异常，onComplete()在事件流结束时调用。
Observable.just("Hello", "RxJava")
.subscribe(new Observer<String>() {
@Override
public void onNext(String s) { System.out.println(s); }
@Override
public void onError(Throwable e) { e.printStackTrace(); }
@Override
public void onComplete() { System.out.println("Completed"); }
});
这段代码会依次输出“Hello”“RxJava”和“Completed”。
转换与过滤：事件流的加工处理
RxJava的强大之处在于丰富的操作符,它们可以对事件流进行转换、过滤、合并等处理，常用操作符包括：
- map：转换事件类型，将
Observable<Integer>转换为Observable<String>：map(i -> "Number: " + i)。 - filter：过滤事件，只保留偶数：
filter(i -> i % 2 == 0)。 - flatMap：将一个事件转换为多个事件，并合并结果，常用于异步嵌套场景，如网络请求后解析数据。
- concatMap：与
flatMap类似，但按顺序发送事件，避免数据错乱。
通过操作符的组合,可以灵活处理复杂逻辑，先过滤数据再转换类型：
Observable.just(1, 2, 3, 4)
.filter(i -> i % 2 == 0)
.map(i -> "Even: " + i)
.subscribe(System.out::println);
// 输出：Even: 2, Even: 4
线程调度：异步任务的掌控
RxJava通过Scheduler实现线程控制，解决Android开发中的主线程卡顿问题，常用的调度器包括：
- Schedulers.io()：适用于IO密集型任务（如网络请求、文件读写）。
- AndroidSchedulers.mainThread()：将任务切换到Android主线程，用于更新UI。
- Schedulers.newThread()：创建新线程执行任务。
线程切换通过subscribeOn()和observeOn()方法实现：subscribeOn()指定被观察者执行线程，observeOn()指定观察者执行线程。
Observable.fromCallable(() -> fetchDataFromNetwork())
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(result -> textView.setText(result));
这段代码在IO线程获取网络数据,再在主线程更新UI，避免阻塞。
生命周期管理：防止内存泄漏
在Android中,若观察者未及时取消订阅，可能导致内存泄漏，RxJava提供Disposable管理订阅关系，通过CompositeDisposable统一管理多个Disposable，在Activity/Fragment销毁时调用clear()或dispose()取消订阅。
private CompositeDisposable disposables = new CompositeDisposable();
// 添加订阅
disposables.add(Observable.interval(1, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(t -> Log.d("Timer", "Tick: " + t)));
// 销毁时取消订阅
@Override
protected void onDestroy() {
super.onDestroy();
disposables.dispose();
}
RxJava通过观察者模式和操作符链,将复杂的异步逻辑转化为清晰的事件流处理，掌握创建订阅、操作符使用、线程调度和生命周期管理，是高效运用RxJava的关键，合理使用RxJava，不仅能提升代码可读性，还能有效避免异步编程中的常见问题。
