Disruptor框架学习(1)--怎么实现

1 Disruptor学习

在上一篇文章中,笔者提到了log4j2中的异步logger。通过测试数据来看,在使用异步logger后,打印日志的时间明显缩短,系统响应时间得到了巨大的提升。

那么,disruptor究竟是什么,为什么它可以提升系统的性能?

1.1 Disruptor简介

Disruptor是一个开源框架,研发的初衷是为了解决高并发下列队锁的问题,最早由LMAX(一种新型零售金融交易平台)提出并使用,能够在无锁的情况下实现队列的并发操作,并号称能够在一个线程里每秒处理6百万笔订单(我是不相信)。

队列的特性:先进先出(FIFO)--先进入队列的元素先出队列(可以理解为我们生活中的排队情况,早办完,早滚蛋)。生产者(Producer)往队列里发布(publish)事件,消费者(Consumer)获得通知,消费事件;如果队列中没有事件时,消费者堵塞,直到生产者发布了新事件。

说到队列,那就不得不提到Java中的concurrent包,其主要实现包括ArrayBlockingQueue、LinkedBlockingQueue、ConcurrentLinkedQueue、LinkedTransferQueue。下面,简单介绍下:

ArrayBlockingQueue:基于数组形式的队列,通过加锁的方式,来保证多线程情况下数据的安全;

LinkedBlockingQueue:基于链表形式的队列,也通过加锁的方式,来保证多线程情况下数据的安全;

ConcurrentLinkedQueue:基于链表形式的队列,通过compare and swap(简称CAS)协议的方式,
来保证多线程情况下数据的安全,不加锁,主要使用了Java中的sun.misc.Unsafe类来实现;

LinkedTransferQueue:同上;

通过查看以上4个类的源码,可以发现:

(1)使用CAS协议实现队列的类,都是无界的,无法保证队列的长度,理论上来说可以是无限扩展,那么如果生产者生产过快,消费者还没来得及消费,最终可能会导致内存溢出,影响系统稳定;

(2)而使用加锁实现队列的类,虽然是有界的(可以设置队列的大小),但是有锁的存在,性能上有了很大的影响,线程由于锁的竞争被挂起,直到锁的释放,才能恢复。此外,由于伪共享的存在,也会影响性能

而Disruptor解决了以上的问题,实现了无锁有界队列操作。主要是使用了环形数组(ringbuffer)、CAS、缓存行填充、解决伪共享等技术,接下来我们一一讲解;

1.2 Disruptor结构

在讲解disruptor所使用的相关技术之前,我觉得有必要简单的介绍下的Disruptor结构!

前面介绍了,Disruptor是一个开源的框架,可以在无锁的情况下对队列进行操作,那么这个队列的设计就是Disruptor的核心所在;

在Disruptor中,采用了RingBuffer来作为队列的数据结构,RingBuffer就是一个环形的数组,既然是数组,我们便可对其设置大小。在这个ringBuffer中,除了数组之外,还有一个序列号,是用来指向数组中的下一个可用元素,供生产者使用或者消费者使用,也就是生产者可以生产的地方,或者消费者可以消费的地方。(序列号和数组索引是两个概念,别搞错了)

Disruptor使用数组作为队列的另一个好处,就是可以快速定位到所需元素,通常使用取摸运算(序列号%数组大小=所需元素角标),但在Disruptor中使用的是位运算(具体实现:UNSAFE.getObject(entries, REF_ARRAY_BASE + ((sequence & indexMask) << REF_ELEMENT_SHIFT))),效率更高,定位更快;此外,在Disruptor中数组内的元素并不会被删除,而是新数据来覆盖原有数据;

1.3 Disruptor代码简单实现

我们就以一个简单例子来实现Disruptor:生产者传递一个long类型变量给消费者,消费者将这个变量打印出来。

单生产者,单消费者模型:

(1)向ringbuffer中插入的事件元素:就是在对象中放了一个long变量

public class LongEvent {

    private long value;

    public long getValue() {
        return value;
    }

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

(2)事件生产工厂:生产事件存入ringbuffer中

public class LongEventFactory implements EventFactory<LongEvent> {

    public LongEvent newInstance() {
        return new LongEvent();
    }
}

(3)事件处理器,也就是消费者,就是将事件的值打印出来

public class LongEventHandler implements EventHandler<LongEvent> {

    public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {
        System.out.println("Event:"+event.getValue());
    }
}

(4)主函数:创建生产者,向ringbuffer中填充元素

public class DisruptorMain {

