Disruptor должен быть намного быстрее, чем BlockingCollection.
В моем предыдущем вопросе Почему мой пример разрушителя такой медленный? Я написал два теста. Disruptor потратил около 1 микросекунды (или меньше), в то время как BlockingCollection потратил около 14 микросекунд.
Итак, я решил использовать Disruptor в своей программе, но когда я его реализовал, я обнаружил, что теперь Disruptor тратит около 50 микросекунд, в то время как BlockingCollection по-прежнему тратит 14-18 микросекунд.
Я изменил свой производственный код, сделав его «автономным тестом», и Disruptor по-прежнему тратит 50 микросекунд. Почему?
Ниже приведен упрощенный тест. В этом тесте у меня есть два варианта. Первый вариант — Sleep for 1 ms. Затем Disruptor тратит 30-50 микросекунд на доставку. Второй вариант — имитировать деятельность. Затем Disruptor тратит 7 микросекунд на доставку. Тот же тест с BlockingCollection дает 14-18 микросекунд. Так почему же Disruptor не быстрее, чем BlockingCollection?
В моем реальном приложении Disruptor тратит 50 микросекунд, чтобы доставить то, что слишком много! Я ожидаю, что он должен доставлять сообщения намного быстрее, чем 1 микросекунда.
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Disruptor;
namespace DisruptorTest
{
public sealed class ValueEntry
{
internal int Id { get; set; }
}
class DisruptorTest
{
public class MyHandler : IEventHandler<ValueEntry>
{
private DisruptorTest _parent;
public MyHandler(DisruptorTest parent)
{
this._parent = parent;
}
public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
{
_parent.sw.Stop();
long microseconds = _parent.sw.ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
// Filter out abnormal delays > 1000
if (microseconds < 1000)
{
_parent.sum += (int)microseconds;
_parent.count++;
if (_parent.count % 1000 == 0)
{
Console.WriteLine("average disruptor delay (microseconds) = {0}", _parent.sum / _parent.count);
}
}
}
}
private RingBuffer<ValueEntry> _ringBuffer;
private const int RingSize = 64;
static void Main(string[] args)
{
new DisruptorTest().Run();
}
public void Run()
{
var disruptor = new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), RingSize, TaskScheduler.Default);
disruptor.HandleEventsWith(new MyHandler(this));
_ringBuffer = disruptor.Start();
for (int i = 0; i < 10001; i++)
{
Do();
// We need to simulate activity to allow event to deliver
// Option1. just Sleep. Result 30-50 microseconds.
Thread.Sleep(1);
// Option2. Do something. Result ~7 microseconds.
//factorial = 1;
//for (int j = 1; j < 100000; j++)
//{
// factorial *= j;
//}
}
}
public static int factorial;
private Stopwatch sw = Stopwatch.StartNew();
private int sum;
private int count;
public void Do()
{
long sequenceNo = _ringBuffer.Next();
_ringBuffer[sequenceNo].Id = 0;
sw.Restart();
_ringBuffer.Publish(sequenceNo);
}
}
}
СТАРЫЙ код. Теперь следует игнорировать:
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Disruptor;
namespace DisruptorTest
{
public sealed class ValueEntry
{
internal int Id { get; set; }
}
class DisruptorTest
{
public class MyHandler : IEventHandler<ValueEntry>
{
private readonly int _ordinal;
private readonly int _consumers;
private DisruptorTest _parent;
public MyHandler(int ordinal, int consumers, DisruptorTest parent)
{
_ordinal = ordinal;
_consumers = consumers;
this._parent = parent;
}
public void OnNext(ValueEntry data, long sequence, bool endOfBatch)
{
if ((sequence % _consumers) == _ordinal)
{
var id = data.Id;
_parent.sw[id].Stop();
long microseconds = _parent.sw[id].ElapsedTicks / (Stopwatch.Frequency / (1000L * 1000L));
// filter out abnormal delays > 1000
if (microseconds < 1000)
{
_parent.sum[id] += (int)microseconds;
_parent.count[id]++;
if (_parent.count[id] % 10 == 0)
{
Console.WriteLine("Id = {0} average disruptor delay (microseconds) = {1}",
id, _parent.sum[id] / _parent.count[id]);
}
}
}
}
}
private const int NumberOfThreads = 1;
private RingBuffer<ValueEntry> _ringBuffer;
private const int RingSize = 64;
static void Main(string[] args)
{
new DisruptorTest().Run();
}
public void Run()
{
var disruptor = new Disruptor.Dsl.Disruptor<ValueEntry>(() => new ValueEntry(), RingSize, TaskScheduler.Default);
for (int i = 0; i < NumberOfThreads; i++)
disruptor.HandleEventsWith(new MyHandler(i, NumberOfThreads, this));
for (int i = 0; i < sw.Length; i++)
{
sw[i] = Stopwatch.StartNew();
}
_ringBuffer = disruptor.Start();
//var rnd = new Random();
for (int i = 0; i < 1000; i++)
{
//Do(rnd.Next(MaxId));
Do(i % MaxId);
Thread.Sleep(1);
}
}
private const int MaxId = 100;
private Stopwatch[] sw = new Stopwatch[MaxId];
private int[] sum = new int[MaxId];
private int[] count = new int[MaxId];
public void Do(int id)
{
long sequenceNo = _ringBuffer.Next();
_ringBuffer[sequenceNo].Id = id;
sw[id].Restart();
_ringBuffer.Publish(sequenceNo);
}
}
}
Выход:
......
Id = 91 average disruptor delay (microseconds) = 50
Id = 92 average disruptor delay (microseconds) = 48
Id = 93 average disruptor delay (microseconds) = 35
Id = 94 average disruptor delay (microseconds) = 35
Id = 95 average disruptor delay (microseconds) = 51
Id = 96 average disruptor delay (microseconds) = 55
Id = 97 average disruptor delay (microseconds) = 38
Id = 98 average disruptor delay (microseconds) = 37
Id = 99 average disruptor delay (microseconds) = 45
