Sinks

Doubao 中英对照 Sinks

目录

• 1. 通过 Sinks.One 和 Sinks.Many 安全地从多线程产生数据
• 2. 可用 Sinks 概览
  • 2.1. Sinks.many().unicast().onBackpressureBuffer(args?)
  • 2.2. Sinks.many().multicast().onBackpressureBuffer(args?)
  • 2.3. Sinks.many().multicast().directAllOrNothing()
  • 2.4. Sinks.many().multicast().directBestEffort()
  • 2.5. Sinks.many().replay()
  • 2.6. Sinks.unsafe().many()
  • 2.7. Sinks.one()
  • 2.8. Sinks.empty()

在 Reactor 中，sink 是一个类，它允许以独立的方式安全地手动触发信号，创建一个类似 Publisher 的结构，能够处理多个 Subscriber（unicast() 变体除外）。

在 3.5.0 之前，还有一组 Processor 实现，但已被逐步淘汰。

1. 通过 Sinks.One 和 Sinks.Many 安全地从多线程产生数据

reactor-core 暴露的 Sinks 默认变体确保多线程使用会被检测到，并且从下游订阅者的角度看不会导致违反规范或未定义行为。使用 tryEmit* API 时，并行调用会快速失败。使用 emit* API 时，所提供的 EmissionFailureHandler 可能允许在竞争时重试（例如忙循环），否则 sink 将以错误终止。

这是对 Processor.onNext 的改进，后者必须在外部进行同步，否则从下游订阅者的角度看会导致未定义行为。

备注

Processor 是一种特殊的 Publisher，它同时也是一个 Subscriber。它们最初被设想为一种中间步骤的可能表示形式，随后可在不同的 Reactive Streams 实现之间共享。然而在 Reactor 中，这类步骤更多地由作为 Publisher 的操作符来表示。

第一次接触 Processor 时常见的错误，就是忍不住直接调用 Subscriber 接口暴露的 onNext、onComplete 和 onError 方法。

这种手动调用应当谨慎进行，特别是要按照 Reactive Streams 规范对调用进行外部同步。实际上，Processor 的用处大概很有限，除非您遇到一个基于 Reactive Streams 的 API，它要求传入一个 Subscriber，而不是暴露一个 Publisher。

Sinks 通常是更好的替代方案。

Sinks 构建器为主要受支持的生产者类型提供了一套引导式 API。您会认出其中一些在 Flux 中见过的行为，例如 onBackpressureBuffer。

Sinks.Many<Integer> replaySink = Sinks.many().replay().all();

多个生产线程可以通过以下方式并发地在 sink 上生成数据：

//thread1
replaySink.emitNext(1, EmitFailureHandler.FAIL_FAST);

//thread2, later
replaySink.emitNext(2, EmitFailureHandler.FAIL_FAST);

//thread3, concurrently with thread 2
//would retry emitting for 2 seconds and fail with EmissionException if unsuccessful
replaySink.emitNext(3, EmitFailureHandler.busyLooping(Duration.ofSeconds(2)));

//thread3, concurrently with thread 2
//would return FAIL_NON_SERIALIZED
EmitResult result = replaySink.tryEmitNext(4);

备注

使用 busyLooping 时请注意，返回的 EmitFailureHandler 实例不能复用，例如每次 emitNext 都应单独调用一次 busyLooping。此外，建议使用 100ms 以上的超时，因为更小的值没有实际意义。

Sinks.Many 可以作为一个 Flux 呈现给下游消费者，如下例所示：

Flux<Integer> fluxView = replaySink.asFlux();
fluxView
	.takeWhile(i -> i < 10)
	.log()
	.blockLast();

类似地，Sinks.Empty 和 Sinks.One 变体可以通过 asMono() 方法被视为一个 Mono。

Sinks 的类别有：

1. many().multicast()：一个只把新推入的数据传输给其订阅者的 sink，并尊重它们的背压（“新推入”指“订阅者订阅之后”）。

2. many().unicast()：与上面相同，区别在于第一个订阅者注册之前推入的数据会被缓冲。

3. many().replay()：一个会向新订阅者重放指定历史大小的已推入数据、然后继续实时推送新数据的 sink。

4. one()：一个只向其订阅者播放单个元素的 sink

5. empty()：一个只向其订阅者播放终止信号（错误或完成）的 sink，但仍可被视为一个 Mono<T>（注意泛型类型 <T>）。

2. 可用 Sinks 概览

2.1. Sinks.many().unicast().onBackpressureBuffer(args?)

一个单播（unicast）Sinks.Many 可以通过使用一个内部缓冲来处理背压。代价是它至多只能有一个 Subscriber。

