致敬disruptor：CAS实现高效（伪）无锁阻塞队列实践

引言

在多线程开发中，我们常常遇到这样一种场景：一些线程接受用户请求，另外一些线程处理这些请求，之所以把接受请求和处理请求的逻辑分开，一方面是出于资源调度的考虑（用户请求也许很多，但这些请求涉及的资源很少），另一方面也可能是异步响应的需求。这种场景存在于NIO的通信框架，存在于Tomcat的回调处理框架，存在于日志系统的异步flush，存在于各种类型的线程池中。总之，这种典型的生产者消费者场景比比皆是，而生产者消费者模式的核心就是阻塞队列。由于阻塞队列会涉及大量的锁竞争和线程阻塞，都是非常耗费CPU的操作，因此阻塞队列的性能好坏能够在很大程度上决定上层应用的性能瓶颈。

不同语言中，都会有（较为）官方的阻塞队列实现，JAVA中用BlockingQueue这个接口来描述阻塞队列，有数组实现的有界阻塞队列为ArrayBlockingQueue，用链表实现的无界阻塞队列为LinkedBlockingQueue，除此之外还有优先级阻塞队列PriorityBlockingQueue，这些阻塞队列除了自身特有逻辑外，都采用基于锁的悲观并发控制，即标准的生产者消费者模型（见操作系统教材）。这种模型由于锁冲突严重，已经被证明是一种低效的实现。那么它有优化空间吗？优化空间能有多少？

最近很火的Disruptor给了我们一个答案，在使用了Ringbuffer，内存屏障，乐观并发控制等众多优化手段后，Disrupter的阻塞队列与传统的阻塞队列相比有超过10倍的吞吐率。

遗憾的是，Disruptor主要应用于订阅模式，即一条消息被所有消费者消费，另外Disruptor支持构建具有复杂依赖关系的消费者，Storm底层便利用Disruptor来构建本地任务拓扑。而语言自带的阻塞队列一般是一个消息被单个消费者消费，这种场景更加通用。

介绍Disruptor不在本文范围内，有机会的话会另开一篇博文做Disruptor的源码解析，本文想做的是，利用Disruptor中RingBuffer和乐观并发控制的设计思想，实现一个单消息单消费的通用阻塞队列。

RingBuffer实现

RingBuffer是一个首尾相连的环形数组，所谓首尾相连，是指当RingBuffer上的指针越过数组是上界后，继续从数组头开始遍历。因此，RingBuffer中至少有一个指针，来表示RingBuffer中的操作位置。另外，指针的自增操作需要做并发控制，Disruptor和本文的OptimizedQueue都使用CAS的乐观并发控制来保证指针自增的原子性，关于乐观并发控制之后会着重介绍。

Disruptor中的RingBuffer上只有一个指针，表示当前RingBuffer上消息写到了哪里，此外，每个消费者会维护一个sequence表示自己在RingBuffer上读到哪里，从这个角度讲，Disruptor中的RingBuffer上实际有消费者数+1个指针。由于我们要实现的是一个单消息单消费的阻塞队列，只要维护一个读指针（对应消费者）和一个写指针（对应生产者）即可，无论哪个指针，每次读写操作后都自增一次，一旦越界，即从数组头开始继续读写。如图1所示：

图1：RingBuffer结构

RingBuffer的元素个数最好设置为2的整数次幂，这样可以通过与运算获取指针的真实位置（ 有人问到溢出问题，解释见评论），Disruptor正是这样做的。RingBuffer的实现摘要如下：

private Object[] ringBuffer = null;
private Object[] ringBuffer = null;
private AtomicInteger offerSeq = new AtomicInteger(-1);
private AtomicInteger takeSeq = new AtomicInteger(-1);
private int size;
private int mask;

public OptimisticQueue(int sizePower) {
	this.size = 1 << sizePower;
	this.ringBuffer = new Object[size];
	for (int i = 0; i < size; i++) {
		ringBuffer[i] = new Entry(i + 1);
	}
	this.mask = 0x7FFFFFFF >> (31 - sizePower);
}

@SuppressWarnings("unchecked")
private Entry nextOffer() {
	return (Entry) ringBuffer[offerSeq.incrementAndGet() & mask];
}

@SuppressWarnings("unchecked")
private Entry nextTake() {
	return (Entry) ringBuffer[takeSeq.incrementAndGet() & mask];
}

