从多个线程同时使用同一个 FluxSink 是否安全答案

【问题标题】：Is it safe to use the same FluxSink from multiple threads concurrently从多个线程同时使用同一个 FluxSink 是否安全
【发布时间】：2018-09-28 15:56:01
【问题描述】：

我知道Publisher 不能同时发布，但如果我使用Flux#create(FluxSink)，我可以安全地同时调用FluxSink#next 吗？

换句话说，Spring 是否具有确保正确串行发布事件的内部魔法，即使同时调用FluxSink#next？

public class FluxTest {

    private final Map<String, FluxSink<Item>> sinks = new ConcurrentHashMap<>();

    // Store a new sink for the given ID
    public void start(String id) {
        Flux.create(sink -> sinks.put(id, sink));
    }

    // Called from different threads
    public void publish(String id, Item item) {
        sinks.get(id).next(item); //<----------- Is this safe??
    }
}

它听起来像官方指南中的this paragraph 表明上述确实是安全的，但我对自己的理解不是很有信心。

create 是一种更高级的 Flux 编程创建形式，适用于每轮多次发射，甚至来自多个线程。

【问题讨论】：

标签： java spring concurrency project-reactor reactive-streams

【解决方案1】：

是的，Flux.create 生成一个 SerializedSink，可以安全地从多个线程用于 next 调用

【讨论】：

我从 stackoverflow 上的多个答案中读到，不建议从 lambda 中转义接收器（例如 stackoverflow.com/a/56063933 及其 cmets），并且您必须改用处理器。你有这方面的信息吗？谢谢
我认为该评论更多地是关于这些FluxSink 是根据Subscription 创建的这一事实，因此在多个订阅的情况下在一个字段中捕获一个可能会产生误导到Flux.create(...)。请注意，在 3.4 中，我们引入了一种新风格的 Sinks，它更适合这种“在具有多个可能订阅的字段中独立”的使用模式。线程安全注释仍然存在，例如。可以从Flux.create 生成多个线程，并将每个馈送数据发送到相同的FluxSink。
非常感谢，我想知道这个示例与文档中创建事件侦听器并在其方法中捕获接收器的示例之间的区别。在这两种情况下，水槽都会“逃逸”。我认为您是对的，真正的区别在于在多个订阅的情况下拥有多个对象。或者，您可以将接收器添加到字段中的列表中。感谢 3.4 的提醒。
@Dachstein（StandaloneFluxSink 来自过时的 Milestone 或 RC）Sinks.Many 默认情况下检测多线程使用并失败。他们有一个tryEmitNext API，它将立即返回一个枚举，其中一个表示尝试并行发射的事实。然后你可以选择乐观地重试/忙循环，或者认为这是一个错误。
仅来自 Flux 创建。它旨在为每个订阅创建一个接收器