高性能高并发队列-Disruptor

Disruptor是英国外汇交易公司LMAX开发的一个高性能队列，研发的初衷是解决内存队列的延迟问题（在性能测试中发现竟然与I/O操作处于同样的数量级）。基于Disruptor开发的系统单线程能支撑每秒600万订单，2010年在QCon演讲后，获得了业界关注。2011年，企业应用软件专家Martin Fowler专门撰写长文介绍。同年它还获得了Oracle官方的Duke大奖。

Java内置队列

介绍Disruptor之前，我们先来看一看常用的线程安全的内置队列有什么问题。Java的内置队列如下表所示。

队列的底层一般分成三种：数组、链表和堆。其中，堆一般情况下是为了实现带有优先级特性的队列，暂且不考虑。

我们就从数组和链表两种数据结构来看，基于数组线程安全的队列，比较典型的是ArrayBlockingQueue，它主要通过加锁的方式来保证线程安全；基于链表的线程安全队列分成LinkedBlockingQueue和ConcurrentLinkedQueue两大类，前者也通过锁的方式来实现线程安全，而后者以及上面表格中的LinkedTransferQueue都是通过原子变量compare and swap（以下简称“CAS”）这种不加锁的方式来实现的。

通过不加锁的方式实现的队列都是无界的（无法保证队列的长度在确定的范围内）；而加锁的方式，可以实现有界队列。在稳定性要求特别高的系统中，为了防止生产者速度过快，导致内存溢出，只能选择有界队列；同时，为了减少Java的垃圾回收对系统性能的影响，会尽量选择array/heap格式的数据结构。这样筛选下来，符合条件的队列就只有ArrayBlockingQueue。

Disruptor论文中讲述了一个实验：

• 这个测试程序调用了一个函数，该函数会对一个64位的计数器循环自增5亿次。
• 机器环境：2.4G 6核
• 运算： 64位的计数器累加5亿次

CAS操作比单线程无锁慢了1个数量级；有锁且多线程并发的情况下，速度比单线程无锁慢3个数量级。可见无锁速度最快。

单线程情况下，不加锁的性能 > CAS操作的性能 > 加锁的性能。

在多线程情况下，为了保证线程安全，必须使用CAS或锁，这种情况下，CAS的性能超过锁的性能，前者大约是后者的8倍。

综上可知，加锁的性能是最差的。

 

Disruptor的设计方案

Disruptor通过以下设计来解决队列速度慢的问题：

• 环形数组结构

为了避免垃圾回收，采用数组而非链表。同时，数组对处理器的缓存机制更加友好。

• 元素位置定位

数组长度2^n，通过位运算，加快定位的速度。下标采取递增的形式。不用担心index溢出的问题。index是long类型，即使100万QPS的处理速度，也需要30万年才能用完。

• 无锁设计

每个生产者或者消费者线程，会先申请可以操作的元素在数组中的位置，申请到之后，直接在该位置写入或者读取数据。

一个生产者

写数据

生产者单线程写数据的流程比较简单：

1. 申请写入m个元素；
2. 若是有m个元素可以写入，则返回最大的序列号。这儿主要判断是否会覆盖未读的元素；
3. 若是返回的正确，则生产者开始写入元素。

图5 单个生产者生产过程示意图

多个生产者

多个生产者的情况下，会遇到“如何防止多个线程重复写同一个元素”的问题。Disruptor的解决方法是，每个线程获取不同的一段数组空间进行操作。这个通过CAS很容易达到。只需要在分配元素的时候，通过CAS判断一下这段空间是否已经分配出去即可。

但是会遇到一个新问题：如何防止读取的时候，读到还未写的元素。Disruptor在多个生产者的情况下，引入了一个与Ring Buffer大小相同的buffer：available Buffer。当某个位置写入成功的时候，便把availble Buffer相应的位置置位，标记为写入成功。读取的时候，会遍历available Buffer，来判断元素是否已经就绪。

下面分读数据和写数据两种情况介绍。

读数据

生产者多线程写入的情况会复杂很多：

1. 申请读取到序号n；
2. 若writer cursor >= n，这时仍然无法确定连续可读的最大下标。从reader cursor开始读取available Buffer，一直查到第一个不可用的元素，然后返回最大连续可读元素的位置；
3. 消费者读取元素。

如下图所示，读线程读到下标为2的元素，三个线程Writer1/Writer2/Writer3正在向RingBuffer相应位置写数据，写线程被分配到的最大元素下标是11。

读线程申请读取到下标从3到11的元素，判断writer cursor>=11。然后开始读取availableBuffer，从3开始，往后读取，发现下标为7的元素没有生产成功，于是WaitFor(11)返回6。

然后，消费者读取下标从3到6共计4个元素。

图6 多个生产者情况下，消费者消费过程示意图

写数据

多个生产者写入的时候：

1. 申请写入m个元素；
2. 若是有m个元素可以写入，则返回最大的序列号。每个生产者会被分配一段独享的空间；
3. 生产者写入元素，写入元素的同时设置available Buffer里面相应的位置，以标记自己哪些位置是已经写入成功的。

如下图所示，Writer1和Writer2两个线程写入数组，都申请可写的数组空间。Writer1被分配了下标3到下表5的空间，Writer2被分配了下标6到下标9的空间。

Writer1写入下标3位置的元素，同时把available Buffer相应位置置位，标记已经写入成功，往后移一位，开始写下标4位置的元素。Writer2同样的方式。最终都写入完成。

图7 多个生产者情况下，生产者生产过程示意图

下面忽略数组的环形结构，介绍一下如何实现无锁设计。整个过程通过原子变量CAS，保证操作的线程安全。

