基于 CAS 无锁实现的 Disruptor.NET 居然慢于 BlockingCollection,是真的吗?

基于 CAS 无锁实现的 Disruptor.NET 居然慢于 BlockingCollection,是真的吗?

StackOverflow 有人说自己的 Disruptor.NET 代码比 BlockingCollection 还有慢 2 倍,并且把完整代码贴出,楼下几个老外也的回复说了一堆,但是没研究出个所以然来,讨论到最后甚至说可能你的场景不适合 Disruptor,我对此表示怀疑,BlockingCollection 内部实现比较简单粗暴,必要时就加锁,取数据时用信号量等待添加操作完成,而 Disruptor 是专门针对 CPU 缓存的特性优化过的,内部没有锁只有 CAS 原子操作,而且还考虑到了 false sharing,因此理论上 Disruptor 不会比 BlockingCollection 慢。

 

可是既然实际应用上出现问题,那就要分析下原因了。

把他的代码弄下来看了一下,问题多多啊。

在 Disruptor EventHandler 里面不定时调用 Console.WriteLine ,但是在 BlockingCollection 的 Handler 里面却只是记录了数据, Console.WriteLine 内部可是有锁的,调用的开销很大,如何能取得公平的结果呢?

另外 RingBuffer 的 Size 太小只有 64,严重影响 Disruptor 的表现,实际测试对比下来,应该 1024 或更大。

还有 BlockingCollection 里面的 while (!dataItems.IsCompleted) 写的也有问题,即使 BlockingCollection Producer 在循环中一直做添加操作,BlockingCollection 内部状态也并不是一直在添加状态中,这样导致添加循环还没做完,可是计时器的循环已经提前结束,导致 BlockingCollection 测得时间少于实际时间。

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 (这里插一句题外话,Disruptor .NET 版本的速度全面快于 Java 版本,不少场景下的速度比 Java 版本要快 10 倍,.NET 版是从 Java 移植过来的实现也和 Java 保持一直,是哪些语言特性导致性能差异这么大呢?)。

 

然后我拿着实测的结果和修改后的代码,在 Stackoverflow 上这个问题下面贴上了我的回答:

I read the BlockingCollecton code, You add many Console.WriteLine in Disruptor but no one in BlockingCollection, Console.WriteLine is slow, it have a lock inside.

Your RingBufferSize is too small, this effects performance, shoule be 1024 or larger.

and while (!dataItems.IsCompleted) may have some problem, BlockCollection isn’t always in adding state, it will cause thread ends early.

I have rewrite you code, Disruptor is 10x faster than BlockingCollection with multi producer (10 parallel producet), 2x faster than BlockingCollection with Single producer:

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"); } } }
 

BlockingCollectionTest with a shared ValueEntry instance (no new ValueEntry() in for loop)

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

    DisruptorTest Time:16.962s

    BlockingCollectionTest Time:18.399

  • RingBufferAndCapacitySize:512, producerCount:10, runCount:1000000 of 0 DisruptorTest Time:6.101s

    BlockingCollectionTest Time:19.526

  • RingBufferAndCapacitySize:1024, producerCount:10, runCount:1000000 of 0

    DisruptorTest Time:2.928s

    BlockingCollectionTest Time:20.25

  • RingBufferAndCapacitySize:2048, producerCount:10, runCount:1000000 of 0

    DisruptorTest Time:2.448s

    BlockingCollectionTest Time:20.649

BlockingCollectionTest create a new ValueEntry() in for loop

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

    DisruptorTest Time:27.374s

    BlockingCollectionTest Time:21.955

  • RingBufferAndCapacitySize:512, producerCount:10, runCount:1000000 of 0

    DisruptorTest Time:5.011s

    BlockingCollectionTest Time:20.127

  • RingBufferAndCapacitySize:1024, producerCount:10, runCount:1000000 of 0

    DisruptorTest Time:2.877s

    BlockingCollectionTest Time:22.656

  • RingBufferAndCapacitySize:2048, producerCount:10, runCount:1000000 of 0

    DisruptorTest Time:2.384s

    BlockingCollectionTest Time:23.567

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/119604.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • ElasticSearch数据库安装与配置[通俗易懂]

    ElasticSearch数据库安装与配置[通俗易懂]1、安装去官网下载安装包:https://www.elastic.co/cn/downloads/elasticsearch解压改压缩包就行,解压后的目录如下2、配置进入bin目录,双击elasticsearch.bat即可启动数据库服务。在浏览器中输入localhost:9200即可查看是否启动成功:显示上面的json串则表示启动成功。下面介绍如何配置可远程连接:需要修改config目录下的elasticsearch.yml文件在elastcisearch.yml文件中加上如下配

  • 计算机中1kb表示的字节数是多少_在计算机中1KB是指几字节

    计算机中1kb表示的字节数是多少_在计算机中1KB是指几字节大家好,我是时间财富网智能客服时间君,上述问题将由我为大家进行解答。1kb等于1024个字节。kB(Kilobyte),是一种资讯计量单位,是计算机数据存贮器存储单位字节的多倍形式。现今通常在标识内存等具有一般容量的储存媒介之储存容量时使用。根据国际单位制标准,1kB=1000B(字节,Byte)。根据按照IEC命名标准,用于二进制存储单位的标准命名是KiB,MiB等,1kiB=1024B。这是由…

  • 小Y的难题

    小Y的难题

  • 解决SqlTransaction用尽的问题

    解决SqlTransaction用尽的问题解决SqlTransaction用尽的问题有时候程序处理的数据量比较小时,四平八稳,一切安然无恙,但数据量一大,原先潜伏的问题就暴露无遗了。我做的一个项目,是负责一个厂的考勤的。厂里有员工1000多号人。按每人每天打4次卡,一个月30天,则产生的考勤记录数目为1000*4*30=120,000条。在处理这些记录时,我采用的办法是先生成SQL语句,然后执行这些SQL语句:Sql…

  • atitit.404错误调查过程汇总

    atitit.404错误调查过程汇总

  • BT Tracker服务器_bt服务器

    BT Tracker服务器_bt服务器有时候用BT软件下载文件的时候,经常会碰到没速度的情况,这个时候你就需要设置BTTracker服务器地址了。网上有人专门做了一个列表放在github上面,不定时更新。BitComet软件可以直接添加URL,然后设置启动时更新。项目地址:https://github.com/ngosang/trackerslist(长期更新)trackers_best(20tracker…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号