1.Disruptor的背景

  disruptor是LAMX架构的一种设计，而LAMX是一种新型的零售金融交易平台。disruptor主要用于大规模低延迟的高并发业务场景，其核心disruptor是一个基于事件源驱动机制的业务逻辑处理器，整个业务逻辑处理器完全运行在内存中，disruptor在无锁的网络情况下，实现了Queue的并发。

2.Disruptor的适用场景

  disruptor适用于大规模低延迟的并发场景。可用于读写操作分离、数据缓存，速度匹配（因为其实现了生产者-消费者模型）、或者是基于内存的事件流处理机制的场景。

曾有人测过，一个线程里1s内可以处理六百万个订单，性能相当感人。

这个框架的结构大概是：数据生产端 --> 缓存 --> 消费端

缓存中的数据是主动发给消费端的，而不是像一般的生产者消费者模式那样，消费端去缓存中取数据。

可以将disruptor理解为，基于事件驱动的高效队列、轻量级的JMS

disruptor学习网站：http://ifeve.com/disruptor-getting-started

3.Disruptor的设计思想与原理

  disruptor的主要设计思想是无锁的高并发，在设计上采用内存屏障的机制和CAS操作实现此思想。主流的并发程序
都离不开锁对资源的管控，或者尽量避开锁的使用。
  其主要的实现原理总结有如下三点，当然还有很多地方设计得很巧妙，需要细细阅读源码和官方文档。虽然这个 过程对我来说很尴尬，但痛并快乐者，有朝闻道、夕可死也的感觉。
　　1.采用消费者-生产者模型进行读写的分离（Actors模式）。
　　2.用循环缓存（实际是一个循环队列）实现了数据的暂存和读写速度的匹配。为了避免垃圾回收，采用数组而非链表。同时，数组对处理器的缓存机制更加友好。数组长度2^n，通过位运算，加快定位的速度。下标采取递增的形式。不用担心index溢出的问题。index是long类型，即使100万QPS的处理速度，也需要30万年才能用完。
　　3.用内存屏障加***的方式实现了无锁的并发机制。每个生产者或者消费者线程，会先申请可以操作的元素在数组中的位置，申请到之后，直接在该位置写入或者读取数据。

性能

disruptor是用于一个JVM中多个线程之间的消息队列，作用与ArrayBlockingQueue有相似之处，但是disruptor从功能、性能都远好于ArrayBlockingQueue，当多个线程之间传递大量数据或对性能要求较高时，可以考虑使用disruptor作为ArrayBlockingQueue的替代者。

性能测评

官方也对disruptor和ArrayBlockingQueue的性能在不同的应用场景下做了对比，本文列出其中一组数据，数据中P代表producer，C代表consumer，ABS代表ArrayBlockingQueue

整体结构图

核心类图

• RingBuffer——Disruptor底层数据结构实现，核心类，是线程间交换数据的中转地；
• Sequencer——序号管理器，负责消费者/生产者各自序号、序号栅栏的管理和协调；
• Sequence——序号，声明一个序号，用于跟踪ringbuffer中任务的变化和消费者的消费情况；
• SequenceBarrier——序号栅栏，管理和协调生产者的游标序号和各个消费者的序号，确保生产者不会覆盖消费者未来得及处理的消息，确保存在依赖的消费者之间能够按照正确的顺序处理；
• EventProcessor——事件处理器，监听RingBuffer的事件，并消费可用事件，从RingBuffer读取的事件会交由实际的生产者实现类来消费；它会一直侦听下一个可用的序号，直到该序号对应的事件已经准备好。
• EventHandler——业务处理器，是实际消费者的接口，完成具体的业务逻辑实现，第三方实现该接口；代表着消费者。
• Producer——生产者接口，第三方线程充当该角色，producer向RingBuffer写入事件。

DSL

Disruptor3.0（domain specific language 特定领域语言）的类图，可以大致知道第三方如何继承Disruptor3.0实现具体业务逻辑。

