文档章节

一起聊聊,JAVA高性能框架Disruptor

小AN
 小AN
发布于 2017/07/19 15:42
字数 2853
阅读 241
收藏 7

Disruptor 的核心概念

disruptor最大特点是高性能,其针对性能做出极度优化并采用了无锁式的设计。

  • Ring Buffer
    Ring Buffer是一个类似对象池模型的实现,所有的”消息“都保存在里面,解决“消息”对象生存周期短、数量多、回收频繁的问题,在高级的应用场景中,Ring Buffer 可由用户自定义实现来替代。

  • Sequence Disruptor
    通过顺序递增的序号来编号管、定位、处理“消息”。同时用于跟踪标识某个事件处理者( RingBuffer/Consumer )的处理进度。Sequence 同时负责处理CPU缓存伪共享(Flase Sharing)问题。

  • Sequencer 
    Sequencer 是 Disruptor 的真正核心。此接口有两个实现类 SingleProducerSequencer、MultiProducerSequencer ,它们定义在生产者和消费者之间快速、正确地传递数据的并发算法。

  • Sequence Barrier
    用于保持对RingBuffer的 main published Sequence 和Consumer依赖的其它Consumer的 Sequence 的引用。 Sequence Barrier 还定义了决定 Consumer 是否还有可处理的事件的逻辑。

  • Wait Strategy
    定义 “消费者”如何进行等待下一个事件的策略。

  • Event
    “生产者”和“消费者”之间传递的“消息”

  • EventProcessor
    EventProcessor 持有特定消费者(Consumer)的 Sequence,并提供用于调用事件处理实现的事件循环(Event Loop)。

  • EventHandler
    处理“消息”内容的实现。

  • Producer
    生产者

 

 

三、disruptor为什么这么快

Ring Buffer:

Ring Buffer是一个存储数据的环形数组结构,在这里“消息”被按顺装填,不会销毁,直到达到尾部后会进行覆盖。

首先,环形模式的最大优点在于没有尾指针,只需维护下一个位置的序号,与链表结构相比,数组的性能更高。

其次,数据内部元素的内存地址连续存储,在硬件级别,cpu在加载元素时相邻元素会被预加载

再次,数组的内存是一直存在的(直到程序退出),内部的元素也会一直存在,避免了大量创建的开销以及垃圾回收时间。

 

 

 

False Sharing:

当CPU的缓存行(cache line) 中同时读取了两个相互独立的数据时,当两份数据分别要进行更新时,根据MESI协议会产生RFO(Request For Ownership)请求,双方会竞争当前缓存行的所有权,导致另外一份数据缓存失效,此时就是伪共享(False Sharing)。Disruptor针对这种问题采用了缓存行填充的方式,将数据补充放大,保证单独缓存在一个缓存行中,避免伪共享带来的性能影响。

 

Lock:

锁的存在会不可避免的性能开销,不合理的设计会导致死锁产生,增加设计成本、降低运行效率。

 

首先,Dispruptor 采用无锁化设计,在多生产者条件下使用CAS(Compare And Swap/Set)进行操作。这是一个CPU级别指令,虽然它们并非没有代价,但比锁消耗资源少的多。

其次,Sequence会为每一个“消息”由同一个线程产生一个序号,避免出现多个线程之间的竞争修改,没有竞争、不需要锁、甚至不需要CAS

 

Memory Barriers:

内存屏障能够确保一些特定操作执行的顺序,影响一些数据的可见性(可能是某些指令执行后的结果)。Java内存模型在处理volatile时写操作后插入一个写屏障指令,在读操作前插入一个读屏障指令,Disruptor对volatile字段(cursor)的写操作创建了一个内存屏障,这个屏障将刷新所有缓存里的值(或者至少相应地使得缓存失效),保证数据准确性。

 

四、disruptor的例子:

 

step 1:Event

 

 
package com.bj58.disruptor.demo;

public class Message
{
private long value;

   public void set(long value)
{
this.value = value;
}
}

 

 
package com.bj58.disruptor.demo;

import com.lmax.disruptor.EventFactory;

public class MessageFactory implements EventFactory<Message>
{
public Message newInstance()
{
return new Message();
}
}
step 2: EventHandler
 
 
package com.bj58.disruptor.demo;

import com.lmax.disruptor.EventHandler;

public class MessageHandler implements EventHandler<Message>
{
public void onEvent(Message event, long sequence, boolean endOfBatch)
{
System.out.println("Event: " + event);
   }
}
step 3: Producer
 
 
package com.bj58.disruptor.demo;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.EventTranslatorOneArg;

import java.nio.ByteBuffer;

