10.3 RxJava 3 与 Reactor 的迁移对照

类型映射

Observable 和 Flux 看起来都是「0..N 个元素」，但这是迁移中最容易出事的一对：Observable 没有请求协议，Flux 有。下面的陷阱 1 会展开。

操作符映射

有两个常见错误值得单独指出：

• throttleFirst 不能映射成 sample。sample 每个窗口取最后一个，与 throttleLast 相同；throttleFirst 取的是窗口内第一个，Reactor 没有现成操作符。
• Reactor 的 doOnEach 一直存在，不需要拆成 doOnNext + doOnError + doOnComplete。

调度器映射

trampoline 与 immediate 的差别在于：trampoline 把任务排入当前线程的队列，按顺序排空，能避免递归；immediate 直接在调用线程执行，递归仍然存在。如果原代码依赖 trampoline 的「排队」语义，迁移到 immediate 会改变调用顺序。

陷阱 1：Observable 没有背压

RxJava 的 Observable 不传递请求量。observeOn 用一个可增长的队列把元素从源线程搬到目标线程；下游慢，队列就涨。

实测：Observable.range(1, 1000) 配 observeOn(computation, delayError=false, bufferSize=1)，下游每个元素睡眠 10ms：

got=228

3 秒内没有任何异常，队列持续吸收元素。bufferSize=1 只是初始容量，不构成硬上限。这意味着长时间运行的 Observable 加慢消费者，风险是内存增长，而不是一个可捕获的 MissingBackpressureException。

Reactor 的 publishOn 不同：它向上游请求有界预取量，下游消费多少，上游才被允许生产多少。慢消费者会让上游变慢，而不是让队列无限增长。如果源本身无法遵守请求（例如回调式数据源），Reactor 要求你显式选择溢出策略，例如 onBackpressureBuffer、onBackpressureDrop 或 onBackpressureError，失败会以 OverflowException 暴露。

迁移时先问：这个 Observable 的上游是否真的能降速？不能的话，换成 Flux 之后要么补一个溢出策略，要么保持 Observable 的语义（用 Flux.create(..., OverflowStrategy.DROP) 之类显式表达）。

陷阱 2：onErrorReturn 没有 Function 重载

RxJava 允许根据异常计算兜底值：

observable.onErrorReturn(throwable -> "fallback-" + throwable.getClass().getSimpleName());

Reactor 的 onErrorReturn 只有三种重载：

flux.onErrorReturn("fallback");                                   // 固定值
flux.onErrorReturn(IOException.class, "fallback");                // 按类型
flux.onErrorReturn(t -> t instanceof IOException, "fallback");    // 按谓词

注意第三种是「谓词 + 固定值」，不是「异常 → 值」。要计算兜底值，用 onErrorResume：

flux.onErrorResume(t -> Mono.just("fallback-" + t.getClass().getSimpleName()));

陷阱 3：Subject 的并发契约不同

RxJava 的裸 Subject 不会串行化 onNext。两个线程同时发射时，同一个观察者的 onNext 可能并发执行；实测最大并发数为 2。必须用 toSerialized() 包一层：

Subject<Integer> safe = PublishSubject.<Integer>create().toSerialized();

Reactor 的 Sinks.Many 默认检测并发发射，返回 FAIL_NON_SERIALIZED，并且 emitNext 在遇到该结果时抛 EmissionException。它不会替调用方排队，也不会静默并发进入下游：

Sinks.EmitResult result = sink.tryEmitNext(v);
if (result == Sinks.EmitResult.FAIL_NON_SERIALIZED) {
    // 说明有并发发射，需要重新设计串行化
}

迁移时不要假设「Subject 能做的事 Sinks 也能直接做」：Sinks 把并发问题变成了显式失败，调用方必须处理。

陷阱 4：Disposable 是两种类型

方法名都是 dispose()，但类型不兼容。跨库抽象时不能共用一个字段，需要各自的适配层。RxJava 还有 CompositeDisposable；Reactor 没有同名工具，通常用 Disposables.composite() 或自己维护集合。

陷阱 5：flatMap 的默认并发数相差极大

RxJava 的 flatMap(mapper) 默认 maxConcurrency = Integer.MAX_VALUE（字节码里的常量 2147483647），也就是不限制同时订阅的内部流数量。

Reactor 的 flatMap(mapper) 默认 concurrency = Queues.SMALL_BUFFER_SIZE，即 256；内部预取量为 XS_BUFFER_SIZE（默认 32）。

对一个会产生大量内部流的源，这个差异直接决定并发压力：

// RxJava：可能同时订阅几千个内部流
source.flatMap(this::fetch);

// Reactor：最多 256 个
source.flatMap(this::fetch);

// 显式对齐
source.flatMap(this::fetch, 1024);

迁移后如果发现吞吐下降，先检查这里，而不是先怀疑调度器。

陷阱 6：Completable 没有元素，但完成信号要对齐

// RxJava
Completable.fromAction(() -> doWork()).subscribe();

// Reactor
Mono.fromRunnable(() -> doWork()).subscribe();

Mono.fromRunnable 推断为 Mono<T>，赋给 Mono<Void> 时由目标类型决定；也可以写 Mono.<Void>fromRunnable(...)。两者都在完成时发 onComplete，错误时发 onError。

如果原代码用 Completable.andThen(...) 串联，Reactor 对应 Mono.then(...)：

// RxJava
Completable.fromAction(this::prepare).andThen(load());

// Reactor
Mono.fromRunnable(this::prepare).then(load());

迁移检查清单

• 把 Observable 逐项确认：上游能否降速？不能就补溢出策略。
• Single / Maybe → Mono，Completable → Mono<Void>。
• onErrorReturn(Function) → onErrorResume。
• retryWhen(Function) → Retry.withThrowable(...) 或 Retry.backoff(...)。
• observeOn → publishOn，doOnDispose → doOnCancel。
• Subject 的 toSerialized() 是否还需要？Sinks 侧是否处理了 FAIL_NON_SERIALIZED？
• flatMap 是否需要显式 concurrency？
• 调度器：io → boundedElastic，computation → parallel，trampoline 需要复核语义。
• 用 StepVerifier 重写 TestObserver 断言。

性能比较不要照搬网上的数字

RxJava 与 Reactor 的吞吐差异高度依赖操作符融合、背压模式、调度器和 JVM 版本。脱离 JMH 的微基准很容易得出相反结论；不同机器、不同 JDK 上的绝对值也没有可比性。

如果确实需要比较，至少做到：

1. 用 JMH，设置预热与足够的迭代；
2. 固定 JDK、GC、CPU 亲和性和数据规模；
3. 分别测「无背压的 Observable」与「有背压的 Flux」，不要混为一谈；
4. 报告分位数与错误率，而不只是平均吞吐。

没有这四条，任何「Reactor 快 20%」的结论都不值得写进选型文档。

何时保留 RxJava

何时迁移到 Reactor

小结

阅读导航

上一篇：10.2 ConnectableObservable 与 connect

下一篇：11.1 RS TCK：测试环境搭建