• Disruptor——对外暴露的门面类，提供start()，stop()，消费者事件注册，生产者事件发布等api；
• RingBuffer——对生产者提供下一序号获取、entry元素获取、entry数据更改等api；
• EventHandler——消费者的接口定义，提供onEvent()方法，负责具体业务逻辑实现；
• EventHandlerGroup——业务处理器分组，管理多个业务处理器的依赖关系，提供then()、before()、after()等api。

关键时序图

整个运行过程的时序图，包括：始化、启动、处理过程。

内存预分配

RingBuffer使用数组Object[] entries作为存储元素，如下图所示，初始化RingBuffer时，会将所有的entries的每个元素指定为特定的Event，这时候event中的detail属性是null；后面生产者向RingBuffer中写入消息时，RingBuffer不是直接将enties[7]指向其他的event对象，而是先获取event对象，然后更改event对象的detail属性；消费者在消费时，也是从RingBuffer中读取出event，然后取出其detail属性。可以看出，生产/消费过程中，RingBuffer的entities[7]元素并未发生任何变化，未产生临时对象，entities及其元素对象一直存活，知道RingBuffer消亡。故而可以最小化GC的频率，提升性能。

注：图中对象Entry写错，应当为Event。

消除伪共享

如果两个不同的并发变量位于同一个缓存行，则在并发情况下，会互相影响到彼此的缓存有效性，进而影响到性能，这叫着‘伪共享’。为了避开‘伪共享’，Disruptor3.0在Sequence.java中使用多个long变量填充，从而确保一个序号独占一个缓存行。关于缓存行和‘伪共享’请参考：伪共享(False Sharing)。

代码如下：

//在序号实际value变量（long型）左边填充7个long变量
class LhsPadding
{
    protected long p1, p2, p3, p4, p5, p6, p7;
}

class Value extends LhsPadding
{
    protected volatile long value;
}

//在序号实际value变量（long型）右边填充7个long变量
class RhsPadding extends Value { 
　　protected long p9, p10, p11, p12, p13, p14, p15; 
} 

public class Sequence extends RhsPadding { 
　　static final long INITIAL_VALUE = -1L; 
　　public Sequence() {
 　　　　this(INITIAL_VALUE); 
　　}
 　　...... 
}

Sequence实际value变量的左右均被填充了7个long型变量，其自身也是long型变量，一个long型变量占据8个字节，所以序号与他上一个/下一个序号之间的最小内存分布距离为：7*8=56byte，加上自身的8个byte，可以确保序号变量独占长度为64byte（通常的一个缓存行长度）缓存行。

序号栅栏和序号配合使用来消除锁和CAS

Disruptor3.0中，序号栅栏（SequenceBarrier）和序号（Sequence）搭配使用，协调和管理消费者与生产者的工作节奏，避免了锁和CAS的使用。在Disruptor3.0中，各个消费者和生产者持有自己的序号，这些序号的变化必须满足如下基本条件：

• 消费者序号数值必须小于生产者序号数值；
• 消费者序号数值必须小于其前置（依赖关系）消费者的序号数值；
• 生产者序号数值不能大于消费者中最小的序号数值，以避免生产者速度过快，将还未来得及消费的消息覆盖。

上述前两点是在SequenceBarrier的waitFor()方法中完成的，源码如下：

public long waitFor(final long sequence) //sequence参数是该消费者期望获取的下一个序号值
        throws AlertException, InterruptedException, TimeoutException
    {
        checkAlert();
　　　　 //根据配置的waitStrategy策略，等待期望的下一序号值变得可用
　　　　 //这里并不保证返回值availableSequence一定等于 given sequence，他们的大小关系取决于采用的WaitStrategy。
　　　　 //eg. 1、YieldingWaitStrategy在自旋100次尝试后，会直接返回dependentSequence的最小seq，这时并不保证返回值>=given sequence
　　　　 //    2、BlockingWaitStrategy则会阻塞等待given sequence可用为止，可用并不是说availableSequence == given sequence，而应当是指 >=

long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);

        //如果当前可用的序号小于期望获取的下一个序号，则返回availableSequence，这将导致调用者EventProcessor继续wait
        if (availableSequence < sequence)
        {
            return availableSequence;
        }
