博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
Disruptor学习之路(一)
阅读量:7112 次
发布时间:2019-06-28

本文共 4497 字,大约阅读时间需要 14 分钟。

hot3.png

Disruptor是什么?

    Disruptor是一个高性能的异步处理框架,或者可以认为是线程间通信的高效低延时的内存消息组件,它最大特点是高性能,其LMAX架构可以获得每秒6百万订单,用1微秒的延迟获得吞吐量为100K+。

    它是如何实现高性能的呢?它由于JDK内置的队列有什么区别呢?

JDK内置内存队列?

我们知道,Java内置了几种内存消息队列,如下所示:

队列 加锁方式 是否有界 数据结构
ArrayBlockingQueue 加锁 有界 ArrayList
LinkedBlockingQueue 加锁 无界 LinkedList
ConcurrentLinkedQueue CAS 无界 LinkedList
LinkedTransferQueue CAS 无界 LinkedList

    我们知道CAS算法比通过加锁实现同步性能高很多,而上表可以看出基于CAS实现的队列都是无界的,而有界队列是通过同步实现的。在系统稳定性要求比较高的场景下,为了防止生产者速度过快,如果采用无界队列会最终导致内存溢出,只能选择有界队列。而有界队列只有ArrayBlockingQueue,该队列是通过加锁实现的,在请求锁和释放锁时对性能开销很大,这时候基于有界队列的高性能的Disruptor就应运而生。

Disruptor如何实现高性能?

Disruptor实现高性能主要体现了去掉了锁,采用CAS算法,同时内部通过环形队列实现有界队列。

  • 环形数据结构
    为了避免垃圾回收,采用数组而非链表。同时,数组对处理器的缓存机制更加友好。
  • 元素位置定位
    数组长度2^n,通过位运算,加快定位的速度。下标采取递增的形式。不用担心index溢出的问题。index是long类型,即使100万QPS的处理速度,也需要30万年才能用完。
  • 无锁设计
    每个生产者或者消费者线程,会先申请可以操作的元素在数组中的位置,申请到之后,直接在该位置写入或者读取数据。整个过程通过原子变量CAS,保证操作的线程安全。

Disruptor可以用来做什么?

    当前业界开源组件使用Disruptor的包括Log4j2、Apache Storm等,它可以用来作为高性能的有界内存队列,基于生产者消费者模式,实现一个/多个生产者对应多个消费者。它也可以认为是观察者模式的一种实现,或者发布订阅模式。

同时,Disruptor还允许开发者使用多线程技术去创建基于任务的工作流。Disruptor能用来并行创建任务,同时保证多个处理过程的有序性,并且它是没有锁的。

为什么要使用Disruptor?

    使用Disruptor,主要用于对性能要求高、延迟低的场景,它通过“榨干”机器的性能来换取处理的高性能。如果你的项目有对性能要求高,对延迟要求低的需求,并且需要一个无锁的有界队列,来实现生产者/消费者模式,那么Disruptor是你的不二选择。

怎么用Disruptor?

    要学会基于Disruptor进行编程,我们先了解下大概流程示意图,其中绿色部分是表示我们需要编写和实现的类。

 

import com.lmax.disruptor.BlockingWaitStrategy;import com.lmax.disruptor.EventFactory;import com.lmax.disruptor.EventHandler;import com.lmax.disruptor.EventTranslatorOneArg;import com.lmax.disruptor.ExceptionHandler;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.dsl.ProducerType;import java.util.concurrent.ThreadFactory;/** * @author: chuanyi.huang **/public class DisruptorTest {    /*     * 消息事件类     */    public static class MessageEvent {        /**         * 原始消息         */        private String message;        public String getMessage() {            return message;        }        public void setMessage(String message) {            this.message = message;        }    }    /*     * 消息事件工厂类     */    public static class MessageEventFactory implements EventFactory
{ public MessageEvent newInstance() { return new MessageEvent(); } } /* * 消息转换类, 负责将消息转换成事件 */ public static class MessageEventTranslator implements EventTranslatorOneArg
{ public void translateTo(MessageEvent messageEvent, long l, String s) { messageEvent.setMessage(s); } } /* * 消费者线程工厂类 */ public static class MessageThreadFactory implements ThreadFactory { public Thread newThread(Runnable r) { return new Thread(r, "Simple Discruptor Test Thread"); } } /* * 消息事件处理类 */ public static class MessageEventHandler implements EventHandler
{ public void onEvent(MessageEvent messageEvent, long l, boolean b) throws Exception { System.out.println(messageEvent.getMessage()); } } /* * 异常处理类 */ public static class MessageExceptionHandler implements ExceptionHandler
{ public void handleEventException(Throwable throwable, long l, MessageEvent messageEvent) { throwable.printStackTrace(); } public void handleOnStartException(Throwable throwable) { throwable.printStackTrace(); } public void handleOnShutdownException(Throwable throwable) { throwable.printStackTrace(); } } /* * 消息生产者类 */ public static class MessageEventProducer { private RingBuffer
ringBuffer; public MessageEventProducer(RingBuffer
ringBuffer) { this.ringBuffer = ringBuffer; } /** * 将接收到的消息输出到ringBuffer */ public void onData(String message) { EventTranslatorOneArg
translator = new MessageEventTranslator(); ringBuffer.publishEvent(translator, message); } } public static void main(String[] args) { String message = "Hello Disruptor"; int ringBufferSize = 1024; Disruptor
disruptor = new Disruptor
(new MessageEventFactory(), ringBufferSize, new MessageThreadFactory(), ProducerType.SINGLE, new BlockingWaitStrategy()); disruptor.handleEventsWith(new MessageEventHandler()); disruptor.setDefaultExceptionHandler(new MessageExceptionHandler()); RingBuffer
start = disruptor.start(); MessageEventProducer messageEventProducer = new MessageEventProducer(start); messageEventProducer.onData(message); }}

 

转载于:https://my.oschina.net/hcy8888/blog/2989557

你可能感兴趣的文章
libev源码分析---整体设计
查看>>
MyEclipse基本开发环境配置!
查看>>
TFBOYS饭票上线引热议,骗局之外,区块链技术能重构娱乐产业吗?
查看>>
云服务 好钢用在刀刃上
查看>>
聊下git merge --squash
查看>>
常见的几种分支开发方式
查看>>
一份书单
查看>>
【AS3代码】更换鼠标箭头样式,并跟随鼠标!
查看>>
Hybrid App 开发初探:使用 WebView 装载页面
查看>>
wcf异常处理
查看>>
asp.net实现OAuth认证服务端接口
查看>>
C#怎样将窗体或某个控件保存成图片(或给窗体截图)
查看>>
第三部分:Android 应用程序接口指南---第一节:应用程序组件---第一章1.Activity...
查看>>
动态规划练习 3
查看>>
经典排序之 插入排序
查看>>
POJ-3252 Round Numbers 按位DP
查看>>
Hadoop:The Definitive Guid 总结 Chapter 10 管理Hadoop
查看>>
Html.RenderPartial与 Html.RenderAction的区别
查看>>
ASP.NET那点不为人知的事(二)
查看>>
LINQ-to-SQL那点事~线程共享的DbContext与私有的DbContext
查看>>