三种多播策略
Sinks.many().multicast() 提供三种背压策略，区别只在「某个订阅者跟不上时怎么办」：
| 策略 | 缓冲 | 慢订阅者 | 对其他订阅者的影响 |
|---|---|---|---|
directBestEffort() | 无 | 丢弃它自己的元素 | 无 |
directAllOrNothing() | 无 | 整个发射失败 | 所有订阅者都收不到 |
onBackpressureBuffer(size) | 共享有界队列 | 队头阻塞 | 快订阅者也被拖住 |
三者都不会为每个订阅者单独维护队列。这是理解多播行为的前提：下游不是三个独立消费者，而是共享同一条分发路径。
directBestEffort：各收各的
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Sinks;
import java.util.ArrayList;
import java.util.List;
public class BestEffortDemo {
public static void main(String[] args) {
Sinks.Many<Integer> sink = Sinks.many().multicast().directBestEffort();
List<Integer> fast = new ArrayList<>();
List<Integer> slow = new ArrayList<>();
sink.asFlux().subscribe(fast::add); // 默认无界请求
sink.asFlux().subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(Subscription s) {
// 故意不请求
}
@Override
protected void hookOnNext(Integer v) {
slow.add(v);
}
});
System.out.println("emit1=" + sink.tryEmitNext(1));
System.out.println("emit2=" + sink.tryEmitNext(2));
System.out.println("fast=" + fast + " slow=" + slow);
}
}
实测输出：
emit1=OK
emit2=OK
fast=[1, 2] slow=[]
fast 有需求，两个元素都收到了；slow 从未 request，一个也没收到。关键在于 tryEmitNext 仍然返回 OK——只要至少有一个订阅者能接收，发射就算成功，慢订阅者的损失不会体现在返回值里。要感知丢弃，只能在下游用 doOnDiscard 或者自己统计。
directAllOrNothing：一人不齐，全部不发
把上面的工厂换成 directAllOrNothing()，输出立刻不同：
emit1=FAIL_OVERFLOW
emit2=FAIL_OVERFLOW
fast=[]
slow 没有需求，于是每次发射都整体失败：fast 明明有需求，也拿不到元素。这里的失败不是「缓冲满了」，而是「无法同时满足所有订阅者」，返回值统一是 FAIL_OVERFLOW。
它适合「所有订阅者必须看到完全相同序列」的场景，例如多个模块共享同一份状态变更。代价是最慢的订阅者决定整条流的进度，而且没有缓冲来吸收差异。
onBackpressureBuffer：共享队列与队头阻塞
onBackpressureBuffer(size) 用一个共享的有界队列吸收差异。直觉上会以为「快订阅者先拿、慢订阅者慢慢追」，实际不是。
Sinks.Many<Integer> sink = Sinks.many().multicast().onBackpressureBuffer(5);
List<Integer> fast = new ArrayList<>();
List<Integer> slow = new ArrayList<>();
AtomicReference<Subscription> slowSub = new AtomicReference<>();
sink.asFlux().subscribe(fast::add);
sink.asFlux().subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(Subscription s) {
slowSub.set(s); // 先不请求
}
@Override
protected void hookOnNext(Integer v) {
slow.add(v);
}
});
for (int i = 0; i < 10; i++) {
System.out.println("emit " + i + "=" + sink.tryEmitNext(i));
}
System.out.println("before request: fast=" + fast + " slow=" + slow);
slowSub.get().request(100);
System.out.println("after request: fast=" + fast + " slow=" + slow);
实测输出：
emit 0=OK
emit 1=OK
emit 2=OK
emit 3=OK
emit 4=OK
emit 5=OK
emit 6=OK
emit 7=OK
emit 8=FAIL_OVERFLOW
emit 9=FAIL_OVERFLOW
before request: fast=[]
after request: fast=[0, 1, 2, 3, 4, 5, 6, 7] slow=[0, 1, 2, 3, 4, 5, 6, 7]
这里有三个值得注意的观察：
fast在慢订阅者补齐需求前一个元素都没收到。队列里的元素在等待所有订阅者都有需求，快订阅者不能越过队头取走。这就是队头阻塞（head-of-line blocking）。- 8 个元素被接受了，尽管构造参数是 5。容量不是 5，见下一节。
- 第 9、10 个元素返回
FAIL_OVERFLOW，但已经入队的 8 个没有丢；慢订阅者一旦请求，所有订阅者按序收到完整的 0–7。
所以 onBackpressureBuffer 的「缓冲」不是「快慢解耦」，只是把「立刻失败」推迟到队列满。只要有一个订阅者长期不消费，整条多播都会停住；队列满了之后，连快订阅者也拿不到新元素。
容量取整：你传 5，实际是 8
onBackpressureBuffer(size) 的 size 不是精确容量。实测：
| 构造参数 | 实际可接受元素数 |
|---|---|
| 1 | 1 |
| 2 | 8 |
| 3 | 8 |
| 5 | 8 |
| 10 | 16 |
机制在 Queues.get(size)：size == 1 时特判为一个元素的队列；其余情况下先取 max(8, size)，再按 SpscArrayQueue 的要求向上取到 2 的幂。所以 2、3、5 都落到 8，10 落到 16。
不传参数时默认 Queues.SMALL_BUFFER_SIZE，即 256；可用系统属性 reactor.bufferSize.small 覆盖（下限 16）。实测无订阅者时连续发射，第 257 个开始返回失败：
default accepted (no subscriber)=256
size 必须严格大于 0：
size 0: java.lang.IllegalArgumentException: bufferSize must be strictly positive, was: 0
size -1: java.lang.IllegalArgumentException: bufferSize must be strictly positive, was: -1
要表达「允许丢弃」，用 directBestEffort()，不要试图用 onBackpressureBuffer(0) 代替。
emitFailureHandler 与重试
tryEmitNext 只尝试一次，结果交给你。emitNext(v, handler) 会在 handler 返回 true 时重试。Reactor 内置两种：
// 不重试
sink.emitNext(v, Sinks.EmitFailureHandler.FAIL_FAST);
// 在预算内自旋重试
sink.emitNext(v, Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(200)));
busyLooping(Duration) 的截止时间在创建 handler 时计算，不是每次发射重新计时。并且它只对 FAIL_NON_SERIALIZED 返回 true：
// OptimisticEmitFailureHandler.onEmitFailure 的等价逻辑
return result == EmitResult.FAIL_NON_SERIALIZED
&& deadline - System.nanoTime() > 0;
这意味着 busyLooping 解决的是并发发射冲突，不解决缓冲溢出。实测 onBackpressureBuffer(1)、订阅者不请求：
emit1 ok
error: OverflowException: Backpressure overflow during Sinks.Many#emitNext
emit2 ok
elapsed ms=5
later try=FAIL_TERMINATED
第二次 emitNext 遇到 FAIL_OVERFLOW，handler 立即返回 false（没有等 200ms 预算），sink 被 OverflowException 终止，后续 tryEmitNext 返回 FAIL_TERMINATED。注意 emitNext 本身不抛异常，它把错误送给了订阅者；emit2 ok 打印出来不代表第二个元素被接受。
策略选择
| 需求 | 选择 |
|---|---|
| 实时事件，允许慢订阅者丢数据 | directBestEffort() |
| 所有订阅者必须看到相同序列，宁可失败 | directAllOrNothing() |
| 允许短暂积压，但必须给积压一个上限 | onBackpressureBuffer(size) |
| 需要在订阅前缓存（warm-up） | onBackpressureBuffer(size) |
三个选项都假设订阅者最终会消费。如果某个订阅者可能永久停住，任何多播策略都救不了它：directAllOrNothing 会让整条流失效，onBackpressureBuffer 会填满后失效，directBestEffort 会一直丢它的数据。真正的解法是给慢消费者单独一条流，或者在下游用 timeout / onBackpressureDrop 明确它的失败方式。
小结
| 策略 | 发射成功条件 | 慢订阅者的代价 |
|---|---|---|
directBestEffort | 至少一个订阅者有需求 | 只丢自己的元素 |
directAllOrNothing | 所有订阅者都有需求 | 所有人一起失败 |
onBackpressureBuffer | 共享队列未满 | 所有人一起队头阻塞 |
onBackpressureBuffer 的容量按 max(8, size) 再向上取 2 的幂；默认 256。busyLooping 只重试 FAIL_NON_SERIALIZED，溢出仍会终止 sink。
阅读导航
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。