为什么我的破坏者示例如此缓慢？答案

【问题标题】：Why is my disruptor example so slow?为什么我的破坏者示例如此缓慢？
【发布时间】：2012-11-11 19:49:43
【问题描述】：

我从 Stack Overflow 问题 Disruptor.NET example 中获取了代码示例，并将其修改为“测量”时间。完整列表如下：

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

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

        public ValueEntry()
        {
            Console.WriteLine("New ValueEntry created");
        }
    }

    public class ValueAdditionHandler : IEventHandler<ValueEntry>
    {
        public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
        {
            Program.sw.Stop();
            long microseconds = Program.sw.ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
            Console.WriteLine("elapsed microseconds = " + microseconds);
            Console.WriteLine("Event handled: Value = {0} (processed event {1}", data.Value, sequence);
        }
    }

    class Program
    {
        public static Stopwatch sw = Stopwatch.StartNew();

        private static readonly Random _random = new Random();
        private static readonly int _ringSize = 16;  // Must be multiple of 2

        static void Main(string[] args)
        {
            var disruptor = new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), _ringSize, TaskScheduler.Default);

            disruptor.HandleEventsWith(new ValueAdditionHandler());

            var ringBuffer = disruptor.Start();

            while (true)
            {
                var valueToSet = _random.Next();
                long sequenceNo = ringBuffer.Next();

                ValueEntry entry = ringBuffer[sequenceNo];

                entry.Value = valueToSet;

                sw.Restart();
                ringBuffer.Publish(sequenceNo);

                Console.WriteLine("Published entry {0}, value {1}", sequenceNo, entry.Value);

                Thread.Sleep(1000);
            }
        }
    }
}

输出是：

