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

一、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()会为每一个订阅者创建单独隔离的数据流，保证每一条流中数据互不影响。

冷热流的选择;