【Java】阻塞队列 BlcokingQueue 原理

while + sleep

自从Java 1.5之后，在java.util.concurrent包下提供了若干个阻塞队列，BlcokingQueue继承了Queue接口，是线程安全的。阻塞功能使得生产者和消费者两端的能力得以平衡，当阻塞队列是空时，从队列中获取元素的操作将会被阻塞。当阻塞队列是满时，往队列里添加元素的操作将会被阻塞。

等待唤醒机制 condition/await/singal 和阻塞队列的关系将在第四部分介绍。

比较和交换（Compare And Swap，CAS） 将在文章第五部分介绍。

channel

#1 阻塞队列的种类

ArrayBlockingQueueLinkedBlockingQueuePriorityBlockingQueueDelayQueueSynchronousQueueLinkedTransferQueue

ConcurrentLickedQueue

#2 常用方法

队列主要方法

对于非阻塞队列，一般情况下建议使用offer、poll和peek三个方法，因为使用offer、poll和peek三个方法可以通过返回值判断操作成功与否，而使用add和remove方法却不能达到这样的效果。注意，非阻塞队列中的方法都没有进行同步措施。

add(E e)remove()element()offer(E e)poll()peek()

阻塞队列主要方法

上面列举的5个方法在阻塞队列中都存在，但是要注意这5个方法在阻塞队列中都进行了同步措施。

put(E e)take()offer(E e,long timeout, TimeUnit unit)poll(long timeout, TimeUnit unit)

操作类型	抛出异常	阻塞线程	有返回值	超时返回
存入	add	put	offer	offer(E e, long timeout, TimeUnit unit)
取出	remove	take	poll	poll(long timeout, TimeUnit unit)

#3 阻塞队列的原理 - 以 ArrayBlockingQueue 为例

takeIndexputIndexcountlocknotEmptynotFull

ReentrantLocknotEmptynotFull

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
 
    private static final long serialVersionUID = -817911632652898426L;
    /** 用于存放元素的数组 The queued items */
    private final E[] items;
    /** 下一次读取的位置 items index for next take, poll or remove */
    private int takeIndex;
    /** 下一次写入的位置 items index for next put, offer, or add. */
    private int putIndex;
    /** 队列中元素的数量 Number of items in the queue */
    private int count;
     
    /*
    * Concurrency control uses the classic two-condition algorithm
    * found in any textbook.
    */
     
    /** Main lock guarding all access */
    private final ReentrantLock lock;
    /** Condition for waiting takes */
    private final Condition notEmpty;
    /** Condition for waiting puts */
    private final Condition notFull;
}

put()

checkNotNulllock.lockInterruptibly()

countenqueueunlock

public void put(E e) throws InterruptedException {
        //判断添加元素e是否为空，为空会抛出异常
        checkNotNull(e);
        //获取锁
        final ReentrantLock lock = this.lock;
        //加锁
        lock.lockInterruptibly();
        try {
            //队列已满，线程等待
            while (count == items.length)
                notFull.await();
            //入队方法 signal()藏在enqueue(e)方法里了
            enqueue(e);
        } finally {
            //释放锁
            lock.unlock();
        }
    }


private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
  //插入数据
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
  //通知消费者线程，当前队列中有数据可供消费
    notEmpty.signal();
}

take()

notFullnotEmpty

//从队列中取出数据
public E take() throws InterruptedException {
        //获取锁
        final ReentrantLock lock = this.lock;
        //加锁
        lock.lockInterruptibly();
        try {
            //当队列中无数据，删除线程阻塞等待
            while (count == 0)
                notEmpty.await();
            //有数据，执行队头元素出队操作
            return dequeue();
        } finally {
            //释放锁
            lock.unlock();
        }
    }

private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex]; //取出takeIndex位置的元素
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0; //设计的精髓： 环形数组，takeIndex 指针到数组尽头了，返回头部
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    //notFull条件队列转同步队列，准备唤醒生产者线程，此时队列有空位
    notFull.signal();
    return x;
}

非阻塞队列ConcurrentLinkedQueue

顺便看看使用CAS非阻塞算法+不停重试来实现线程安全的这个队列。

ConcurrentLinkedQueueLinkedBlockingQueue