同是数组，我们会发现实际上ArrayBlockingQueue也用的是RingBuffer，从源码中可以看到ArrayBlockingQueue包含一个putIndex和一个takeIndex，分别对应写指针和读指针，每次读写后Index自增，且判断是否等于数组长度，等于的话归0：

final int inc(int i) {
        return (++i == items.length)? 0 : i;
    }

那么ArrayBlockingQueue中的RingBuffer和我们这里说的RingBuffer有区别吗？

区别很大。

RingBuffer中之所以分离读指针和写指针，为的是使生产者和消费者互不干扰，两者可以同时操作指针，达到完全并发执行。然而ArrayBlockingQueue中对所有的读写操作都进行了加锁互斥，也就是说同一时刻只能有一个生产者或消费者操作，这其实和用一个指针没什么区别，没有发挥RingBuffer的真正优势，同时这也是ArrayBlockingQueue本身的一个瓶颈。

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        try {
            while (count == 0)
            notEmpty.await();
        } catch (InterruptedException ie) {
            notEmpty.signal(); // propagate to non-interrupted thread
            throw ie;
        }
        E x = extract();
        return x;
    } finally {
        lock.unlock();
    }
}

那么为什么ArrayBlockingQueue中要对所有的操作都做互斥？这个问题一言两语说不清楚，留给读者自己推敲，一言以蔽之： 在基于锁的悲观并发控制下，在阻塞队列中互斥所有操作是最正确的做法。

既然基于锁的悲观并发控制需要互斥所有操作，RingBuffer无法发挥其“读写分离”的优势，那么怎样才能不互斥读写操作呢？

这就要说到CAS与乐观并发控制了。

CAS与乐观并发控制

悲观并发控制，是用锁把可能出现并发操作的临界区保护起来，以杜绝临界区内的并发，将并发错误防范于未然，正如其名，是在“悲观”地认为会发生错误而设计的并发控制策略。

乐观并发控制则相反：在临界区的执行过程中，乐观地认为不会发生并发冲突，没有并发错误自然无需加锁。OK，不加锁性能自然好多了，那如果真地发生并发冲突了呢?首先我们需要随时知道并发冲突发生了没有，也就是需要一种并发冲突检测机制，冲突检测可以在执行临界区之前，也可以在执行临界区之后。OK，假设我们现在有一种冲突检测机制，接下来怎么办呢？自然是需要一种冲突解决机制了，不同的冲突检测机制对应不同的冲突解决机制，这要视场景和情况而定。

说的这么玄乎，举个例子：

public final int incrementAndGet() {
    for (;;) {
        int current = get();
        int next = current + 1;
        if (compareAndSet(current, next))
            return next;
    }
}

上述代码为JDK中AtomicInteger的原子自增操作，可以看到它依赖的是CAS，即compareAndSet操作来检测自增操作是否有并发冲突，当没有并发冲突时，compareAndSet比较当前值等于自增前的值，并成功将自增后的值赋给自己，若有并发冲突发生，也就是有其他线程已经更改了当前值，compareAndSet会返回false，并重新执行自增操作

上述案例中，CAS是冲突检测机制，重新执行自增操作是冲突解决机制。CAS的冲突检测发生在临界区执行之前（这里的临界区是给自己赋值）。

顺带一提，上述的循环操作是不是有点像spinlock？ 其实spinlock也是一种乐观并发控制。而Mutex，在JAVA中又叫Monitor这种包含阻塞逻辑的锁才真正对应悲观并发控制。

我这里也有一个冲突检测发生在临界区执行之后的案例：著名的MySQL高可用解决方案Galera是通过对比WriteSet中主键的值来检测是否有不同Master的事务对同一行数据进行了操作，这种检测是发生在事务commit前，同步WriteSet后，这时事务中所有的操作都已执行完毕，若检测到事务冲突，直接进行回滚。这里检测writeset主键值冲突是冲突检测机制，事务回滚是冲突解决机制。

由此可见，虽然悲观并发控制都是基于锁的，但乐观并发控制的方式就多种多样了，需要根据场景对症下药。

我们先来看看怎样通过CAS来检测队列中的并发冲突。

图2. RingBuffer的Entry结构

