Subject 是什么
RxJava 的 Subject 同时是 Observer 和 Observable：既能接收上游信号，也能把信号转发给下游。这和 Reactor 的 Sinks.Many 定位相同，但两者的并发契约不同，后面会单独讲。
RxJava 3 提供五种 Subject：
| Subject | 缓存策略 | 后订阅者 | 订阅者数 |
|---|---|---|---|
PublishSubject | 不缓存 | 丢失订阅前的元素 | 多个 |
BehaviorSubject | 最新 1 个 | 立即收到当前值 | 多个 |
ReplaySubject | 全部或有界 | 收到历史 | 多个 |
AsyncSubject | 最后 1 个 | 完成时才收到最后值 | 多个 |
UnicastSubject | 有界队列 | 只允许一个订阅者 | 1 个 |
PublishSubject：只转发新事件
import io.reactivex.rxjava3.subjects.PublishSubject;
import java.util.ArrayList;
import java.util.List;
PublishSubject<Integer> subject = PublishSubject.create();
subject.onNext(1); // 还没有订阅者
List<Integer> late = new ArrayList<>();
subject.subscribe(late::add);
subject.onNext(2);
subject.onComplete();
System.out.println("late=" + late);
实测输出：
late=[2]
1 在订阅前发射，直接丢失；后订阅者只看到 2 和完成信号。这与 Reactor 的 Sinks.many().multicast().directBestEffort() 最接近。
BehaviorSubject：总有「当前值」
BehaviorSubject.createDefault(v) 要求一个初始值；create() 则允许在首个值到来前为空。
BehaviorSubject<Integer> subject = BehaviorSubject.createDefault(0);
subject.onNext(1);
List<Integer> late = new ArrayList<>();
subject.subscribe(late::add);
subject.onNext(2);
System.out.println("late=" + late); // [1, 2]
后订阅者先收到当前值 1，再收到之后发射的 2。如果使用 create() 且尚未发射任何值，后订阅者什么也收不到（实测输出为空列表），直到第一个 onNext 到达。
Reactor 的对应写法是 Sinks.many().replay().limit(1)；需要初始值时用 replay().latestOrDefault(v)。区别在于 BehaviorSubject 总是有「当前值」这个概念，而 replay sink 的历史可以为空。
ReplaySubject：按数量或时间重放
ReplaySubject<Integer> subject = ReplaySubject.create(); // 无界
subject.onNext(1);
subject.onNext(2);
List<Integer> late = new ArrayList<>();
subject.subscribe(late::add);
subject.onNext(3);
System.out.println("late=" + late); // [1, 2, 3]
ReplaySubject.createWithSize(2) 只保留最近 2 个，实测发射 1、2、3 后订阅：
replay size 2: [2, 3]
还有 createWithTime(maxAge, unit) 和 createWithTimeAndSize(...)，分别按时间和双重上界保留历史。Reactor 对应 Sinks.many().replay().all()、limit(n)、limit(Duration)。
注意 ReplaySubject.create() 是无界的：对无限流使用它，内存会随元素数量持续增长。
AsyncSubject：完成时才发射最后一个值
AsyncSubject 的行为容易想当然。它在 onComplete() 之前不向下游发射任何元素，只记住最后一个值；onComplete() 时才把这个值发出去：
AsyncSubject<Integer> subject = AsyncSubject.create();
List<Integer> values = new ArrayList<>();
subject.subscribe(values::add);
subject.onNext(1);
subject.onNext(2);
System.out.println("before complete=" + values); // []
subject.onComplete();
System.out.println("after complete=" + values); // [2]
实测：
before complete=[]
after complete=[2]
完成后才订阅的观察者同样只收到 2。如果 onComplete() 前一个值都没有，它会正常完成但没有元素：
async empty: values=[] completed=true
Reactor 没有完全等价的 Sinks。Sinks.one() 也是单值 sink，但它在 tryEmitValue(v) 时就发射并终止，不等待「完成」这个额外信号。需要「完成时才取最后值」的语义，可以用 replay().latest() 配合 asFlux().last()：last() 会等源完成后再发出最后一个元素。
UnicastSubject：单订阅者 + 队列
UnicastSubject 只允许一个观察者，并且在观察者到来之前把元素缓冲起来：
UnicastSubject<Integer> subject = UnicastSubject.create();
subject.onNext(1);
subject.onNext(2);
List<Integer> late = new ArrayList<>();
subject.subscribe(late::add);
subject.onNext(3);
System.out.println("late=" + late); // [1, 2, 3]
第二个订阅者会在 onError 收到 IllegalStateException: Only a single observer allowed.。Reactor 对应 Sinks.many().unicast().onBackpressureBuffer()，零订阅者时同样接受发射并缓冲。
并发发射：Subject 不会替你串行化
一个常见的说法是「Subject 是线程安全的，可以从任意线程 onNext」。更准确的说法是：Subject 的订阅/取消是线程安全的，但 onNext 信号本身不会自动串行化。
用一个会睡眠的观察者做探针：两个线程各发射 10 次，观察者记录同时进入 onNext 的最大数量。
AtomicInteger inside = new AtomicInteger();
AtomicInteger maxConcurrent = new AtomicInteger();
subject.subscribe(new Observer<Integer>() {
@Override public void onSubscribe(Disposable d) { }
@Override public void onNext(Integer v) {
int now = inside.incrementAndGet();
maxConcurrent.accumulateAndGet(now, Math::max);
try { Thread.sleep(5); } catch (InterruptedException e) { }
inside.decrementAndGet();
}
@Override public void onError(Throwable e) { }
@Override public void onComplete() { }
});
Thread t1 = new Thread(() -> { for (int i = 0; i < 10; i++) subject.onNext(i); });
Thread t2 = new Thread(() -> { for (int i = 0; i < 10; i++) subject.onNext(i); });
t1.start(); t2.start();
t1.join(); t2.join();
实测：
裸 PublishSubject: maxConcurrentOnNext=2
PublishSubject.toSerialized(): maxConcurrentOnNext=1
裸 Subject 确实让同一个观察者的 onNext 并发执行了，这违反 Reactive Streams 的串行信号规则。修复方式是 subject.toSerialized()：它把所有发射串行化到内部队列，代价是每次发射多一层同步与排队。
这条规则对 BehaviorSubject、ReplaySubject、AsyncSubject 同样适用：需要多线程发射时，一律先 toSerialized()。
Reactor 的选择不同：Sinks.Many 默认检测并发发射并返回 FAIL_NON_SERIALIZED（见 9.1），不会静默地让两个线程同时进入下游。它也没有 toSerialized()，调用方必须自己保证串行，或者使用 Sinks.unsafe() 并自行承担后果。
Subject 与 Sinks 的映射
| RxJava | Reactor | 语义说明 |
|---|---|---|
PublishSubject | Sinks.many().multicast().directBestEffort() | 不缓存，只转发新元素 |
PublishSubject + 缓冲 | Sinks.many().multicast().onBackpressureBuffer(n) | 多播 + 有界缓冲 |
BehaviorSubject | Sinks.many().replay().limit(1) | 后订阅者补最新值 |
BehaviorSubject.createDefault(v) | Sinks.many().replay().latestOrDefault(v) | 带初始值 |
ReplaySubject.create() | Sinks.many().replay().all() | 无界历史 |
ReplaySubject.createWithSize(n) | Sinks.many().replay().limit(n) | 有界历史 |
ReplaySubject.createWithTime(...) | Sinks.many().replay().limit(Duration) | 按时间保留 |
AsyncSubject | 无完全等价；Sinks.one() 近似 | Sinks.one() 发射即完成，不等待显式完成 |
UnicastSubject | Sinks.many().unicast().onBackpressureBuffer() | 单订阅者队列 |
订阅者泄漏
Subject 会持有当前所有观察者的引用，直到 Disposable.dispose() 把它们移除。长生命周期的 Subject（单例事件总线、静态状态容器）配上忘记释放的短生命周期订阅者，就会持续累积：
// 每个请求都订阅，但从不释放
for (Request r : requests) {
subject.subscribe(event -> handle(r, event)); // Disposable 被丢弃
}
这些 handle 闭包连同 r 一起被 Subject 持有，请求对象无法回收。修复不是「少订阅几次」，而是把 Disposable 纳入生命周期管理：
CompositeDisposable perRequest = new CompositeDisposable();
perRequest.add(subject.subscribe(event -> handle(r, event)));
// 请求结束时
perRequest.dispose();
Reactor 的 Flux.subscribe() 返回的 Disposable 同理；Sinks 本身也不会自动清理订阅者，它只负责在取消时把它们从数组里移除。
选择建议
| 条件 | 建议 |
|---|---|
| Android 项目，已有 RxJava 依赖 | 继续用 Subject，注意 toSerialized() |
| Spring WebFlux / 服务端 | 用 Sinks，FAIL_NON_SERIALIZED 比静默并发更早暴露问题 |
需要 AsyncSubject 的「完成时取最后值」 | RxJava 直接用；Reactor 用 replay().latest() + last() |
| 需要单订阅者队列 | RxJava UnicastSubject；Reactor Sinks.many().unicast() |
| 多线程发射 | RxJava toSerialized()；Reactor 自己串行化 |
两个库的能力高度重叠，真正的差异在失败模型：RxJava Subject 在并发发射时没有统一的结果反馈，Reactor Sinks 用 EmitResult 把「没发出去」变成可检查的返回值。
小结
| 关注点 | 结论 |
|---|---|
| PublishSubject | 不缓存，后订阅者丢失历史 |
| BehaviorSubject | 后订阅者立即收到当前值；无值且未完成时为空 |
| ReplaySubject | 可无界、按数量或按时间保留历史 |
| AsyncSubject | 完成时才发射最后一个值；空完成则无元素 |
| UnicastSubject | 只允许一个观察者，订阅前缓冲 |
| 并发 | 裸 Subject 不串行化 onNext，必须 toSerialized() |
阅读导航
下一篇：10.2 ConnectableObservable 与 connect
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。