当生产者比消费者快得多时，为什么 Disruptor 会持有大量数据？答案

【问题标题】：Why does the Disruptor hold lots of data when the producer is much faster than the consumer?当生产者比消费者快得多时，为什么 Disruptor 会持有大量数据？
【发布时间】：2019-04-28 19:47:12
【问题描述】：

我正在学习 LMAX Disruptor 并遇到一个问题：当我有一个非常大的环形缓冲区（例如 1024）并且我的生产者比我的消费者快得多时，环形缓冲区将保存大量数据，但不会发布事件，直到我的应用程序结束。这意味着我的应用程序将丢失大量数据（我的应用程序不是守护程序）。

我试图降低生产者的速度，这很有效。但是我不能在我的应用程序中使用这种方法，它会大大降低我的应用程序的性能。

val ringBufferSize = 1024
val disruptor = new Disruptor[util.Map[String, Object]](new MessageEventFactory, ringBufferSize, new MessageThreadFactory, ProducerType.MULTI, new BlockingWaitStrategy)

      disruptor.handleEventsWith(new MessageEventHandler(batchSize, this))
      disruptor.setDefaultExceptionHandler(new MessageExceptionHandler)
      val ringBuffer = disruptor.start
      val producer = new MessageEventProducer(ringBuffer)
 part.foreach { row =>
//        Thread.sleep(2000)
        accm.add(1)
        producer.onData(row)

//        flush(row)
      }

我想找到一种方法来自己控制disruptor的batch size，有没有什么方法可以消耗我的应用程序结束时保存的其余数据？

【问题讨论】：

你见过this question吗？ @jasonk 建议的方法之一可能会在您的情况下使用。 // 另外，如果您的主要问题是生产者在整个“批次”被消耗之前无法发布第 1025 个事件，您可能需要查看EarlyReleaseHandler 示例。
@Michael Barker 的This answer 还展示了一个较短的SequenceReportingEventHandler 实现示例。

标签： disruptor-pattern

【解决方案1】：

如果您让您的应用程序突然结束，您的消费者当然也会突然结束。无需减慢生产者的速度，您只需阻止您的应用程序退出，直到所有消费者（即事件处理程序）完成处理未完成的事件。

执行此操作的正常方法是在主线程上调用 Disruptor.shutdown()，从而阻止应用程序退出，直到 Disruptor.shutdown() 返回。

在您上面的代码片段中，您将在 part.foreach 语句之后退出例程之前添加该命令，阻塞直到例程正常返回。这将确保所有事件都得到妥善处理。

Disruptor 主要擅长缓冲（平滑）来自单个（极快）或多个（仍然相当快）生产者线程的数据突发，以将该数据提供给以可预测方式执行的消费者，从而消除由于锁争用，尽可能多的延迟和开销。您可能会发现，如果您的生产者实际上比您的消费者快得多，那么简单地从您的 lambda 中调用消费者代码可能会产生更好或相似的结果，除非您使用诸如批处理或设置 Disruptor 之类的高级技术来运行并行线程中的相同消费者，这需要修改事件处理程序实现（请参阅the Disruptor FAQ）。

在您的示例中，您似乎要做的只是将一组已经可用的数据（您的“部分”集合）提供给单个事件处理程序（MessageEventHandler）。在这样的用例中，你可能会更好地说出类似parts.stream().parallel().foreach(... messageEventHanler.onEvent(event) ...)的内容

【讨论】：