如何异步向所有侦听器/订阅者发送事件？ [关闭]答案

【问题标题】：How to send event to all listeners/subscribers asynchronously? [closed]如何异步向所有侦听器/订阅者发送事件？ [关闭]
【发布时间】：2018-07-26 04:21:41
【问题描述】：

我们有广播服务，它维护着一个听众名单。广播者调用的方法是阻塞的。如何将其转换为异步或以非阻塞方式向所有侦听器发送事件。为每个调用创建线程是唯一的方法吗？

public class EventBroadCaster {
List<Listener> allListeners = new ArrayList();

private void sendEventToAll() {
    for(Listener listener:allListeners)
        listener.accept(String event);
}
}

public interface Listener {
    public void accept(String event);
}

结构是这样的，问题是accept(String event) 方法被阻塞，EventBroadCaster 必须等到一个侦听器的accept 完成，然后再调用另一个侦听器的accept。有没有办法让accept方法立即返回？

allListeners 中可以有 100 个 Listeners，不要为每个接受创建 100 个线程。

【问题讨论】：

你的问题不是很清楚，请详细说明
解释准确事件链是什么——什么叫什么，什么时候叫。什么方法在阻塞，它做了什么，它阻塞了什么？
你能提供一个minimal reproducible example吗？你应该在你的问题中edit它。
@AustineGwa 现在看起来还可以吗？
这听起来像是一个设计缺陷。您的接受方法阻塞！？您可以使用执行程序服务并使用多个线程进行调用。那么只要太多了，别挡住你的罚款。您也可以尝试使用并行流。

标签： java multithreading asynchronous

【解决方案1】：

你可以使用，

String event = "wow events are strings";
allListeners.stream().parallel().forEach(l->l.accept(event));

或者您可以显式使用 ExecutorService。

ExecutorService service = Executors.newFixedThreadPool(8);
...
...

List<Callable> tasks = allListeners.stream().map(l->{ l.accept(event); return l;}).collect(Collectors.toList());

service.invokeAll(tasks);

这将一直阻塞，直到每个人都收到通知，但它会并行通知。有很多变体可以避免阻塞。

一个这样的例子，虽然不推荐，但就是开枪就忘了。

allListeners.forEach(l->service.execute(()->l.accept(event)));

这将停止广播方法的阻塞，但如果太多的接受导致阻塞，你的所有线程都可能会被阻塞。

我觉得实际的解决方案是修复接受方法。类似于 EDT（事件调度线程）的工作方式。您可以让 ActionListener 花费很长时间，但这会导致 gui 暂停。相反，您让动作侦听器生成一个任务，因此 EDT 不必等待。

另外，你可能会研究 RxJava。

【讨论】：

谢谢马特，我怎样才能避免阻塞？
我添加了一个不会被屏蔽的版本。
@Manish 真正的解决方案在最后一段 - 事件侦听器/处理程序代码通常应该快速返回。由每个处理逻辑来决定执行是否足够快，以至于它不会“足够阻塞”或产生一个新的任务/线程/服务......将在后台执行。
@matt 我喜欢火而忘记的方法。这就是我一直在寻找的。你为什么不推荐它？多线程有点新。
@matt 我知道最后一种方法是理想的，但我在想听听者摆布快速完成接受块是否是个好主意。一个错误的 accept 方法实现，所有的通知都会被延迟。