RxJava

RxJava 是什么

RxJava 是 Reactive Extensions 在 JVM 上的实现，用可观察序列来组合异步与事件驱动的程序。它把观察者模式扩展到数据/事件序列，并提供大量操作符，让你以声明式方式组合序列，同时屏蔽底层线程、同步、线程安全与并发数据结构等细节。

核心类型

RxJava 4 的组件位于 io.reactivex.rxjava4 包下，基础类与接口位于 io.reactivex.rxjava4.core：

• Flowable：0..N 个元素，支持 Reactive-Streams 与背压
• Observable：0..N 个元素，不支持背压
• Single：恰好 1 个元素或一个错误
• Completable：无元素，只有完成或错误信号
• Maybe：0 或 1 个元素，或一个错误
• Streamable：基于虚拟线程阻塞与 CompletionStage 状态机，天然具备背压

快速上手

以 Gradle 引入依赖（版本号请替换为最新）：

implementation "io.reactivex.rxjava4:rxjava:4.x.y"

一个最小示例：

import io.reactivex.rxjava4.core.*;

public class HelloWorld {
    public static void main(String[] args) {
        Flowable.just("Hello world").subscribe(System.out::println);
    }
}

关键概念

上游与下游

数据流由源、若干中间操作符和消费者组成。以某个操作符为参照，向左朝源的方向是上游，向右朝订阅者的方向是下游。

source
  .operator1()
  .operator2()
  .operator3()
  .subscribe(consumer);

背压

当数据流经过异步步骤时，各步骤处理速度不同。为避免下游被压垮（表现为缓冲导致内存上涨或丢数据），背压让步骤声明自己还能处理多少元素，从而约束内存占用。Flowable 专门支持背压，Observable 用于无需背压的短序列与 GUI 交互等场景。

装配期与订阅期

• 装配期：应用各种中间操作符准备数据流，此时数据尚未流动、无副作用。
• 订阅期：调用 subscribe() 建立处理链，触发订阅副作用（如 doOnSubscribe）。
• 运行期：数据流真正开始发射元素、错误或完成信号。

4.x 版本要点

• 原生 Java 实现，运行时不需要第三方库
• 基于 java.util.concurrent.Flow，保留 JPMS 与 OSGi 支持
• 支持虚拟线程：virtualCreate()、virtualTransform()、Schedulers.virtual()
• 新增 Streamable<T>，可理解为 Java 版的 IAsyncEnumerable
• 使用 Java Cleaner API 检测资源泄漏并做自适应清理
• 通过 Reactive Streams TCK 兼容性测试

注意：4.x 对 JDK 版本要求较高，Android 兼容性取决于 API level 与可用的脱糖方案；3.x 的支持将在 4.x 正式发布后逐步减弱。

适合谁用

如果你在 Android 或 Java 服务端需要处理异步事件、组合多个数据源、控制并发与背压，RxJava 是经过大规模验证的选择。代价是概念较多、学习曲线偏陡，建议先从 Flowable/Observable 与常用操作符入手。