public class MessageProducer
{
private final RingBuffer<Message> ringBuffer;

   public MessageProducer(RingBuffer<Message> ringBuffer)
{
this.ringBuffer = ringBuffer;
   }

private static final EventTranslatorOneArg<Message, ByteBuffer> TRANSLATOR =
new EventTranslatorOneArg<Message, ByteBuffer>()
{
public void translateTo(Message event, long sequence, ByteBuffer bb)
{
event.set(bb.getLong(0));
           }
};

   public void onData(ByteBuffer bb)
{
ringBuffer.publishEvent(TRANSLATOR, bb);
   }
}
 
step 4 : Main
 
package com.bj58.disruptor.demo;

import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.RingBuffer;
import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class Main
{
public static void main(String[] args) throws Exception
{
// Executor that will be used to construct new threads for consumers
Executor executor = Executors.newCachedThreadPool();

// The factory for the event
MessageFactory factory = new MessageFactory();

// Specify the size of the ring buffer, must be power of 2.
int bufferSize = 1024;

// Construct the Disruptor
Disruptor<Message> disruptor = new Disruptor<Message>(factory, bufferSize, executor);

// Connect the handler
disruptor.handleEventsWith(new MessageHandler());

// Start the Disruptor, starts all threads running
disruptor.start();

// Get the ring buffer from the Disruptor to be used for publishing.
RingBuffer<Message> ringBuffer = disruptor.getRingBuffer();

MessageProducer producer = new MessageProducer(ringBuffer);

ByteBuffer bb = ByteBuffer.allocate(8);
       for (long l = 0; true; l++)
{
bb.putLong(0, l);
producer.onData(bb);
Thread.sleep(1000);
}
}
}

 

五、disruptor 的性能:

官方数据:

Multiple Producer

Disruptor=26,553,372 ops/sec

 

Single Producer

Disruptor=89,365,504 ops/sec
数据对比:
处理的消息数量:32*1024*1024=33554432
 
Duration(ms) LinkedBlockingQueue Disruptor 对比
1 5771 4209 73%
2 5725 3479 60.77%
3 5433 3228 59.41%
4 5297 2805 52.95%
5 5745 2477 43.12%
6 6124 2428 39.65%

分析:
当线程数=Cpu核心数时,LinkedblockQueue处理速度达到最高,Disruptor随线程数增加处理速度变快,总体响应时间Disruptor优势明显
 
QPS

Disruptor 的核心概念

disruptor最大特点是高性能,其针对性能做出极度优化并采用了无锁式的设计。

  • Ring Buffer
    Ring Buffer是一个类似对象池模型的实现,所有的”消息“都保存在里面,解决“消息”对象生存周期短、数量多、回收频繁的问题,在高级的应用场景中,Ring Buffer 可由用户自定义实现来替代。

  • Sequence Disruptor
    通过顺序递增的序号来编号管、定位、处理“消息”。同时用于跟踪标识某个事件处理者( RingBuffer/Consumer )的处理进度。Sequence 同时负责处理CPU缓存伪共享(Flase Sharing)问题。

  • Sequencer 
    Sequencer 是 Disruptor 的真正核心。此接口有两个实现类 SingleProducerSequencer、MultiProducerSequencer ,它们定义在生产者和消费者之间快速、正确地传递数据的并发算法。

  • Sequence Barrier
    用于保持对RingBuffer的 main published Sequence 和Consumer依赖的其它Consumer的 Sequence 的引用。 Sequence Barrier 还定义了决定 Consumer 是否还有可处理的事件的逻辑。

  • Wait Strategy
    定义 “消费者”如何进行等待下一个事件的策略。

  • Event
    “生产者”和“消费者”之间传递的“消息”

  • EventProcessor
    EventProcessor 持有特定消费者(Consumer)的 Sequence,并提供用于调用事件处理实现的事件循环(Event Loop)。

  • EventHandler
    处理“消息”内容的实现。

  • Producer
    生产者

 

 

三、disruptor为什么这么快

Ring Buffer:

Ring Buffer是一个存储数据的环形数组结构,在这里“消息”被按顺装填,不会销毁,直到达到尾部后会进行覆盖。

首先,环形模式的最大优点在于没有尾指针,只需维护下一个位置的序号,与链表结构相比,数组的性能更高。

其次,数据内部元素的内存地址连续存储,在硬件级别,cpu在加载元素时相邻元素会被预加载

再次,数组的内存是一直存在的(直到程序退出),内部的元素也会一直存在,避免了大量创建的开销以及垃圾回收时间。

 

 

 

False Sharing:

