ReplaySpec：先选策略，再建 sink
Sinks.many().replay() 返回的是 MulticastReplaySpec，不是 sink 本身。必须先选一种历史保留策略：
// 全部历史，无界
Sinks.Many<Integer> all = Sinks.many().replay().all();
// 最近 n 个
Sinks.Many<Integer> last3 = Sinks.many().replay().limit(3);
// 最近 ttl 内
Sinks.Many<Integer> recent = Sinks.many().replay().limit(Duration.ofMinutes(5));
// 同时受数量与时间限制
Sinks.Many<Integer> bounded = Sinks.many().replay().limit(3, Duration.ofMinutes(5));
// 只保留最新一个
Sinks.Many<Integer> latest = Sinks.many().replay().latest();
没有无参的 replay()。这个设计是刻意的：历史保留一定伴随内存代价，API 不提供「默认无界」这种容易误用的入口。
limit(n)：只给后订阅者最近 n 个
Sinks.Many<Integer> sink = Sinks.many().replay().limit(3);
sink.tryEmitNext(1);
sink.tryEmitNext(2);
sink.tryEmitNext(3);
sink.tryEmitNext(4); // 1 被挤出历史
List<Integer> late = new ArrayList<>();
sink.asFlux().subscribe(late::add);
System.out.println("late=" + late);
实测输出：
late=[2, 3, 4]
后订阅者拿到的是最近 3 个，不是「从订阅点开始」：4 是订阅之前发射的，但它仍在历史里，所以也会补给后订阅者。这与 publish() 的 Hot 语义不同——replay 的历史对后订阅者是可见的。
历史在 sink 完成之后仍然保留：
sink.tryEmitComplete();
List<Integer> afterComplete = new ArrayList<>();
boolean[] completed = {false};
sink.asFlux().subscribe(afterComplete::add, e -> {}, () -> completed[0] = true);
实测：
late=[2, 3, 4] completed=true
emit after complete=FAIL_TERMINATED
完成后订阅仍能拿到历史，并立即收到 onComplete；再发射返回 FAIL_TERMINATED。
重放仍受下游 request 约束
一个常见误解是「重放是预加载，不需要 request」。实际上历史元素和实时元素一样，按下游的请求额度逐个发送：
Sinks.Many<Integer> sink = Sinks.many().replay().limit(3);
sink.tryEmitNext(1);
sink.tryEmitNext(2);
sink.tryEmitNext(3);
List<Integer> one = new ArrayList<>();
sink.asFlux().subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(Subscription s) {
request(1); // 只要 1 个
}
@Override
protected void hookOnNext(Integer v) {
one.add(v); // 不再 request
}
});
Thread.sleep(100);
System.out.println("one=" + one);
实测输出：
one=[1]
即使历史里有 3 个元素，下游只请求 1 个，就只收到 1 个；剩下 2 个留在重放缓冲里，等下一次 request。所以 replay 不会绕过 Reactive Streams 的背压规则，它只是把「历史」变成了可被请求的数据源。
这也解释了为什么 replay().limit(100) 不能替代 onBackpressureBuffer：前者控制的是保留多少历史，后者控制的是当前在途元素能积压多少。两者解决的不是同一个问题。
limit(Duration)：按时间淘汰
传 Duration 时，过期元素不再补给后订阅者。实测 limit(Duration.ofMillis(100))：
Sinks.Many<Integer> sink = Sinks.many().replay().limit(Duration.ofMillis(100));
sink.tryEmitNext(1);
sink.tryEmitNext(2);
Thread.sleep(150); // 超过 TTL
List<Integer> late = new ArrayList<>();
sink.asFlux().subscribe(late::add);
System.out.println("late=" + late);
输出：
late=[]
即使期间没有新元素触发清理，订阅时也不会重放已过期的 1、2。再看「过期后又有新元素」：
Sinks.Many<Integer> sink = Sinks.many().replay().limit(Duration.ofMillis(100));
sink.tryEmitNext(1);
sink.tryEmitNext(2);
Thread.sleep(150);
sink.tryEmitNext(3); // 新元素，1、2 已过期
sink.asFlux().subscribe(v -> System.out.println(v));
输出：
3
limit(int, Duration) 同时施加两个上界，先到者生效。实测 limit(2, Duration.ofMillis(200)) 连续发射 1、2、3 后立即订阅：
late=[2, 3]
这里数量上限先起作用。
TTL 的单位是「元素进入历史的时刻」，不是「元素被消费的时刻」。下游订阅得越晚，能补到的历史越少，这一点和 limit(n) 不同。
latest() 与 latestOrDefault()
latest() 只保留最后一个元素：
Sinks.Many<Integer> sink = Sinks.many().replay().latest();
sink.tryEmitNext(1);
sink.tryEmitNext(2);
List<Integer> late = new ArrayList<>();
sink.asFlux().subscribe(late::add);
sink.tryEmitNext(3);
System.out.println("latest=" + late);
实测输出：
latest=[2, 3]
后订阅者先拿到订阅前的最新值 2，再收到订阅后的 3；1 不会出现。它适合「状态快照」类数据：新订阅者只关心当前值，不需要完整变更历史。
如果订阅时还没有任何元素，latest() 不补任何东西。需要默认值时用 latestOrDefault(T)：
Sinks.Many<Integer> sink = Sinks.many().replay().latestOrDefault(0);
List<Integer> late = new ArrayList<>();
sink.asFlux().subscribe(late::add);
System.out.println("no emission late=" + late);
输出：
no emission late=[0]
0 是后订阅者在无历史时收到的初始值，不会进入后续的历史淘汰。
与 Flux.cache() 的区别
| 维度 | Sinks...replay() | Flux.cache() |
|---|---|---|
| 发射端 | 外部显式 tryEmitNext | 由上游 Publisher 驱动 |
| 启动时机 | 创建 sink 后即可发射 | 第一个订阅者出现时订阅源 |
| 无订阅者时 | 按策略缓存历史 | 源不执行 |
| 后订阅者 | 按 limit / latest 补历史 | 默认补全部；cache(n) 补最近 n 个 |
| 终止信号 | 缓存并重放完成/错误 | 缓存并重放完成/错误 |
| 多订阅者 | 原生多播 | 原生支持，但共享同一份缓存 |
Flux.cache(n) 的 n 只限制后订阅者看到的历史长度，第一个订阅者仍会收到完整序列；Sinks...replay().limit(n) 则从一开始就只保留 n 个。这个差别在排查「为什么第二个订阅者少了数据」时很关键。
选择指南
| 场景 | 选择 | 注意 |
|---|---|---|
| 新订阅者只需要当前状态 | latest() / latestOrDefault(v) | 无历史的订阅者会立即拿到默认值 |
| 聊天室最近 100 条 | limit(100) | 内存按元素数封顶 |
| 配置在 5 分钟内有效 | limit(Duration.ofMinutes(5)) | 过期后订阅者可能什么都拿不到 |
| 既要数量又要时间上界 | limit(n, ttl) | 先到者生效 |
| 完整审计历史 | all() | 无界，必须评估内存与保留时长 |
replay 的历史是进程内内存，进程退出即消失。需要跨进程、跨重启的重放，应该用消息队列或持久化日志，而不是把 replay() 当成可靠存储。
小结
| 策略 | 保留规则 | 后订阅者看到 |
|---|---|---|
all() | 全部历史 | 完整历史 + 新元素 |
limit(n) | 最近 n 个 | 最近 n 个 + 新元素 |
limit(Duration) | TTL 内 | 未过期历史 + 新元素 |
limit(n, Duration) | 数量与时间双重上界 | 两者都满足的历史 + 新元素 |
latest() | 最新 1 个 | 最新值 + 新元素 |
latestOrDefault(v) | 最新 1 个，初始为 v | v 或最新值 + 新元素 |
两点必须记住：重放仍受下游 request 约束；limit(0) 会抛 IllegalArgumentException: historySize must be > 0，不存在「退化为 publish」的写法。
阅读导航
下一篇：10.1 RxJava Subject：与 Sinks 的对照
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。