具有1个发布者和4个并行消费者的中断器示例

Question

在本例中https://stackoverflow.com/a/9980346/93647为什么我的破坏者的例子如此缓慢？（在问题的末尾）有一个发布项目的出版商和一个消费者。

但是在我的例子中，消费者的工作要复杂得多，需要一些时间。所以我想要4个并行处理数据的消费者。

例如，如果生产者生产数字：1,2,3,4,5,6,7,8,9,10,11...

我想让消费者1抓住1,5,9，。。。消费者2捕捉2,6,10，。。。消费者要抓住3,7,11，。。。消费者4要抓住4,8,12。。。（好吧，不完全是这些数字，想法是数据应该并行处理，我不在乎哪个消费者处理哪个特定数字）

请记住，这需要并行完成，因为在实际html" target="_blank">应用程序中，消费者的工作非常昂贵。我希望消费者在不同的线程中执行，以使用多核系统的功能。

当然，我可以创建4个环形缓冲区，并将1个消费者连接到1个环形缓冲区。这样我就可以使用原始示例。但我觉得这是不对的。可能创建1个发布服务器（1个ringbuffer）和4个消费者是正确的，因为这正是我所需要的。

在google群组中添加一个非常相似的问题的链接：https://groups.google.com/forum/#!消息/lmax干扰器/-CLapWuwWLU/GHEP4UkxrAEJ

因此，我们有两种选择：

• 一环多个消费者（每个消费者在每次添加时都会“醒来”，所有消费者都应该有相同的等待策略）

Answer 1

从环形缓冲区的规格中，您将看到每个消费者都会尝试处理您的ValueEvent。在您的情况下，您不需要它。

我是这样解决的：

将处理过的字段添加到您的ValueEvent中，当消费者获取事件时，他会在该字段上进行测试，如果已经处理过，他会转到下一个字段。

不是最漂亮的方式，但这就是缓冲区的工作方式。

Answer 2

编辑：我忘了提到代码部分取自常见问题解答。我不知道这种方法是比弗兰克的建议好还是坏。

该项目的文档严重不足，这是一个遗憾，因为它看起来不错
无论如何，请尝试以下剪贴（基于您的第一个链接）-在mono上测试，似乎还可以：

using System;
using System.Threading.Tasks;
using Disruptor;
using Disruptor.Dsl;

namespace DisruptorTest
{
    public sealed class ValueEntry
    {
        public long Value { get; set; }
    }

    public class MyHandler : IEventHandler<ValueEntry>
    {
        private static int _consumers = 0;
        private readonly int _ordinal;

        public MyHandler()
        {
            this._ordinal = _consumers++;
        }

        public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
        {
            if ((sequence % _consumers) == _ordinal)
                Console.WriteLine("Event handled: Value = {0}, event {1} processed by {2}", data.Value, sequence, _ordinal);
            else
                Console.WriteLine("Event {0} rejected by {1}", sequence, _ordinal);                     
        }
    }

    class Program
    {
        private static readonly Random _random = new Random();
        private const int SIZE = 16;  // Must be multiple of 2
        private const int WORKERS = 4; 

        static void Main()
        {
            var disruptor = new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), SIZE, TaskScheduler.Default);
            for (int i=0; i < WORKERS; i++)
                disruptor.HandleEventsWith(new MyHandler());
            var ringBuffer = disruptor.Start();

            while (true)
            {
                long sequenceNo = ringBuffer.Next();
                ringBuffer[sequenceNo].Value =  _random.Next();;
                ringBuffer.Publish(sequenceNo);
                Console.WriteLine("Published entry {0}, value {1}", sequenceNo, ringBuffer[sequenceNo].Value);
                Console.ReadKey();
            }
        }
    }
}