New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
New ValueEntry created
Published entry 0, value 1510145842
elapsed microseconds = 2205
Event handled: Value = 1510145842 (processed event 0
Published entry 1, value 1718075893
elapsed microseconds = 85
Event handled: Value = 1718075893 (processed event 1
Published entry 2, value 1675907645
elapsed microseconds = 32
Event handled: Value = 1675907645 (processed event 2
Published entry 3, value 1563009446
elapsed microseconds = 75
Event handled: Value = 1563009446 (processed event 3
Published entry 4, value 1782914062
elapsed microseconds = 34
Event handled: Value = 1782914062 (processed event 4
Published entry 5, value 1516398244
elapsed microseconds = 50
Event handled: Value = 1516398244 (processed event 5
Published entry 6, value 76829327
elapsed microseconds = 50
Event handled: Value = 76829327 (processed event 6

因此，将数据从一个线程传递到另一个线程大约需要 50 微秒。但它一点也不快！ “当前版本的 Disruptor 可以在线程之间以每秒 100 万条消息的速度执行约 50 ns。”所以我的结果比预期慢 1000 倍。

我的示例有什么问题，如何实现 50 ns 的速度？

我已经修改了上面的程序，现在收到 1 微秒的延迟，这要好得多。但是，我仍在等待disruptor 模式专家的回复。我正在寻找一个可以证明我实际上可以在 50 ns 内传递数据的示例。

另外，我使用BlockingCollection 编写了相同的测试，平均收到14 微秒，这证明Disruptor 更快：

使用 BlockingCollection：

average = 14 minimum = 0 0-5 = 890558, 5-10 = 1773781, 10-30 = 6900128, >30 = 435433

使用干扰器：

average = 0 minimum = 0 0-5 = 9908469, 5-10 = 64464, 10-30 = 19902, >30 = 7065

BlockingCollection 代码：

using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

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

        public ValueEntry()
        {
            //   Console.WriteLine("New ValueEntry created");
        }
    }

    //public class ValueAdditionHandler : IEventHandler<ValueEntry>
    //{
    //    public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
    //    {

    //        long microseconds = Program.sw[data.Value].ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
    //        Program.results[data.Value] = microseconds;
    //        //Console.WriteLine("elapsed microseconds = " + microseconds);
    //        //Console.WriteLine("Event handled: Value = {0} (processed event {1}", data.Value, sequence);
    //    }
    //}

    class Program
    {
        public const int length = 10000000;
        public static Stopwatch[] sw = new Stopwatch[length];
        public static long[] results = new long[length];

        static BlockingCollection<ValueEntry> dataItems = new BlockingCollection<ValueEntry>(150);

        static void Main(string[] args)
        {
            for (int i = 0; i < length; i++)
            {
                sw[i] = Stopwatch.StartNew();
            }

            // A simple blocking consumer with no cancellation.
            Task.Factory.StartNew(() =>
            {
                while (!dataItems.IsCompleted)
                {

                    ValueEntry ve = null;
                    try
                    {
                        ve = dataItems.Take();
                        long microseconds = sw[ve.Value].ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
                        results[ve.Value] = microseconds;

                        //Console.WriteLine("elapsed microseconds = " + microseconds);
                        //Console.WriteLine("Event handled: Value = {0} (processed event {1}", ve.Value, ve.Value);
                    }
                    catch (InvalidOperationException) { }
                }
            }, TaskCreationOptions.LongRunning);

            for (int i = 0; i < length; i++)
            {
                var valueToSet = i;

                ValueEntry entry = new ValueEntry();
                entry.Value = valueToSet;

                sw[i].Restart();
                dataItems.Add(entry);

                //Console.WriteLine("Published entry {0}, value {1}", valueToSet, entry.Value);
                //Thread.Sleep(1000);
            }

            // Wait until all events are delivered
            Thread.Sleep(5000);

            long average = 0;
            long minimum = 10000000000;
            int firstFive = 0;
            int fiveToTen = 0;
            int tenToThirty = 0;
            int moreThenThirty = 0;

            // Do not count first 100 items because they could be extremely slow
            for (int i = 100; i < length; i++)
            {
                average += results[i];
                if (results[i] < minimum)
                {
                    minimum = results[i];
                }
                if (results[i] < 5)
                {
                    firstFive++;
                }
                else if (results[i] < 10)
                {
                    fiveToTen++;
                }
                else if (results[i] < 30)
                {
                    tenToThirty++;
                } else
                {
                    moreThenThirty++;
                }
            }
            average /= (length - 100);
            Console.WriteLine("average = {0} minimum = {1} 0-5 = {2}, 5-10 = {3}, 10-30 = {4}, >30 = {5}", average, minimum, firstFive, fiveToTen, tenToThirty, moreThenThirty);
        }
    }
}

干扰代码：

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

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

        public ValueEntry()
        {
            //   Console.WriteLine("New ValueEntry created");
        }
    }

    public class ValueAdditionHandler : IEventHandler<ValueEntry>
    {
        public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
        {

            long microseconds = Program.sw[data.Value].ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
            Program.results[data.Value] = microseconds;
            //Console.WriteLine("elapsed microseconds = " + microseconds);
            //Console.WriteLine("Event handled: Value = {0} (processed event {1}", data.Value, sequence);
        }
    }

    class Program
    {
        public const int length = 10000000;
        public static Stopwatch[] sw = new Stopwatch[length];
        public static long[] results = new long[length];

        private static readonly Random _random = new Random();
        private static readonly int _ringSize = 1024;  // Must be multiple of 2

        static void Main(string[] args)
        {
            for (int i = 0; i < length; i++)
            {
                sw[i] = Stopwatch.StartNew();
            }

            var disruptor = new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), _ringSize, TaskScheduler.Default);

            disruptor.HandleEventsWith(new ValueAdditionHandler());

            var ringBuffer = disruptor.Start();

            for (int i = 0; i < length; i++)
            {
                var valueToSet = i;
                long sequenceNo = ringBuffer.Next();

                ValueEntry entry = ringBuffer[sequenceNo];

                entry.Value = valueToSet;

                sw[i].Restart();
                ringBuffer.Publish(sequenceNo);

                //Console.WriteLine("Published entry {0}, value {1}", sequenceNo, entry.Value);

                //Thread.Sleep(1000);
            }

            // wait until all events are delivered
            Thread.Sleep(5000);

            long average = 0;
            long minimum = 10000000000;
            int firstFive = 0;
            int fiveToTen = 0;
            int tenToThirty = 0;
            int moreThenThirty = 0;

            // Do not count first 100 items because they could be extremely slow
            for (int i = 100; i < length; i++)
            {
                average += results[i];
                if (results[i] < minimum)
                {
                    minimum = results[i];
                }
                if (results[i] < 5)
                {
                    firstFive++;
                }
                else if (results[i] < 10)
                {
                    fiveToTen++;
                }
                else if (results[i] < 30)
                {
                    tenToThirty++;
                }
                else
                {
                    moreThenThirty++;
                }
            }
            average /= (length - 100);
            Console.WriteLine("average = {0} minimum = {1} 0-5 = {2}, 5-10 = {3}, 10-30 = {4}, >30 = {5}", average, minimum, firstFive, fiveToTen, tenToThirty, moreThenThirty);
        }
    }
}

【问题讨论】：

您无法计算单个项目的经过时间，秒表（或任何计时器机制）没有足够的精度。您应该数以千计、数以百万计的项目来获得统计上正确的测量结果。
@Euphoric 秒表有足够的精度来计算微秒
不。此外，运行时环境、GC、打印到控制台的更改很容易破坏此类测试。只需先修复您的测试，然后再尝试查找代码错误。
您的测量结果仍然错误。计算整个运行，然后除以项目数。每个项目的计算时间是错误的。
@Euphoric 我该怎么做？然后我需要等待上一次运行完成后再开始新的运行。所以我需要某种同步，但这会破坏破坏者的想法。

标签： c# disruptor-pattern

【解决方案1】：

在这里，我修复了您的代码：

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

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

        public ValueEntry()
        {
         //   Console.WriteLine("New ValueEntry created");
        }
    }

    class Program
    {
        public const int length = 1000000;
        public static Stopwatch sw;

        private static readonly Random _random = new Random();
        private static readonly int _ringSize = 1024;  // Must be multiple of 2

        static void Main(string[] args)
        {
            sw = Stopwatch.StartNew();

            var disruptor = new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), _ringSize, TaskScheduler.Default);

            var ringBuffer = disruptor.Start();

            for (int i = 0; i < length; i++)
            {
                var valueToSet = i;
                long sequenceNo = ringBuffer.Next();

                ValueEntry entry = ringBuffer[sequenceNo];

                entry.Value = valueToSet;

                ringBuffer.Publish(sequenceNo);

                //Console.WriteLine("Published entry {0}, value {1}", sequenceNo, entry.Value);

                //Thread.Sleep(1000);
            }

            var elapsed = sw.Elapsed.Miliseconds();
            // wait until all events are delivered
            Thread.Sleep(10000);

            double average = /(double)length;
            Console.WriteLine("average = " + average);
        }
    }
}

