Disruptor - 消费者是多线程的吗？

Posted 2023-04-12

【中文标题】Disruptor - 消费者是多线程的吗？

【英文标题】：Disruptor - Are the Consumers Multithreaded?

【发布时间】：2013-06-05 19:59:03

【问题描述】：

我有以下关于破坏者的问题：

消费者（事件处理器）没有实现他们实现EventHandler的任何Callable或Runnable接口，那么他们如何并行运行， 所以例如我有一个中断器实现，其中有一个像这样的菱形图案

c1 P1 - c2 - c4 - c5 c3

其中c1到c3可以在p1之后并行工作，C4和C5在它们之后工作。

所以通常我会有这样的东西（P1 和 C1-C5 是可运行/可调用的）

p1.start();
p1.join();

c1.start();
c2.start();
c3.start();
c1.join();
c2.join();
c3.join();

c4.start();
c4.join();
c5.start();
c5.join();

但是在 Disruptor 的情况下，我的事件处理程序都没有实现 Runnable 或 Callable，那么 Disruptor 框架最终是如何并行运行它们的呢？

采取以下场景：

我的消费者 C2 需要对事件进行一些注释的 web 服务调用，在 SEDA 中，我可以为这样的 10 个 C2 请求启动 10 个线程 [用于将消息拉出队列 + 进行 Web 服务调用并更新下一个 SEDA 队列]这将确保我不会依次等待 10 个请求中的每一个的 Web 服务响应 在这种情况下，我的事件处理器 C2（如果）是单个实例，将依次等待 10 个 C2 请求。

【参考方案1】：

您的 EventHandler 被组合成一个 BatchEventProcessor 的实例，它是一个 Runnable。

BatchEventProcessor 实现了 run() 循环。 它负责从环形缓冲区中获取更新的事件序列号。 当事件可用于处理时，它会将它们传递给您的 EventHandler 进行处理。

当使用 DSL 时，Disruptor 负责通过 Executor 实例创建这些线程。

您在 Disruptor 构造函数中提供一个 Executor。 您在 DSL 上使用 and/then/etc 提供您的 EventHandlers 列表。 然后，调用 Disruptor start() 方法。作为 start() 方法的一部分，每个 EventProcessor (Runnable) 都会提交给 Executor。 Executor 将依次在线程上启动您的 EventProcessor/EventHandler。

关于您的具体场景（即：长时间运行的事件处理程序），您可以参考这个问题：

Solution to slow consumer(eventProcessor) issue in LMAX Disruptor pattern

【讨论】：

非常感谢您的及时答复。在创建依赖图（handleWith 和/then 等）时，我已经提供了我的消费者的实例。因此，虽然我可以让三个批处理事件处理器（每个 c1、c2 和 c3 一个）并行运行，但它们每个都只有一个各自事件处理程序的实例（我在一开始就提供了）。换句话说，假设 C1 对 WebSrv1 进行了 Web 服务调用，永远不能在环形缓冲区中的两个事件桶中并行调用两个 WebServ1（插槽 2 和 3 同时执行 C1 的场景）。那不是真的吗？

我解释为 c2 和 c3 处理管道中的不同阶段（即：不调用 WebSrv1）。如果是这样，是的。你可能想看看使用 Netty / async http；将 http 响应反馈回环形缓冲区，并将环形缓冲区变成状态机。或者，请参阅上面链接的条目。

嗨 Jason，我们观察到，如果我们在一个应用程序中一起使用 5-10 个破坏者（有点像一个破坏者链，每个破坏者都有一个消费者执行指定的任务，然后交给通过消息传递给下一个中断器/环形缓冲区），发生的情况是 CPU 利用率达到 90% 及以上，系统变得无响应，直到我们关闭应用程序，我们觉得这是因为有这么多活跃的中断器线程。即使破坏者没有真正处理任何事情，也会发生这种情况。你能对此发表评论吗？

检查您的 WaitStrategy。 sleep 和 busy spin 会在空闲时消耗 CPU。如果您的线程数多于内核数，那么 Disruptor 可能不适合您的用例。考虑将事件处理程序从多个中断器合并到一个中断器中？

【参考方案2】：

Disruptor 默认策略是多线程的，所以如果您的每个处理器都在不同的 Handler（消费者）中工作，那么应该没问题，并且您的处理器是多线程的。