public boolean offer(E e) {
    // 检查e是不是null，是的话抛出NullPointerException异常。
    checkNotNull(e);
    // 创建新的节点
    final Node<E> newNode = new Node<E>(e);

    // 将“新的节点”添加到链表的末尾。
    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        // 情况1：q为空，p是尾节点插入
        if (q == null) {
            // CAS操作：如果“p的下一个节点为null”(即p为尾节点)，则设置p的下一个节点为newNode。
            // 如果该CAS操作成功的话，则比较“p和t”(若p不等于t，则设置newNode为新的尾节点)，然后返回true。
            // 如果该CAS操作失败，这意味着“其它线程对尾节点进行了修改”，则重新循环。
            if (p.casNext(null, newNode)) {
                if (p != t) // hop two nodes at a time
                    casTail(t, newNode);  // Failure is OK.
                return true;
            }
        }
        // 情况2：p和q相等
        else if (p == q)
            p = (t != (t = tail)) ? t : head;
        // 情况3：其它
        else
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

非阻塞队列ConcurrentLinkedQueue主要流程就是在判空以后，会进入一个大循环中，p.casNext()方法，这个方法正是利用了CAS来操作的，这个死循环去配合CAS其实就是我们平时说的乐观锁的思想。

#4 等待唤醒机制 wait/notify/notifyAll、condition/await/singal

从上一部分可以发现，阻塞队列 = 等待唤醒机制 + 队列，例如 ArrayBlockingQueue 使用了 ReentrantLock + Condition 和一个列表队列实现。这里就顺便说说 Java 中的阻塞唤醒机制。此部分参考这篇好文。

轮询

生产者和消费者线程各自使用while循环，每隔片刻就去判断Queue的状态，队列为空时生产者才可插入数据，队列不为空时消费者才能取出数据，否则一律sleep等待。

轮询最简单但是也最不优雅。

• sleep 时不会释放线程拥有的锁。
• 轮询的方式太耗费CPU资源：如果线程过多，比如几百上千个线程同时在那轮询，会给CPU带来较大负担。
• 无法保证原子性：如果生产者的操作非原子性，消费者极可能获取到脏数据。

wait/notify/notifyAll

等待唤醒机制底层通过维护线程队列的方式，避免了过多线程同时自旋造成的CPU资源浪费，颇有点“用空间换时间”的味道。

当一个生产者线程无法插入数据时，就让它在队列里休眠（阻塞），此时生产者线程会释放CPU资源，等到消费者抢到CPU执行权并取出数据后，再由消费者唤醒生产者继续生产。

synchronizedwaitnotify

// 线程consumer  
synchronize(obj){ 
    obj.wait();//没东西了，等待
}  
// 线程producer
synchronize(obj){ 
    obj.notify();//有东西了，唤醒 
}

在synchronized机制下，所有等待的线程都在同一个队列里，而notify又恰巧是随机唤醒线程（也就是说，有可能生产者唤醒生产者）。

notifyAll

• 如果线程调用了对象的 wait()方法，那么线程便会处于该对象的等待池中，等待池中的线程不会去竞争该对象的锁。
• 当有线程调用了对象的 notifyAll()方法（唤醒所有 wait 线程）或 notify()方法（只随机唤醒一个 wait 线程），被唤醒的的线程便会进入该对象的锁池中，锁池中的线程会去竞争该对象锁。也就是说，调用了notify后只要一个线程会由等待池进入锁池，而notifyAll会将该对象等待池内的所有线程移动到锁池中，等待锁竞争
• 优先级高的线程竞争到对象锁的概率大，假若某线程没有竞争到该对象锁，它还会留在锁池中，唯有线程再次调用 wait()方法，它才会重新回到等待池中。而竞争到对象锁的线程则继续往下执行，直到执行完了 synchronized 代码块，它会释放掉该对象锁，这时锁池中的线程会继续竞争该对象锁。

condition/await/singal

wait/notifyAll

Conditionwait/notify

// 显式锁（相对地，synchronized锁被称为隐式锁）
private final ReentrantLock lock = new ReentrantLock();
private final Condition producerCondition = lock.newCondition();
private final Condition consumerCondition = lock.newCondition();

// 队列满，生产者等待
lock.lock(); 
producerCondition.await(); 
lock.unlock();  

// 提醒消费者消费
lock.lock(); 
consumerCondition.signal(); 
lock.unlock();

lock.newCondition()producerCondition.await()producerCondition.signal()

也就是说，ReentrantLock的Condition通过拆分线程等待队列，让线程的等待唤醒更加精确了，想唤醒哪一方就唤醒哪一方。

put()take()

#5 多线程安全 - CAS

补充一下这个概念。

在计算机科学中，比较和交换（Compare And Swap，CAS） 是用于实现多线程同步的原子指令。 它将内存位置的内容与给定值进行比较，只有在相同的情况下，将该内存位置的内容修改为新的给定值。

CAS(V,E,N)

CAS 操作是抱着乐观的态度进行的(乐观锁)，它总是认为自己可以成功完成操作。当多个线程同时 使用 CAS 操作一个变量时，只有一个会胜出，并成功更新，其余均会失败。失败的线程不会被挂起，仅是被告知失败，并且允许再次尝试，当然也允许失败的线程放弃操作。基于这样的原理， CAS 操作即使没有锁，也可以发现其他线程对当前线程的干扰，并进行恰当的处理。

独占锁是一种悲观锁，synchronized就是一种独占锁，会导致其它所有需要锁的线程挂起，等待持有锁的线程释放锁。而另一个更加有效的锁就是乐观锁。所谓乐观锁就是，每次不加锁而是假设没有冲突而去完成某项操作，如果因为冲突失败就重试，直到成功为止。乐观锁用到的机制就是CAS，Compare and Swap。

自旋： 就是不停的判断比较，看能否将值交换

CAS的三大问题：

• ABA问题：一个值原来是A，变成了B，又变成了A，那么使用CAS进行检查时会发现它的值没有发生变化，但是实际上却变化了。ABA问题的解决思路就是使用版本号，每次变量更新的时候把版本号加一，那么A－B－A 就会变成1A-2B－3A。
• 循环时间长开销大：自旋CAS如果长时间不成功，会给CPU带来非常大的执行开销。如果JVM能支持处理器提供的pause指令那么效率会有一定的提升。
• 只能保证一个共享变量的原子操作：对多个共享变量操作时，循环CAS就无法保证操作的原子性，这个时候就可以用锁。还有一个取巧的办法，就是把多个共享变量合并成一个共享变量来操作。