    public static void main(String[] agrs) throws InterruptedException {
        
        //创建线程池:
        Executor executor = Executors.newCachedThreadPool();

        //事件生产工厂:
        LongEventFactory longEventFactory = new LongEventFactory();

        //ringbuffer的大小:
        int bufferSize = 256;

        //实例化disruptor对象:初始化ringbuffer
         Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(longEventFactory, bufferSize, executor,ProducerType.SINGLE, new BlockingWaitStrategy());

        //设置事件的执行者:(单消费者)
        disruptor.handleEventsWith(new LongEventHandler());
        
        //disruptor启动:
        disruptor.start();

        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

        //设置事件单生产者:
        for(int x = 0;x<256; x++){
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next();  
            try{
                // 返回可用位置的元素
                LongEvent event = ringBuffer.get(sequence); 
                // 设置该位置元素的值
                event.set(x); 
            }finally{
                //发布事件 
                ringBuffer.publish(sequence);
            }
            Thread.sleep(10);
        }
    }
}

1.4 Disruptor主要实现类

通过以上代码,我们来简单的分析下Disruptor的构成:

Disruptor:Disruptor的入口,主要封装了环形队列RingBuffer、消费者集合ConsumerRepository的引用;主要提供了获取环形队列、添加消费者、生产者向RingBuffer中添加事件(可以理解为生产者生产数据)的操作;

RingBuffer:Disruptor中队列具体的实现,底层封装了Object[]数组;在初始化时,会使用Event事件对数组进行填充,填充的大小就是bufferSize设置的值;此外,该对象内部还维护了Sequencer(序列生产器)具体的实现;

Sequencer:序列生产器,分别有MultiProducerSequencer(多生产者序列生产器) 和 SingleProducerSequencer(单生产者序列生产器)两个实现类。上面的例子中,使用的是SingleProducerSequencer;在Sequencer中,维护了消费者的Sequence(序列对象)和生产者自己的Sequence(序列对象);以及维护了生产者与消费者序列冲突时候的等待策略WaitStrategy;

Sequence:序列对象,内部维护了一个long型的value,这个序列指向了RingBuffer中Object[]数组具体的角标。生产者和消费者各自维护自己的Sequence;但都是指向RingBuffer的Object[]数组;

Wait Strategy:等待策略。当没有可消费的事件时,消费者根据特定的策略进行等待;当没有可生产的地方时,生产者根据特定的策略进行等待;

Event:事件对象,就是我们Ringbuffer中存在的数据,在Disruptor中用Event来定义数据,并不存在Event类,它只是一个定义;

EventProcessor:事件处理器,单独在一个线程内执行,判断消费者的序列和生产者序列关系,决定是否调用我们自定义的事件处理器,也就是是否可以进行消费;

EventHandler:事件处理器,由用户自定义实现,也就是最终的事件消费者,需要实现EventHandler接口;

Producer:事件生产者,也就是我们上面代码中最后那部门的for循环;

1.5 Disruptor的生产和消费

上面我们通过代码简单的实现了Disruptor,阐述其中具体实现类的含义,接下来再用图文的方式进一步介绍Disruptor的生产和消费;

暂时还是以单生产和单消费者举例:

(1)当Disruptor框架启动:


(2)此时,还没有数据进行写入


(3)准备写入数据前的准备,获取可以写入数据的最大序列;


(4)写入数据完成,更新生产者序列对象的值;


以上,就是单生产者写入数据的过程。要注意的是,无论是生产者还是消费者,序列的初始值都是-1;

当引入消费者后,生产者在获取可写入的序列之前,都会判断消费者所处的序列。

我们假设一种情况,当在我们的消费者端使用Thread.sleep(巨大的值)的时候,消费者使用被等待,无法进行消费。

那么此时,生产者会一直对数组中的元素进行生产,当生产到7准备生产序列8的时候,通过计算序列8对应的是index = 0的元素,我们此时会判断覆盖点所对应的角标是否大于消费者的序列大小,如果大于消费者序列,那么生产者不会进行生产,直到消费者消费了此角标下的元素;

public long next(int n){
    if (n < 1)
    {
        throw new IllegalArgumentException("n must be > 0");
    }

    long nextValue = this.nextValue;

    long nextSequence = nextValue + n;
    long wrapPoint = nextSequence - bufferSize;
    long cachedGatingSequence = this.cachedValue;

    if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue){
        cursor.setVolatile(nextValue);  // StoreLoad fence

        long minSequence;
        //此处进行判断,如果覆盖点的大小,超过了消费者的序列,那么会一直while循环进行判断
        while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue))){
            waitStrategy.signalAllWhenBlocking();
            LockSupport.parkNanos(1L); // TODO: Use waitStrategy to spin?
        }

        this.cachedValue = minSequence;
    }

    this.nextValue = nextSequence;

    return nextSequence;
}

单消费者,进行消费的逻辑,与单生产者类似,大家可以进行深入研究;

以上便是单消费者和单生产者的大体流程;

下一篇,笔者将着重要介绍,Disruptor中使用的技术方案!!!

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 206,968评论 6 482
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 88,601评论 2 382
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 153,220评论 0 344
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 55,416评论 1 279
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 64,425评论 5 374
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,144评论 1 285
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,432评论 3 401
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,088评论 0 261
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,586评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,028评论 2 325
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,137评论 1 334
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,783评论 4 324
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,343评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,333评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,559评论 1 262
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,595评论 2 355
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,901评论 2 345

推荐阅读更多精彩内容