这应该正确测试每个项目需要多长时间。

【讨论】：

这可能是错误的。您测试发布每个项目需要多长时间。我需要测试发布项目并在另一个线程中接收它需要多长时间。我认为ringBuffer.Publish(sequenceNo); 调用是异步的，这就是为什么你的代码只测量了一半的工作。可能有很多“订阅者”，我认为发布者不会等到所有订阅者处理数据。

【解决方案2】：

我读了 BlockingCollection 的代码，你在 Disruptor 中添加了很多 Console.WriteLine 而在 BlockingCollection 中没有一个，Console.WriteLine 很慢，它里面有一个锁。

你的RingBufferSize太小，影响性能，应该是1024或更大。

而while (!dataItems.IsCompleted)可能有问题，BlockCollection并不总是处于adding状态，会导致线程提前结束。

Task.Factory.StartNew(() => {
    while (!dataItems.IsCompleted)
    {

        ValueEntry ve = null;
        try
        {
    ve = dataItems.Take();
    long microseconds = sw[ve.Value].ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
    results[ve.Value] = microseconds;

    //Console.WriteLine("elapsed microseconds = " + microseconds);
    //Console.WriteLine("Event handled: Value = {0} (processed event {1}", ve.Value, ve.Value);
        }
        catch (InvalidOperationException) { }
    }
}, TaskCreationOptions.LongRunning);


for (int i = 0; i < length; i++)
{
    var valueToSet = i;

    ValueEntry entry = new ValueEntry();
    entry.Value = valueToSet;

    sw[i].Restart();
    dataItems.Add(entry);

    //Console.WriteLine("Published entry {0}, value {1}", valueToSet, entry.Value);
    //Thread.Sleep(1000);
}

我已经重写了你的代码，Disruptor 比具有多个生产者（10 个并行生产者）的 BlockingCollection 快 10 倍，比具有单个生产者的 BlockingCollection 快 2 倍：

using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Disruptor;
using Disruptor.Dsl;
using NUnit.Framework;

namespace DisruptorTest.Ds
{
    public sealed class ValueEntry
    {
        internal int Id { get; set; }
    }

    class MyHandler : IEventHandler<ValueEntry>
    {
        public void OnEvent(ValueEntry data, long sequence, bool endOfBatch)
        {
        }
    }

    [TestFixture]
    public class DisruptorPerformanceTest
    {
        private volatile bool collectionAddEnded;

        private int producerCount = 10;
        private int runCount = 1000000;
        private int RingBufferAndCapacitySize = 1024;