如图2所示，Entry即RingBuffer中的元素，这里为每个Entry定义一个frontDoor和backDoor，即前门和后门，frontDoor和backDoor各为一个AtomicReference（对象的原子引用，对这个类不熟的Google之），对生产者A，当它要往一个Entry中写消息时，需要先调用frontDoor的CAS(null, A)来检测是否能进入前门，若返回true，表示A获得了写Entry的权利，在A写完Entry后，会将frontDoor的值重置为null。

在A写Entry的这段时间，frontDoor的值为A，其他生产者也想写这个Entry时，会在CAS frontDoor时检测到并发冲突。同理，对消费者B，当它想要消费Entry中的消息时，要通过backDoor.CAS(null, B)来检测是否有读的并发冲突。

前门和后门的AtomicReference就像两个单人通道，保证了一个Entry在一个时间点上只会有一个生产者和消费者来读写消息。

这里有个疑问，CAS实现指针自增可以保证多个生产者或多个消费消费者不会进入一个entry，那这个前门后门的设计是为了什么？我们假设这样一个场景：有entry为10的ringbuffer，2个生产者，生产者1获取entry1消息的速度非常慢（假设该线程因为不明原因卡住了），生产者2同时会快速消费掉entry2-10，绕了一圈后进入entry1获取消息，这时候entry1中就会有2个生产者，也就是说 指针上的CAS并不能严格保证一个entry上不会有多个生产者或多个消费者并发访问，但是有了frontDoor这个屏障，在生产者2进入entry1后，发现entry1的前门中已经有了生产者1，即可立即发现冲突。

冲突检测机制有了，还需要一个冲突解决机制，很容易想到，无论对生产者还是消费者，检测到冲突后只要对下一个指针上的Entry执行操作即可。当然对下一个Entry也要做冲突检测。

目前为止，一切看起来很美好，我们用CAS来检测生产者与生产者，消费者与消费者之间的并发冲突，通过指针+1来解决冲突。我们是不是遗漏了什么？

• 当生产者A进入frontDoor后，如果发现Entry中已经有了消息怎么办？相应的，若消费者B进入backDoor后，发现Entry中还没有消息怎么办？
• AS实现指针+1解决了生产者与生产者之间，消费者与消费者之间的并发冲突问题，同一个Entry上的生产者和消费者就不会出现并发冲突吗？

先回答第一个问题，我们知道，任何有界队列都可能出现队列被写满的情况，队列被写满后生产者在写消息时被阻塞，直到队列被消费一点后才被唤醒，把消息塞进队列。相应的，在队列为空时消费者会阻塞，直到队列中有消息后才会被唤醒，来消费消息。这种生产者满则阻塞和消费者空则阻塞的模式正是阻塞队列的名称由来。

当生产者A进入frontDoor后，发现Entry中已经有了消息，生产者满则阻塞，相应的，若消费者B进入backDoor后，发现Entry中还没有消息，消费者空则阻塞。传统的队列中，生产者消费者阻塞都发生在队列全体满或者空的状态后，而我们这种队列的阻塞则是针对每个Entry而言的。

稍作斟酌，将阻塞逻辑细化到每个Entry中，会不会使阻塞频繁发生呢？倘若如此，这种优化就很难笃定地说是一种优化了。这里RingBuffer的作用再次凸显出来，在RingBuffer中读写就像赛跑，只有生产者们超越消费者们一圈，才可能出现上述的生产者满则阻塞，而这时也正是队列被写满的情况，相应的，只有消费者们追上生产者们，才可能出现消费者空则阻塞，这时也正是队列为空的情况。所以不会使阻塞频繁发生。

对第二个问题，同一个Entry上的生产者和消费者实际上是通过Entry上的消息进行线程同步，当生产者A写完消息后，通知消费者B获取消息，或者消费者B消费完消息后，通知生产者A可以写新的消息。两者之间通过wait和notify原语做线程同步，因此不会有并发冲突。

写到这里，我们已经实现了一个基于RingBuffer和CAS乐观并发控制的无锁阻塞队列，接下来我们来聊聊这个队列的实现细节。

两阶段发布

上文提到，无论是生产者还是消费者，在操作完消息后，都需要通知Entry的“另一头”，Disruptor中管这种行为叫做“两阶段发布”，例如对生产者来说，第一次发布是将消息赋给Entry中的item，第二次发布是通知backDoor中的消费者（如果有）消息到了，如图3所示：