基本的单播 sink 通过 Sinks.many().unicast().onBackpressureBuffer() 创建。但 Sinks.many().unicast() 中还有一些额外的 unicast 静态工厂方法，允许更精细的调整。

例如，默认情况下它是无界的：如果在其 Subscriber 尚未请求数据时推入任意数量的数据，它会缓冲所有数据。您可以通过在 Sinks.many().unicast().onBackpressureBuffer(Queue) 工厂方法中为内部缓冲提供一个自定义的 Queue 实现来改变这一点。如果该队列是有界的，那么当缓冲已满且尚未收到足够的下游请求时，sink 可能会拒绝推入某个值。

2.2. Sinks.many().multicast().onBackpressureBuffer(args?)

一个多播（multicast）Sinks.Many 可以向多个订阅者发射，同时尊重每个订阅者的背压。订阅者只收到它们订阅之后通过 sink 推入的信号。

基本的多播 sink 通过 Sinks.many().multicast().onBackpressureBuffer() 创建。

默认情况下，如果它的所有订阅者都被取消了（这基本上意味着它们都已取消订阅），它会清空内部缓冲并停止接受新的订阅者。您可以通过 Sinks.many().multicast() 下 multicast 静态工厂方法中的 autoCancel 参数来调整这一行为。

2.3. Sinks.many().multicast().directAllOrNothing()

一个对背压做简单处理的多播 Sinks.Many：如果任一订阅者太慢（需求量为零），则对所有订阅者丢弃该 onNext。

然而，慢速订阅者不会被终止，一旦它们重新开始请求，所有订阅者都会恢复接收此后推入的元素。

一旦 Sinks.Many 已终止（通常是通过调用其 emitError(Throwable) 或 emitComplete() 方法），它仍允许更多订阅者订阅，但会立即向它们重放终止信号。

2.4. Sinks.many().multicast().directBestEffort()

一个对背压做尽力而为处理的多播 Sinks.Many：如果某个订阅者太慢（需求量为零），则只对这个慢速订阅者丢弃该 onNext。

然而，慢速订阅者不会被终止，一旦它们重新开始请求，就会恢复接收新推入的元素。

一旦 Sinks.Many 已终止（通常是通过调用其 emitError(Throwable) 或 emitComplete() 方法），它仍允许更多订阅者订阅，但会立即向它们重放终止信号。

2.5. Sinks.many().replay()

一个重放（replay）Sinks.Many 会缓存已发射的元素，并向迟到的订阅者重放它们。

它可以有多种配置方式：

• 缓存有限历史（Sinks.many().replay().limit(int)）或无界历史（Sinks.many().replay().all()）。

• 缓存基于时间的重放窗口（Sinks.many().replay().limit(Duration)）。

• 缓存历史大小与时间窗口的组合（Sinks.many().replay().limit(int, Duration)）。

上述配置的额外重载也可以在 Sinks.many().replay() 下找到，此外还有一个允许缓存单个元素的变体（latest() 和 latestOrDefault(T)）。

2.6. Sinks.unsafe().many()

高级用户和操作符构建者可能希望考虑使用 Sinks.unsafe().many()，它会提供相同的 Sinks.Many 工厂，但不附带额外的生产者线程安全保障。因此每个 sink 的开销会更小，因为线程安全的 sink 必须检测多线程访问。

库的开发者不应暴露不安全的 sink，但可以在受控的调用环境内部使用它们——在这种环境中，他们可以确保对导致 onNext、onComplete 和 onError 信号的调用进行外部同步，以遵守 Reactive Streams 规范。

2.7. Sinks.one()

这个方法直接构造一个 Sinks.One<T> 的简单实例。这种 Sinks 变体可被视为一个 Mono（通过其 asMono() 视图方法），并具有略有不同的 emit 方法，以更好地传达这种类似 Mono 的语义：

• emitValue(T value) 生成一个 onNext(value) 信号，并且——在大多数实现中——还会触发一个隐式的 onComplete()

• emitEmpty() 生成一个孤立的 onComplete() 信号，旨在生成等价于空 Mono 的结果

• emitError(Throwable t) 生成一个 onError(t) 信号

Sinks.one() 只接受这些方法中任意一个的一次调用，实际上生成一个要么带值完成、要么空完成、要么失败的 Mono。

2.8. Sinks.empty()

这个方法直接构造一个 Sinks.Empty<T> 的简单实例。这种 Sinks 变体类似于 Sinks.One<T>，区别在于它不提供 emitValue 方法。

因此，它只能生成一个空完成或失败的 Mono。

尽管这个 sink 无法触发 onNext，它仍然带有一个泛型 <T>，因为这样便于组合，并能纳入要求特定类型的操作符链中。