防止不同生产者对同一段空间写入的代码，如下所示：

public long tryNext(int n) throws InsufficientCapacityException
{
    if (n < 1)
    {
        throw new IllegalArgumentException("n must be > 0");
    }
 
    long current;
    long next;
 
    do
    {
        current = cursor.get();
        next = current + n;
 
        if (!hasAvailableCapacity(gatingSequences, n, current))
        {
            throw InsufficientCapacityException.INSTANCE;
        }
    }
    while (!cursor.compareAndSet(current, next));
 
    return next;
}

通过do/while循环的条件cursor.compareAndSet(current, next)，来判断每次申请的空间是否已经被其他生产者占据。假如已经被占据，该函数会返回失败，While循环重新执行，申请写入空间。

消费者的流程与生产者非常类似，这儿就不多描述了。

总结

Disruptor通过精巧的无锁设计实现了在高并发情形下的高性能。

在美团点评内部，很多高并发场景借鉴了Disruptor的设计，减少竞争的强度。其设计思想可以扩展到分布式场景，通过无锁设计，来提升服务性能。

 

代码样例

/**
 * @description disruptor代码样例。每10ms向disruptor中插入一个元素，消费者读取数据，并打印到终端
 */
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import java.util.concurrent.ThreadFactory;
 
public class DisruptorMain
{
    public static void main(String[] args) throws Exception
    {
        // 队列中的元素
        class Element {
 
            private int value;
 
            public int get(){
                return value;
            }
 
            public void set(int value){
                this.value= value;
            }
 
        }
 
        // 生产者的线程工厂
        ThreadFactory threadFactory = new ThreadFactory(){
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "simpleThread");
            }
        };
 
        // RingBuffer生产工厂,初始化RingBuffer的时候使用
        EventFactory<Element> factory = new EventFactory<Element>() {
            @Override
            public Element newInstance() {
                return new Element();
            }
        };
 
        // 处理Event的handler
        EventHandler<Element> handler = new EventHandler<Element>(){
            @Override
            public void onEvent(Element element, long sequence, boolean endOfBatch)
            {
                System.out.println("Element: " + element.get());
            }
        };
 
        // 阻塞策略
        BlockingWaitStrategy strategy = new BlockingWaitStrategy();
 
        // 指定RingBuffer的大小
        int bufferSize = 16;
 
        // 创建disruptor，采用单生产者模式
        Disruptor<Element> disruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);
 
        // 设置EventHandler
        disruptor.handleEventsWith(handler);
 
        // 启动disruptor的线程
        disruptor.start();
 
        RingBuffer<Element> ringBuffer = disruptor.getRingBuffer();
 
        for (int l = 0; true; l++)
        {
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next();  
            try
            {
                // 返回可用位置的元素
                Element event = ringBuffer.get(sequence); 
                // 设置该位置元素的值
                event.set(l); 
            }
            finally
            {
                ringBuffer.publish(sequence);
            }
            Thread.sleep(10);
        }
    }
}

  切记：一定要在设置值的地方加上

  try{

  }finally{

  },

      否则如果数据发布不成功，最后数据会逐渐填满ringbuffer,最后后面来的数据根本没有办法调用可用空间，导致方法阻塞，占用CPU和内存，无法释放资源，最后导致服务器死机

      注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。

Disruptor 还提供另外一种形式的调用来简化以上操作，并确保 publish 总是得到调用。

public class LongEvent
{
    private long value;
 
    public void set(long value)
    {
        this.value = value;
    }
}
static class Translator implements EventTranslatorOneArg<LongEvent, Long>{
    @Override
    publicvoid translateTo(LongEvent event, long sequence, Long data) {
        event.set(data);
    }    
}
    
public static Translator TRANSLATOR = new Translator();
    
public staticvoid publishEvent2(Disruptor<LongEvent> disruptor) {
    // 发布事件；
    RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    long data = getEventDataxxxx();//获取要通过事件传递的业务数据；    
    ringBuffer.publishEvent(TRANSLATOR,data);
}

多个生产者

    在构建Disruptor实例的时候,需要指定生产者是单生产者（ProducerType.SINGLE）还是多生产者（ProducerType.MULTI）

多个消费者

   （类型1）   多个消费者每个消费者都有机会消费相同数据，使用handleEventsWith方法

class ComsumerHandler implements EventHandler<ResultTxt>{
    
    private int no;
 
    public  ComsumerHandler(int no) {
        this.no=no;
    }
 
    @Override
    public void onEvent(ResultTxt resultTxt, long sequence, boolean endOfBatch)
        throws Exception {
        System.out.println(no+" data commming......"+resultTxt.getBarCode());
    }
}
 
//设置多个消费者
disruptor.handleEventsWith(new ComsumerHandler(1),new ComsumerHandler(2));

     （类型2）  多个消费者，每个消费者消费不同数据。也就是说每个消费者竞争数据，竞争到消费，其他消费者没有机会。使用handleEventsWithWorkerPool方法

class ComsumerHandler implements WorkHandler<ResultTxt>{
 
    private int no;
 
    public  ComsumerHandler(int no) {
        this.no=no;
    }
 
    @Override
    public void onEvent(ResultTxt event)
        throws Exception {
        System.out.println(no+"  data commming......"+event.getBarCode());
    }
}
 
//多个消费者，每个消费者竞争消费不同数据
disruptor.handleEventsWithWorkerPool(new ComsumerHandler(1),new ComsumerHandler(2));

 

等待策略

生产者的等待策略

暂时只有休眠1ns。

LockSupport.parkNanos(1);

消费者的等待策略

更多详细信息请参考: http://tech.meituan.com/disruptor.html