　　　　 //这一句是‘批处理’的精妙所在，放在后面讲
        return sequencer.getHighestPublishedSequence(sequence, availableSequence);
    }

上面第三点是针对生产者建立的Barrier，逻辑判定发生在生产者从ringBuffer获取下一个可用的entry时，RingBuffer会将获取下一个可用的entry委托给Sequencer。我们以最简单的单生产者SingleProducerSequencer的next()实现来说明。SingleProducerSequencer.next()的源码如下：

public long next(int n) 
    {
        if (n < 1) //n表示此次生产者期望获取多少个序号，通常是1
        {
            throw new IllegalArgumentException("n must be > 0");
        }

        long nextValue = this.nextValue;

        long nextSequence = nextValue + n;  //生产者当前序号值+期望获取的序号数量后达到的序号值
        long wrapPoint = nextSequence - bufferSize; //减掉RingBuffer的总的buffer值，用于判断是否出现‘覆盖’
        long cachedGatingSequence = this.cachedValue;  //从后面代码分析可得：cachedValue就是缓存的消费者中最小序号值，他不是当前最新的‘消费者中最小序号值’，而是上次程序进入到下面的if判定代码段是，被赋值的当时的‘消费者中最小序号值’
                //这样做的好处在于：在判定是否出现覆盖的时候，不用每次都调用getMininumSequence计算‘消费者中的最小序号值’，从而节约开销。只要确保当生产者的节奏大于了缓存的cachedGateingSequence一个bufferSize时，从新获取一下 getMinimumSequence()即可。
//(wrapPoint > cachedGatingSequence) ： 当生产者已经超过上一次缓存的‘消费者中最小序号值’（cachedGatingSequence）一个‘Ring’大小（bufferSize），需要重新获取cachedGatingSequence，避免当生产者一直在生产，但是消费者不再消费的情况下，出现‘覆盖’
        //(cachedGatingSequence > nextValue) ： 生产者和消费者均为顺序递增的，且生产者的seq“先于”消费者的seq，注意是‘先于’而不是‘大于’。当nextValue>Long.MAXVALUE时，nextValue+1就会变成负数，wrapPoint也会变成负数，这时候必然会是：cachedGatingSequence > nextValue
        //                                     这个变化的过程会持续bufferSize个序号，这个区间，由于getMinimumSequence()得到的虽然是名义上的‘消费者中最小序号值’，但是不代表是走在‘最后面’的消费者
        if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue) 
        {
            cursor.setVolatile(nextValue);  // StoreLoad fence

            long minSequence;
            //生产者停下来，等待消费者消费，知道‘覆盖’现象清除。
            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;
    }

批处理效应

当生产者节奏快于消费者，消费者可以通过‘批处理效应’快速追赶，即：消费者可以一次性从RingBuffer中获取多个已经准备好的enties，从而提高效率。代码实现如下：

SequenceBarrier的waitFor()方法：

public long waitFor(final long sequence)
        throws AlertException, InterruptedException, TimeoutException
    {
        checkAlert();

        long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);

        if (availableSequence < sequence)
        {
            return availableSequence;
        }

        //获取消费者可以消费的最大的可用序号，支持批处理效应，提升处理效率。
        //当availableSequence > sequence时，需要遍历 sequence --> availableSequence，找到最前一个准备就绪，可以被消费的event对应的seq。
        //最小值为：sequence-1
        return sequencer.getHighestPublishedSequence(sequence, availableSequence);
    }

 

4.开发流程

1.建Event类（数据对象）

2.建立一个生产数据的工厂类，EventFactory，用于生产数据；

3.监听事件类（处理Event数据）

4.实例化Disruptor，配置参数，绑定事件；

5.建存放数据的核心 RingBuffer，生产的数据放入 RungBuffer。