当CPU的缓存行(cache line) 中同时读取了两个相互独立的数据时,当两份数据分别要进行更新时,根据MESI协议会产生RFO(Request For Ownership)请求,双方会竞争当前缓存行的所有权,导致另外一份数据缓存失效,此时就是伪共享(False Sharing)。Disruptor针对这种问题采用了缓存行填充的方式,将数据补充放大,保证单独缓存在一个缓存行中,避免伪共享带来的性能影响。

 

Lock:

锁的存在会不可避免的性能开销,不合理的设计会导致死锁产生,增加设计成本、降低运行效率。

 

首先,Dispruptor 采用无锁化设计,在多生产者条件下使用CAS(Compare And Swap/Set)进行操作。这是一个CPU级别指令,虽然它们并非没有代价,但比锁消耗资源少的多。

其次,Sequence会为每一个“消息”由同一个线程产生一个序号,避免出现多个线程之间的竞争修改,没有竞争、不需要锁、甚至不需要CAS

 

Memory Barriers:

内存屏障能够确保一些特定操作执行的顺序,影响一些数据的可见性(可能是某些指令执行后的结果)。Java内存模型在处理volatile时写操作后插入一个写屏障指令,在读操作前插入一个读屏障指令,Disruptor对volatile字段(cursor)的写操作创建了一个内存屏障,这个屏障将刷新所有缓存里的值(或者至少相应地使得缓存失效),保证数据准确性。

 

四、disruptor的例子:

 

step 1:Event

 

 
package com.bj58.disruptor.demo;

public class Message
{
private long value;

   public void set(long value)
{
this.value = value;
}
}

 

 
package com.bj58.disruptor.demo;

import com.lmax.disruptor.EventFactory;

public class MessageFactory implements EventFactory<Message>
{
public Message newInstance()
{
return new Message();
}
}
step 2: EventHandler
 
 
package com.bj58.disruptor.demo;

import com.lmax.disruptor.EventHandler;

public class MessageHandler implements EventHandler<Message>
{
public void onEvent(Message event, long sequence, boolean endOfBatch)
{
System.out.println("Event: " + event);
   }
}
step 3: Producer
 
 
package com.bj58.disruptor.demo;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.EventTranslatorOneArg;

import java.nio.ByteBuffer;

public class MessageProducer
{
private final RingBuffer<Message> ringBuffer;

   public MessageProducer(RingBuffer<Message> ringBuffer)
{
this.ringBuffer = ringBuffer;
   }

private static final EventTranslatorOneArg<Message, ByteBuffer> TRANSLATOR =
new EventTranslatorOneArg<Message, ByteBuffer>()
{
public void translateTo(Message event, long sequence, ByteBuffer bb)
{
event.set(bb.getLong(0));
           }
};

   public void onData(ByteBuffer bb)
{
ringBuffer.publishEvent(TRANSLATOR, bb);
   }
}
 
step 4 : Main
 
package com.bj58.disruptor.demo;

import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.RingBuffer;
import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class Main
{
public static void main(String[] args) throws Exception
{
// Executor that will be used to construct new threads for consumers
Executor executor = Executors.newCachedThreadPool();

// The factory for the event
MessageFactory factory = new MessageFactory();

// Specify the size of the ring buffer, must be power of 2.
int bufferSize = 1024;

// Construct the Disruptor
Disruptor<Message> disruptor = new Disruptor<Message>(factory, bufferSize, executor);

// Connect the handler
disruptor.handleEventsWith(new MessageHandler());

// Start the Disruptor, starts all threads running
disruptor.start();

// Get the ring buffer from the Disruptor to be used for publishing.
RingBuffer<Message> ringBuffer = disruptor.getRingBuffer();

MessageProducer producer = new MessageProducer(ringBuffer);

ByteBuffer bb = ByteBuffer.allocate(8);
       for (long l = 0; true; l++)
{
bb.putLong(0, l);
producer.onData(bb);
Thread.sleep(1000);
}
}
}

 

五、disruptor 的性能:

官方数据:

Multiple Producer

Disruptor=26,553,372 ops/sec

 

Single Producer

Disruptor=89,365,504 ops/sec
数据对比:
处理的消息数量:32*1024*1024=33554432
 
Duration(ms) LinkedBlockingQueue Disruptor 对比
1 5771 4209 73%
2 5725 3479 60.77%
3 5433 3228 59.41%
4 5297 2805 52.95%
5 5745 2477 43.12%
6 6124 2428 39.65%
分析:
当线程数=Cpu核心数时,LinkedblockQueue处理速度达到最高,Disruptor随线程数增加处理速度变快,总体响应时间Disruptor优势明显
 