图3. 两阶段发布流程

需要特别留意的是，如果我们不在item上做任何互斥，可能出现“干等”的情况，例如当生产者A发现item不为空，进入阻塞之前，消费者B取走了消息，并向生产者A发出通知，由于这时候A还没有进入阻塞，导致通知失效，A会永无止境地等下去，而且Entry中的消息也会一直为空，新进来的消费者也会一直等下去，进入活锁。

生产者A　　　　　　　　　　　　　消费者B

check item != null?

　　　　　　　　　　　　　　check item != null?

　　　　　　　　　　　　　　take(item), item = null

　　　　　　　　　　　　　　notify producer wakeup

wait consumer notify….

为了避免活锁，我们这里引入JAVA中常用的一个编程技术：双重检查锁定，我们知道Object的wait和notify需要写在synchronized中，双重检查锁定就是除了在一开始判断item是否为null外，在synchonized内部再判断一次。

下面给出Entry的完整实现，其中publish(T event)为发布消息，take(T event)为取走消息，由于Entry是OptimizedQueue的内部类，类型参数T无需再声明。

private class Entry {
    private volatile T event = null;
    private AtomicReference<Object> backDoor 
        = new AtomicReference<Object>();
    private AtomicReference<Object> frontDoor 
        = new AtomicReference<Object>();
    private int id;

    public Entry(int id) {
        this.id = id;
    }

    public void publish(T event) {
        this.event = event;
        frontDoor.set(null);
        Object barrier = backDoor.get();
        if (barrier != null) {
            synchronized (barrier) {
                barrier.notify();
            }
        }
    }

    public boolean enterFront(Object barrier) {
        return frontDoor.compareAndSet(null, barrier);
    }

    public boolean enterBack(Object barrier) {
        return backDoor.compareAndSet(null, barrier);
    }

    public T take() {
        T e = event;
        event = null;
        backDoor.set(null);
        Object barrier = frontDoor.get();
        if (barrier != null) {
            synchronized (barrier) {
                barrier.notify();			
            }
        }
        return e;
    }

    public int getId() {
         return id;
    }
}

通过synchronized内部的while判断，保障了生产者和消费者都不会死等，活锁的问题是解决了，但我们也发现我们的代码中再次引入了“锁”（虽然没有while循环也无法消除synchronized，但纯粹为了线程同步而加的synchronized严格意义上讲不算锁，如果JAVA能从native层保障wait和notify的原子性，完全可以做到消除synchronized）。所以严格地说，OptimizedQueue没有消除锁，而是将整个队列上的一把大锁细化到了每个Entry中，即所谓的“ 锁细化”，锁细化在很大程度上降低了锁冲突概率，并且把加锁的时间控制到了最少，通过上一篇synchronized原理深探，可以知道synchornized的代价在这里非常小（基本不会生成Monitor）。

下面贴出OptimizedQueue的完整实现：

/*
 * every producer and consumer must implement this interface
 * Barrier is just a object lock for the synchronization between 
 * producer and consumer
 */
public interface BarrierHolder {
	Object getBarrier();
}

public class OptimisticQueue<T> {
	private Object[] ringBuffer = null;
	private AtomicInteger offerSeq = new AtomicInteger(-1);
	private AtomicInteger takeSeq = new AtomicInteger(-1);
	private int size;
	private int mask;

	public OptimisticQueue(int sizePower) {
		this.size = 1 << sizePower;
		this.ringBuffer = new Object[size];
		for (int i = 0; i &lt; size; i++) {
			ringBuffer[i] = new Entry(i + 1);
		}
		this.mask = 0x7FFFFFFF >> (31 - sizePower);
	}

	@SuppressWarnings("unchecked")
	private Entry nextOffer() {
		return (Entry) ringBuffer[offerSeq.incrementAndGet() & mask];
	}

	@SuppressWarnings("unchecked")
	private Entry nextTake() {
		return (Entry) ringBuffer[takeSeq.incrementAndGet() & mask];
	}

	public void offer(BarrierHolder holder, T event) throws 
            InterruptedException {
		Entry entry = nextOffer();
		Object barrier = holder.getBarrier();
		if (!entry.enterFront(barrier)) {
			offer(holder, event);
			return;
		}
		if (entry.event != null) {
			synchronized (barrier) {
				while (entry.event != null) {
					barrier.wait();
				}
			}
		}
		entry.publish(event);
	}

