JAVA Reactor中Sinks.Many类三种常见的创建方式及使用

一、Java中的Sinks.Many

在Java编程中，我们经常需要处理数据流。为了更有效地处理数据流，我们可以使用Reactor库中的Sinks.Many类。这个类提供了一种简单而强大的方式来处理多个事件流，并且可以通过异步或者同步的方式处理这些事件。

Sinks.Many是Reactor库中的一个类，它可以用来处理多个事件流。它提供了一种简单而强大的方式来处理多个事件，可以在任何时间添加、获取以及完成事件。

Sinks.Many为什么被设计为 Sinks 类的内部接口？

实现细节隐藏​​： Sinks.Many 的实际实现（如 MulticastReplayProcessor 或其他内部类）被封装在 Sinks 类的内部。用户只需通过工厂方法（如 Sinks.many().multicast()）获取接口，无需关心底层实现。 ​​

接口职责分离​​： Sinks 类作为工厂，统一管理所有类型的 Sink（如 Sinks.One、Sinks.Many），而 Sinks.Many 作为内部接口，明确表示它是一个​​多播数据源​​，与单播（Sinks.One）等场景隔离。

二、Sinks.Many的创建

在源码中可以看待，Sinks.Many提供了三中常见的创建方式。

（1）unicast()

Sinks.Many<String> unicastSink = Sinks.many().unicast().onBackpressureBuffer();

这种创建方式中提供设置背压缓冲区的方法

• 用途：

• 创建一个 单播（Unicast） 的 Sinks.Many，仅允许 一个订阅者 订阅。

• 核心特性：

  • 单订阅者限制：第二个订阅者尝试订阅时会触发 IllegalStateException。

  • 背压支持：通过缓冲区处理生产者和消费者的速率不匹配，默认使用无界缓冲区（需手动配置限制）。

  • 无历史数据：订阅者只能收到订阅后产生的数据。

• 适用场景：

  • 点对点通信（如任务队列）。

  • 需要严格保证单订阅者的场景（如资源独占型操作）。

（2）multicast()

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

这个创建方式，相比较Unicast，就是可以允许多个订阅者订阅，同样提供了多种设置缓存区的方式

• 用途：
  创建一个 多播（Multicast） 的 Sinks.Many，允许多个订阅者订阅。

• 核心特性：

  • 多订阅者支持：所有订阅者共享同一数据流。

  • 无历史数据：新订阅者只能收到订阅后产生的数据，无法获取之前的数据。

  • 背压策略：默认使用无界缓冲区，但可以通过配置限制（如 onBackpressureBuffer(int capacity)）。

• 适用场景：

  • 实时广播（如股票行情推送）。

  • 需要多个消费者并行处理相同数据的场景。

（3）replay()

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

这个方法中提供了限制存放历史数据的方法

• 用途：
  创建一个 支持数据重放（Replay） 的 Sinks.Many，允许多个订阅者订阅，并回放历史数据。

• 核心特性：

  • 历史数据缓存：新订阅者可以收到订阅前一定数量的数据（通过 limit(int) 或 time(Duration) 配置）。

  • 多订阅者支持：与 multicast() 类似，但增加了数据重放能力。

  • 内存管理：缓存数据量或时间窗口可配置，避免内存无限增长。

• 适用场景：

  • 需要新订阅者获取历史数据的场景（如聊天记录回放）。

  • 实时监控面板（多个订阅者需要看到完整上下文）。

三、Sinks.Many的使用

（1）添加事件

一旦我们创建了一个Sinks.Many对象，我们可以使用emitNext()方法来添加一个事件。这个方法接受一个参数，表示要添加的事件。

Sinks.Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();
sink.emitNext("Event 1");
sink.emitNext("Event 2");
sink.emitNext("Event 3");

我们首先创建了一个Sinks.Many对象，然后使用emitNext()方法添加了三个事件。

当然，如果我们选择接受Flux流中的数据的时候，可以这样添加数据

 //创建Flux流
        Flux<String> flux = Flux.<String>create(sink->{
            sink.next("你好");
            sink.complete();
        });

        //创建Sinks.Many处理流数据
        Sinks.Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();

        //订阅Flux流，并将数据交给sink处理（像做流数据的缓存，筛选流数据）
        flux.subscribe(
                sink::tryEmitNext,
                sink::tryEmitError,
                sink::tryEmitComplete
        );

（2）获取流数据

我们可以使用Sinks.Many对象来获取已添加的事件。可以使用asFlux()方法将Sinks.Many对象转换为一个Flux对象，然后使用Flux对象的方法来订阅和处理事件。

 // 1. .unicast()使用创建单播 Sink
        Sinks.Many<String> unicastSink = Sinks.many().unicast().onBackpressureBuffer();

        // 2. 将 Sink 转换为 Flux（供订阅者订阅）
        Flux<String> flux = unicastSink.asFlux();

        // 3. 第一个订阅者（合法）
        flux.subscribe(
                data -> System.out.println("订阅者1收到数据: " + data),
                error -> System.err.println("订阅者1发生错误: " + error),
                () -> System.out.println("订阅者1完成")
        );

        // 4. 推送数据到 Sink
        unicastSink.tryEmitNext("数据1");
        unicastSink.tryEmitNext("数据2");

        // 5. 尝试第二个订阅者（会抛出 IllegalStateException）
        try {
            flux.subscribe(
                    data -> System.out.println("订阅者2收到数据: " + data),
                    error -> System.err.println("订阅者2发生错误: " + error),
                    () -> System.out.println("订阅者2完成")
            );
        } catch (IllegalStateException e) {
            System.err.println("订阅者2订阅失败: " + e.getMessage());
        }

        // 6. 关闭 Sink（发送完成信号）
        unicastSink.tryEmitComplete();

这里我使用了unicast()方法创建的Sinks.Many，这个时候我通过asFlux()方法转换的flux流，只能被一个订阅者订阅到，第二个订阅者，订阅的时候就出报错，当然，如果你想要多个订阅者订阅，可以使用multicast()或者replay()方式创建，

四、热冷流

上面三种方式创建的是热流。

热流：数据独立于订阅者持续生成，多订阅者共享实时数据，适用于实时事件推送。

在添加事件代码中我也有使用到Flux.create()方法创建流，要注意这里创建的是冷流。

冷流：每次调用 subscribe() 时，会触发 Flux.create() 的回调函数（即 Consumer<SynchronousSink<T>>），​​重新生成数据流​​。多个订阅者之间数据独立。这个跟ThreadLocal有点像，Flux.create()会为每一个订阅者创建单独隔离的数据流，保证每一条流中数据互不影响。

冷热流的选择;

• 若需要多个订阅者共享实时数据 → 热流。
• 若需要每个订阅者独立消费完整数据 → 冷流。
• 若需要历史数据 → 使用 replay() 缓存。

总结 

到此这篇关于JAVA Reactor中Sinks.Many类三种常见的创建方式及使用的文章就介绍到这了,更多相关JAVA Reactor中Sinks.Many类内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

最新评论