QPS LinkedBlockingQueue Disruptor 对比
1 5814318 7972067 137%
2 5861036 9644850 165%
3 6176041 10394805 168%
4 6334611 11962364 189%
5 5840632 13546400 232%
6 5479169 13819783 252%
分析:
当线程数较少时,LinkedblockedQueue和Disruptor性能差距并不明显,随着线程数量提升,Disruptor性能优势逐渐变大。
Disruptor 对比
1 5814318 7972067 137%
2 5861036 9644850 165%
3 6176041 10394805 168%
4 6334611 11962364 189%
5 5840632 13546400 232%
6 5479169 13819783 252%

分析:
当线程数较少时,LinkedblockedQueue和Disruptor性能差距并不明显,随着线程数量提升,Disruptor性能优势逐渐变大。

多说一句,希望或的更多的学习资料的同学可以加群:647631030

点击链接加入群【JAVA高级开发】:https://jq.qq.com/?_wv=1027&k=4C8wH5s

© 著作权归作者所有

小AN
粉丝 1
博文 4
码字总数 9140
作品 0
长沙
私信 提问
加载中

评论(1)

小AN
小AN
点击链接加入群【JAVA高级开发】:https://jq.qq.com/?_wv=1027&k=4C8wH5s
【微服务框架到生态】Apache Dubbo 开发者沙龙 南京站

Dubbo 诞生于 2008 年,是阿里巴巴开源的高性能分布式服务框架(A High Performance Java RPC Framework),使得应用可通过高性能的 RPC 实现服务的输出和输入功能,可以和 Spring 框架无缝集...

AlibabaInfra
03/07
0
0
(转)disruptor - Concurrent Programming Framework...

disruptor发布了Java的2.0版本(.Net版本见这里),disruptor是一个高性能的异步处理框架,或者可以认为是最快的消息框架(轻量的JMS),也可以认为是一个观察者模式实现,或者事件-监听模式的实...

猴子
2011/10/23
0
0
报名 | Apache Dubbo™ 开发者沙龙@广州

Dubbo 诞生于 2008 年,是阿里巴巴开源的高性能分布式服务框架(A High Performance Java RPC Framework),使得应用可通过高性能的 RPC 实现服务的输出和输入功能,可以和 Spring 框架无缝集...

amber涂南
01/03
0
0
Disruptor 3.3.10 和 3.4.1 发布,并发编程框架

Disruptor 是一个 Java 的并发编程框架,大大的简化了并发程序开发的难度,在性能上也比 Java 本身提供的一些并发包要好。 Disruptor 3.3.10 和 3.4.1 均修复了以下 bug : Fix race between...

淡漠悠然
2018/03/08
1K
0
ApacheDubbo:tm: 开发者沙龙 广州站

Dubbo 诞生于 2008 年,是阿里巴巴开源的高性能分布式服务框架(A High Performance Java RPC Framework),使得应用可通过高性能的 RPC 实现服务的输出和输入功能,可以和 Spring 框架无缝集...

AlibabaInfra
01/07
0
0

没有更多内容

加载失败,请刷新页面

加载更多

我为什么要写微信公众号

埋一颗种子,细心呵护,静待她枝繁叶茂,葱郁参天 V2论坛上有个帖子【做程序员最重要的还是一定要有自己的作品】,作者写道: 能有一个作品和你的名字联系在一起,应当成为在职业生涯前期着意...

运维咖啡吧
37分钟前
3
0
数据库

数据库架构 数据库架构可以分为存储文件系统和程序实例两大块,而程序实例根据不同的功能又可以分为如下小模块。 1550644570798 索引模块 常见的问题有: 为什么要使用索引 什么样的信息能成...

一只小青蛙
今天
5
0
PHP常用经典算法实现

<? //-------------------- // 基本数据结构算法 //-------------------- //二分查找(数组里查找某个元素) function bin_sch($array, $low, $high, $k){ if ( $low <= $high){ $mid = int......

半缘修道半缘君丶
昨天
5
0
GIL 已经被杀死了么?

本文原创并首发于公众号【Python猫】,未经授权,请勿转载。 原文地址:https://mp.weixin.qq.com/s/8KvQemz0SWq2hw-2aBPv2Q 花下猫语: Python 中最广为人诟病的一点,大概就是它的 GIL 了。...

豌豆花下猫
昨天
6
0
git commit message form

commit message一般包括3部分:Header、Body、Footer。 <type>(<scope>):<subject>blank line<body>blank line<footer> header是必需的,body、footer可以省略。 header中type、subject......

ninjaFrog
昨天
5
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部