5 实例分析

模拟一个停车场场景

public class MyInParkingDataEvent {
    // 车牌号
    private String carLicense;
     public String getCarLicense() {
         return carLicense;    }
     public void setCarLicense(String carLicense) {
         this.carLicense = carLicense; }
}

/**
 * Created by cby on 2018/9/29.
 * 生产者，进入停车场的车辆
 */
public class MyInParkingDataEventPublisher implements Runnable{
    private CountDownLatch countDownLatch; // 用于监听初始化操作，等初始化执行完毕后，通知主线程继续工作
    private Disruptor<MyInParkingDataEvent> disruptor;
    private static final Integer NUM = 10; // 1,10,100,1000
    @Override
    public void run() {
        MyInParkingDataEventTranslator eventTranslator = new MyInParkingDataEventTranslator();
        try {
            RingBuffer rb= disruptor.getRingBuffer();
            for(int i = 0; i < NUM; i ++) {
                rb.publishEvent(eventTranslator);
                Thread.sleep(1000); // 假设一秒钟进一辆车

            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        finally {
            countDownLatch.countDown(); // 执行完毕后通知 await()方法
            System.out.println(NUM + "辆车已经全部进入进入停车场！");

       }
    }


    public MyInParkingDataEventPublisher(CountDownLatch countDownLatch,
                                         Disruptor<MyInParkingDataEvent> disruptor) {
        this.countDownLatch = countDownLatch;
        this.disruptor = disruptor;
    }

}
class MyInParkingDataEventTranslator implements EventTranslator<MyInParkingDataEvent> {
    @Override
    public void translateTo(MyInParkingDataEvent myInParkingDataEvent, long sequence) {
        this.generateData(myInParkingDataEvent);
    }
    private MyInParkingDataEvent generateData(MyInParkingDataEvent myInParkingDataEvent) {
        myInParkingDataEvent.setCarLicense("车牌号： 鄂A-" + (int)(Math.random() * 100000)); // 随机生成一个车牌号
        System.out.println("Thread Id " + Thread.currentThread().getId() + " 写完一个event");
        return myInParkingDataEvent;
     }

}

/**
 * Created by cby on 2018/9/29.
 * Handler 第一个消费者，负责保存进场汽车的信息
 */
public class MyParkingDataInDbHandler implements EventHandler<MyInParkingDataEvent>, WorkHandler<MyInParkingDataEvent> {
    @Override
    public void onEvent(MyInParkingDataEvent myInParkingDataEvent, long l, boolean b) throws Exception {
        // 获取当前线程id
        long threadId = Thread.currentThread().getId();
        // 获取车牌号
        String carLicense = myInParkingDataEvent.getCarLicense();
        System.out.println(String.format("Thread Id %s 保存 %s 到数据库中 ....", threadId, carLicense));

    }

    @Override
    public void onEvent(MyInParkingDataEvent myInParkingDataEvent) throws Exception {
        this.onEvent(myInParkingDataEvent);
    }
}

/**
 * Created by cby on 2018/9/29.
 * 第二个消费者，负责发送通知告知工作人员(Kafka是一种高吞吐量的分布式发布订阅消息系统)
 */
public class MyParkingDataToKafkaHandler implements EventHandler<MyInParkingDataEvent> {
    @Override
    public void onEvent(MyInParkingDataEvent myInParkingDataEvent, long l, boolean b) throws Exception {

        long threadId = Thread.currentThread().getId(); // 获取当前线程id
        String carLicense = myInParkingDataEvent.getCarLicense(); // 获取车牌号
        System.out.println(String.format("Thread Id %s 发送 %s 进入停车场信息给 kafka系统...", threadId, carLicense));
    }
}

/**
 * Created by cby on 2018/9/29.
 * 第三个消费者，sms短信服务，告知司机你已经进入停车场，计费开始。
 */
public class MyParkingDataSmsHandler implements EventHandler<MyInParkingDataEvent> {
    @Override
    public void onEvent(MyInParkingDataEvent myInParkingDataEvent, long l, boolean b) throws Exception {

        long threadId = Thread.currentThread().getId(); // 获取当前线程id
        String carLicense = myInParkingDataEvent.getCarLicense(); // 获取车牌号
        System.out.println(String.format("Thread Id %s 给  %s 的车主发送一条短信，并告知他计费开始了 ....", threadId, carLicense));
    }
}

6 分析伪共享

什么是共享

下图是计算的基本结构。L1、L2、L3分别表示一级缓存、二级缓存、三级缓存，越靠近CPU的缓存，速度越快，容量也越小。所以L1缓存很小但很快，并且紧靠着在使用它的CPU内核；L2大一些，也慢一些，并且仍然只能被一个单独的CPU核使用；L3更大、更慢，并且被单个插槽上的所有CPU核共享；最后是主存，由全部插槽上的所有CPU核共享。

图 计算机CPU与缓存示意图

当CPU执行运算的时候，它先去L1查找所需的数据、再去L2、然后是L3，如果最后这些缓存中都没有，所需的数据就要去主内存拿。走得越远，运算耗费的时间就越长。所以如果你在做一些很频繁的事，你要尽量确保数据在L1缓存中。

另外，线程之间共享一份数据的时候，需要一个线程把数据写回主存，而另一个线程访问主存中相应的数据。

下面是从CPU访问不同层级数据的时间概念:

从CPU到	大约需要的CPU周期	大约需要的时间
主存		约60-80ns
QPI 总线传输(between sockets, not drawn)		约20ns
L3 cache	约40-45 cycles	约15ns
L2 cache	约10 cycles	约3ns
L1 cache	约3-4 cycles	约1ns
寄存器	1 cycle

可见CPU读取主存中的数据会比从L1中读取慢了近2个数量级。

缓存行

Cache是由很多个cache line组成的。每个cache line通常是64字节，并且它有效地引用主内存中的一块儿地址。一个Java的long类型变量是8字节，因此在一个缓存行中可以存8个long类型的变量。

CPU每次从主存中拉取数据时，会把相邻的数据也存入同一个cache line。

在访问一个long数组的时候，如果数组中的一个值被加载到缓存中，它会自动加载另外7个。因此你能非常快的遍历这个数组。事实上，你可以非常快速的遍历在连续内存块中分配的任意数据结构。

下面的例子是测试利用cache line的特性和不利用cache line的特性的效果对比

public class CacheLineEffect {
    //考虑一般缓存行大小是64字节，一个 long 类型占8字节
    static  long[][] arr;

    public static void main(String[] args) {
        arr = new long[1024 * 1024][];
        for (int i = 0; i < 1024 * 1024; i++) {
            arr[i] = new long[8];
            for (int j = 0; j < 8; j++) {
                arr[i][j] = 0L;
            }
        }
        long sum = 0L;
        long marked = System.currentTimeMillis();
        for (int i = 0; i < 1024 * 1024; i+=1) {
            for(int j =0; j< 8;j++){
                sum = arr[i][j];
            }
        }
        System.out.println("Loop times:" + (System.currentTimeMillis() - marked) + "ms");

        marked = System.currentTimeMillis();
        for (int i = 0; i < 8; i+=1) {
            for(int j =0; j< 1024 * 1024;j++){
                sum = arr[j][i];
            }
        }
        System.out.println("Loop times:" + (System.currentTimeMillis() - marked) + "ms");
    }
}

在2G Hz、2核、8G内存的运行环境中测试，速度差一倍。

结果：
Loop times:30ms
Loop times:65ms

在2.8 Hz、8核、8G内存的运行环境中测试，速度差三倍多。

Loop times:18ms
Loop times:60ms

什么是伪共享

ArrayBlockingQueue有三个成员变量：

• takeIndex：需要被取走的元素下标
• putIndex：可被元素插入的位置的下标
• count：队列中元素的数量

这三个变量很容易放到一个缓存行中，但是之间修改没有太多的关联。所以每次修改，都会使之前缓存的数据失效，从而不能完全达到共享的效果。

图 ArrayBlockingQueue伪共享示意图

如上图所示，当生产者线程put一个元素到ArrayBlockingQueue时，putIndex会修改，从而导致消费者线程的缓存中的缓存行无效，需要从主存中重新读取。

这种无法充分使用缓存行特性的现象，称为伪共享。

对于伪共享，一般的解决方案是，增大数组元素的间隔使得由不同线程存取的元素位于不同的缓存行上，以空间换时间。

/**
 * Created by cby on 2018/9/30.
 */
public class FalseSharing implements Runnable{
    public final static long ITERATIONS = 500L * 1000L * 100L;
    private int arrayIndex = 0;

    private static ValuePadding[] longs;
    public FalseSharing(final int arrayIndex) {
        this.arrayIndex = arrayIndex;
    }

    public static void main(final String[] args) throws Exception {
        for(int i=1;i<10;i++){
            System.gc();
            final long start = System.currentTimeMillis();
            runTest(i);
            System.out.println("Thread num "+i+" duration = " + (System.currentTimeMillis() - start));
        }

    }

    private static void runTest(int NUM_THREADS) throws InterruptedException {
        Thread[] threads = new Thread[NUM_THREADS];
        longs = new ValuePadding[NUM_THREADS];
        for (int i = 0; i < longs.length; i++) {
            longs[i] = new ValuePadding();
        }
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(new FalseSharing(i));
        }

        for (Thread t : threads) {
            t.start();
        }

        for (Thread t : threads) {
            t.join();
        }
    }

    public void run() {
        long i = ITERATIONS + 1;
        while (0 != --i) {
            longs[arrayIndex].value = 0L;
        }
    }

    public final static class ValuePadding {
        protected long p1, p2, p3, p4, p5, p6, p7;
        protected volatile long value = 0L;
        protected long p9, p10, p11, p12, p13, p14;
        protected long p15;
    }
    public final static class ValueNoPadding {
        // protected long p1, p2, p3, p4, p5, p6, p7;
        protected volatile long value = 0L;
        // protected long p9, p10, p11, p12, p13, p14, p15;
    }
}

在2G Hz，2核，8G内存, jdk 1.7.0_45 的运行环境下，使用了共享机制比没有使用共享机制，速度快了4倍左右。

结果：
Thread num 1 duration = 447
Thread num 2 duration = 463
Thread num 3 duration = 454
Thread num 4 duration = 464
Thread num 5 duration = 561
Thread num 6 duration = 606
Thread num 7 duration = 684
Thread num 8 duration = 870
Thread num 9 duration = 823

把代码中ValuePadding都替换为ValueNoPadding后的结果：
Thread num 1 duration = 446
Thread num 2 duration = 2549
Thread num 3 duration = 2898
Thread num 4 duration = 3931
Thread num 5 duration = 4716
Thread num 6 duration = 5424
Thread num 7 duration = 4868
Thread num 8 duration = 4595
Thread num 9 duration = 4540

备注：在jdk1.8中，有专门的注解@Contended来避免伪共享，更优雅地解决问题。

 

总结

假如某一天 jdk 并发包中队列被disruptor 替换性能应该有很大提升

当然disruptor 的actors 设计模式是一种很好的选择

参考文献

1 https://www.cnblogs.com/sigm/p/6251910.html

2 disruptor github wiki：
Home · LMAX-Exchange/disruptor Wiki

3 disruptor github：
LMAX-Exchange/disruptor: High Performance Inter-Thread Messaging Library

4 这个地方也有很多不错的资料：Disruptor by LMAX-Exchange

5 http://brokendreams.iteye.com/blog/2255720

6 http://ifeve.com/dissecting-disruptor-whats-so-special/

7 https://github.com/LMAX-Exchange/disruptor/wiki/Performance-Results

8 https://lmax-exchange.github.io/disruptor/

9 https://logging.apache.org/log4j/2.x/manual/async.html