	public T take(BarrierHolder consumer) throws InterruptedException {
		Entry entry = nextTake();
		Object barrier = consumer.getBarrier();
		if (!entry.enterBack(barrier))
			return take(consumer);
		if (entry.event == null) {
			synchronized (barrier) {
				while (entry.event == null) {
					barrier.wait();
				}
			}
		}
		return entry.take();
	}

	private class Entry {
	......
	}
}

使用场景

本文中实现的OptimizedQueue的使用场景有如下限制：

1． OptimizedQueue类似于JDK中的ArrayBlockingQueue，是有界数组，当队列中挤压的消息数大于数组长度时，生产者再想塞入消息会阻塞。

2． OptimizedQueue的参数为RingBuffer长度的自然对数，例如当参数为10时，RingBuffer长度为1024，这样设计是为了用更高效的方式获取真实指针位置。

3． OptimizedQueue的生产者和消费者的个数都必须小于RingBuffer长度，试想，倘若生产者的个数大于RingBuffer长度，可能出现某个生产者因为RingBuffer中所有frontDoor都被占用而一直找不到槽位，以至于进入死循环。消费者亦然。

OptimizedQueue的实际限制其实只有3，而在大多数应用场景中，都能保障3的条件，因为线程是非常消耗资源的对象，相对的RingBuffer的Entry只会消耗内存中非常少量的空间，在使用OptimizedQueue需要保证RingBuffer长度大于生产者消费者上限。

阻塞队列一个典型的应用是线程池，例如在网易分布式数据库DDB中，一个简单的select语句在创建完执行计划后，可能要将SQL下发给多个mysql节点，这里就需要线程池来完成多个mysql节点上的查询，而单个client连接数有1024的最大限制，这种情况下只要保证RingBuffer长度大于1024，并且连接池中的线程个数设上限为1023，即可用OptimizedQueue构建线程池。

Benchmark

最后我们来看看OptimizedQueue在性能上究竟比JDK的自带阻塞队列高出多少。

为了衡量队列性能，设计了一个简单的生产者消费者场景，生产者初始化一个整数2000，并从1一直累加到2000，然后将2000放入队列，生产者从队列中取出2000，也同样从1累加到2000，累加完后吞吐量+1，这个场景本身没什么逻辑，只是为了测量性能而设计。这里就不再贴代码了，附件中的源码中包含完整的benchmark实现。

用于对比的是OptimizedQueue和JDK中的ArrayBlockingQueue，因为ArrayBlockingQueue也是数组实现，而且相比无界队列LinkedBlockingQueue性能更好。性能指标为队列在各种生产者和消费者个数下的吞吐率（实际上是单位时间内生产者完成的计算量）。

测试分别在两个机器上进行，第一个是公司的办公机，性能较差，第二个是我自己的笔记本，因为比较新，性能好。

办公机的性能测试结果如下：

配置烂，用eclipse写代码没法网页听歌，干什么都卡的一B

我的笔记本的测试结果如下：

i7 2.9GHZ 双核，平时用起来很流畅

在我的办公机上，两种队列性能差距非常明显，尤其是10个生产者和1个消费者的场景下，OptimizedQueue与ArrayBlockingQueue有将近 60倍的性能差距。相对的，在我的笔记本上，两者差距也存在，但不那么明显了，两者差距大小应该和业务场景，机器配置甚至CPU架构都相关。

值得关注的是表格中标红的数据，用办公机在1个生产者和10个消费者的测试场景下，ArrayBlockingQueue的吞吐率反而高出很多，而在接下来的测试中，TPS直接降到了个位数，而在我自己性能更好的笔记本中，同等场景下的吞吐率也才只有31W。因只能用“异常”来形容这组数据了。ArrayBlockingQueue是用ReentrantLock来实现的，大概是恰巧踩到了什么优化点。当然我们不能期待现实环境中总能遇到这样的情况。另外，这几组数据中，使用ArrayBlockingQueue在生产者个数大于消费者个数时性能较差，各种原因值得玩味。

本文100%原创，转载请注明出处，另外本人能力有限，若觉文中有任何谬误之处，欢迎交流讨论，源码在附件中。

OptimisticQueue源码下载