9.1 Sinks：程序化发射数据

为什么需要 Sinks

第 2 章手写的 Publisher 只在 subscribe() 时产生数据。但很多数据来自订阅之外的异步事件：

• WebSocket 收到消息；
• 数据库变更监听回调；
• 用户点击、消息队列投递。

Flux.create 可以桥接这类回调，但它把发射能力藏在 create 的 lambda 里，外部拿不到一个稳定的句柄。Sinks 把「发射端」显式暴露出来：

Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();

// 消费端
sink.asFlux().subscribe(System.out::println);

// 生产端，可以在任意回调里调用
sink.tryEmitNext("hello");
sink.tryEmitNext("world");
sink.tryEmitComplete();

实测输出：

hello
world

Sinks.Many 的三种模式

// 单播：只允许一个订阅者；订阅前的元素会保留，订阅后回放
Sinks.Many<String> unicast = Sinks.many().unicast().onBackpressureBuffer();

// 多播：多个订阅者共享；订阅前的元素作为 warm-up 缓冲
Sinks.Many<String> multicast = Sinks.many().multicast().onBackpressureBuffer();

// 重放：为后订阅者保留历史
Sinks.Many<String> replay = Sinks.many().replay().limit(10);

三者的共同点是「发射端独立于订阅者」，区别在于多订阅者和历史保留：

零订阅者时是否接受发射，取决于实现。实测：

multicast buffer zero-subscriber: OK
best-effort zero-subscriber: FAIL_ZERO_SUBSCRIBER
replay zero-subscriber: OK
unicast zero-subscriber: OK

带缓冲的多播、单播和 replay 都允许在没人订阅时先发射（缓冲或历史）；directBestEffort 和 directAllOrNothing 会返回 FAIL_ZERO_SUBSCRIBER。

发射 API：tryEmitNext 与 emitNext

Sinks.Many 有两组发射方法：

tryEmitNext 是「尝试一次、把结果交给你」：

Sinks.EmitResult result = sink.tryEmitNext("data");
if (result.isFailure()) {
    log.warn("emit failed: {}", result);
}

emitNext 需要显式传入失败处理器。FAIL_FAST 表示不重试：

sink.emitNext("data", Sinks.EmitFailureHandler.FAIL_FAST);

这里有个容易忽略的行为：emitNext 遇到 FAIL_OVERFLOW 时，不只是丢弃当前元素，还会把整条 sink 以 OverflowException 终止。源码里的处理是 emitError(failWithOverflow("Backpressure overflow during Sinks.Many#emitNext"))。

实测（onBackpressureBuffer(1)、订阅者不请求、连续 emitNext）：

subscriber error: OverflowException: Backpressure overflow during Sinks.Many#emitNext
later try=FAIL_TERMINATED

对比 tryEmitNext 在同样场景下的结果：

try 0=OK
try 1=FAIL_OVERFLOW
try 2=FAIL_OVERFLOW
try 3=FAIL_OVERFLOW
try 4=FAIL_OVERFLOW

订阅者没有收到错误，sink 也没有终止；后续发射仍会返回 FAIL_OVERFLOW。如果要「溢出时丢当前元素、但继续服务」，用 tryEmitNext 并自己处理结果；如果要「溢出即失败」，用 emitNext。

EmitResult

FAIL_NON_SERIALIZED 是理解 Sinks 并发模型的关键，下一节展开。

线程安全：默认 Sinks 不序列化并发发射

一个流传很广的说法是「Sinks 是线程安全的，可以从任意线程发射」。这个说法不准确。默认 Sinks 会检测并发发射并返回 FAIL_NON_SERIALIZED，但它不会替你把并发发射排成序。

实测：两个线程各向同一个 multicast().onBackpressureBuffer() 发射 500 个元素，并发开始：

results={FAIL_NON_SERIALIZED=772, OK=228}
received=228

1000 次尝试只有 228 次成功，772 次因为并发冲突被拒绝；下游只收到 228 个元素。如果代码不检查返回值，就会静默丢数据。

正确做法有两种：

1. 自己序列化：所有发射都从一个线程或一把锁里发出；
2. 用 Sinks.unsafe().many()：它跳过并发检测，调用方必须保证串行。它不会让并发变安全，只是把检测开销去掉；并发调用会导致未定义行为。

// 安全：所有发射都在同一把锁内
synchronized (lock) {
    sink.tryEmitNext(value);
}

// 或者：明确由单线程发射
emitterThread.submit(() -> sink.tryEmitNext(value));

emitNext 遇到 FAIL_NON_SERIALIZED 不会重试（FAIL_FAST），而是抛出 Sinks.EmissionException：

Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.

这条异常信息直接指向 Reactive Streams 的串行信号规则。

发射端规则

除了串行，发射端还要遵守几件事：

• 终止只发一次。tryEmitComplete() 或 tryEmitError() 成功后，后续发射返回 FAIL_TERMINATED。
• 不要在 onNext 里递归发射。回调里再次 tryEmitNext 可能触发并发/重入检测；需要级联时用 concatMap 或异步边界。
• 检查返回值，至少监控失败。FAIL_OVERFLOW、FAIL_NON_SERIALIZED 都不是「可忽略」的成功。
• 完成时释放外部资源。注册 onCancel / onDispose 回调，关闭 WebSocket、取消监听器。

与 Processor 的对比

Sinks 并没有把并发问题变没，只是把它从「未定义行为」变成「可检测的失败结果」。这已经比旧 Processor 好，但仍然要求调用方处理结果。

小结

阅读导航

上一篇：8.3 refCount、autoConnect 与 cache

下一篇：9.2 Sinks 多播：背压与重放