Disruptor.net的用法

Disruptor据说是最快的并发消息处理器。我今天学习了下，有了一个初步的了解。
Disruptor可以当做队列使用，不过可以支持多消费者，既一条生产者的消息可以提供给多个消费者同时使用。
gethub 安装 Install-Package Disruptor -Version 3.3.7

我们添加一个实体类，这个类的主要是生产者和消费者的信息载体

public sealed class CmdEntry
{
    public string Cmd { get; set; }
}

我们增加一个类实现IEventHandler接口，这个类的作用相当于消费者触发口，也可以当作消费者。

public class ExecuteCmdHandler : IEventHandler<CmdEntry>
{
    public void OnNext(CmdEntry data, long sequence, bool endOfBatch)
    {
        Console.WriteLine("this is CmdHandler0 Cmd = {0} (index:{1})", data.Cmd, sequence);

    }
}

然后我们添加一个窗体界面。并生成生产者信息。

public partial class MainWindow : Window
{

    RingBuffer<CmdEntry> _ringBuffer;
    public MainWindow()
    {
        InitializeComponent();
       var disruptor = new Disruptor<CmdEntry>(() => new CmdEntry(), 512, TaskScheduler.Default);

       disruptor.HandleEventsWith(new ExecuteCmdHandler());
       //disruptor.HandleEventsWith(new ExecuteCmdHandler1());

       _ringBuffer = disruptor.Start();


       Thread tr = new Thread(Product);
       tr.Start();
     
    }

    private void Product(object o)
    {

        int i = 0;
        while (i<500)
        {
            long index = _ringBuffer.Next();
            var entity = _ringBuffer[index];
            entity.Cmd = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff");
            _ringBuffer.Publish(index);
            Thread.Sleep(1);
            i = i + 1;
        }

    }

}

运行

image.png

如果我们有多个消费者，只需要在实现IEventHandler接口

public class ExecuteCmdHandler1 : IEventHandler<CmdEntry>
{
    public void OnNext(CmdEntry data, long sequence, bool endOfBatch)
    {
        Console.WriteLine("this is CmdHandler1 Cmd = {0} (index:{1})", data.Cmd, sequence);
    }
}

然后再通过Disruptor对象的HandleEventsWith方法添加进去

image.png

运行，从结果可以看出，两个消费者是并行处理。

image.png