        [TestCase()]
        public async Task TestBoth()
        {
            for (int i = 0; i < 1; i++)
            {
                foreach (var rs in new int[] {64, 512, 1024, 2048 /*,4096,4096*2*/})
                {
                    Console.WriteLine($"RingBufferAndCapacitySize:{rs}, producerCount:{producerCount}, runCount:{runCount} of {i}");
                    RingBufferAndCapacitySize = rs;
                    await DisruptorTest();
                    await BlockingCollectionTest();
                }
            }
        }

        [TestCase()]
        public async Task BlockingCollectionTest()
        {
            var sw = new Stopwatch();
            BlockingCollection<ValueEntry> dataItems = new BlockingCollection<ValueEntry>(RingBufferAndCapacitySize);

            sw.Start();

            collectionAddEnded = false;

            // A simple blocking consumer with no cancellation.
            var task = Task.Factory.StartNew(() =>
            {
                while (!collectionAddEnded && !dataItems.IsCompleted)
                {
                    //if (!dataItems.IsCompleted && dataItems.TryTake(out var ve))
                    if (dataItems.TryTake(out var ve))
                    {
                    }
                }
            }, TaskCreationOptions.LongRunning);


            var tasks = new Task[producerCount];
            for (int t = 0; t < producerCount; t++)
            {
                tasks[t] = Task.Run(() =>
                {
                    for (int i = 0; i < runCount; i++)
                    {
                        ValueEntry entry = new ValueEntry();
                        entry.Id = i;
                        dataItems.Add(entry);
                    }
                });
            }

            await Task.WhenAll(tasks);

            collectionAddEnded = true;
            await task;

            sw.Stop();

            Console.WriteLine($"BlockingCollectionTest Time:{sw.ElapsedMilliseconds/1000d}");
        }


        [TestCase()]
        public async Task DisruptorTest()
        {
            var disruptor =
                new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), RingBufferAndCapacitySize, TaskScheduler.Default,
                    producerCount > 1 ? ProducerType.Multi : ProducerType.Single, new BlockingWaitStrategy());
            disruptor.HandleEventsWith(new MyHandler());

            var _ringBuffer = disruptor.Start();

            Stopwatch sw = Stopwatch.StartNew();

            sw.Start();


            var tasks = new Task[producerCount];
            for (int t = 0; t < producerCount; t++)
            {
                tasks[t] = Task.Run(() =>
                {
                    for (int i = 0; i < runCount; i++)
                    {
                        long sequenceNo = _ringBuffer.Next();
                        _ringBuffer[sequenceNo].Id = 0;
                        _ringBuffer.Publish(sequenceNo);
                    }
                });
            }


            await Task.WhenAll(tasks);


            disruptor.Shutdown();

            sw.Stop();
            Console.WriteLine($"DisruptorTest Time:{sw.ElapsedMilliseconds/1000d}s");
        }
    }
}

具有共享 ValueEntry 实例的 BlockingCollectionTest（for 循环中没有新的 ValueEntry()）

RingBufferAndCapacitySize:64, producerCount:10, runCount:1000000 of 0

DisruptorTest 时间：16.962s

BlockingCollectionTest 时间：18.399
RingBufferAndCapacitySize:512, producerCount:10, runCount:1000000 of 0 DisruptorTest Time:6.101s

BlockingCollectionTest 时间：19.526
RingBufferAndCapacitySize:1024, producerCount:10, runCount:1000000 of 0

DisruptorTest Time:2.928s

BlockingCollectionTest 时间：20.25
RingBufferAndCapacitySize:2048, producerCount:10, runCount:1000000 of 0

DisruptorTest Time:2.448s

BlockingCollectionTest 时间：20.649

BlockingCollectionTest 在 for 循环中创建一个新的 ValueEntry()

RingBufferAndCapacitySize:64, producerCount:10, runCount:1000000 of 0

DisruptorTest 时间：27.374s

BlockingCollectionTest 时间：21.955
RingBufferAndCapacitySize:512, producerCount:10, runCount:1000000 of 0

DisruptorTest 时间：5.011s

BlockingCollectionTest 时间：20.127
RingBufferAndCapacitySize:1024, producerCount:10, runCount:1000000 of 0

DisruptorTest 时间：2.877s

BlockingCollectionTest 时间：22.656
RingBufferAndCapacitySize:2048, producerCount:10, runCount:1000000 of 0

DisruptorTest 时间：2.384s

BlockingCollectionTest 时间：23.567

https://www.cnblogs.com/darklx/p/11755686.html

【讨论】：