BlockingQueue、ArrayBlockingQueue、LinkedBlockingQueue原理分析

阻塞队列与非阻塞队

阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操作将会被阻塞，或者当队列是满时，往队列里添加元素的操作会被阻塞。试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。同样，试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来，如从队列中移除一个或者多个元素，或者完全清空队列.

1.ArrayDeque, （数组双端队列） 
2.PriorityQueue, （优先级队列） 
3.ConcurrentLinkedQueue, （基于链表的并发队列） 
4.DelayQueue, （延期阻塞队列）（阻塞队列实现了BlockingQueue接口） 
5.ArrayBlockingQueue, （基于数组的并发阻塞队列） 
6.LinkedBlockingQueue, （基于链表的FIFO阻塞队列） 
7.LinkedBlockingDeque, （基于链表的FIFO双端阻塞队列） 
8.PriorityBlockingQueue, （带优先级的无界阻塞队列） 
9.SynchronousQueue （并发同步阻塞队列）

BlockingQueue是一个接口单向的阻塞队列，

BlockingDueue是一个接口双向的阻塞队列，

其源码分析如下：

public interface BlockingQueue<E> extends Queue<E> {
   
    //将指定的元素添加到对列中，如果队列容量未满。如果满了，抛出异常
    boolean add(E e);

    //如果可以在不违反容量限制的情况下立即将指定元素插入此队列，则在成功时返回 true，如果当前 
    //没有可用空间则返回 false。 当使用容量受限的队列时，这种方法通常比 add 更可取，它可以仅通 
    //过抛出异常来插入元素失败。
    boolean offer(E e);

    //将指定的元素插入此队列的尾部，如果该队列已满，则一直等到（阻塞）。 
    void put(E e) throws InterruptedException;

     //带有等待时间的入队方法
    boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException;

   //获取并移除此队列的头部，如果没有元素则等待（阻塞）， 
	//直到有元素将唤醒等待线程执行该操作 
    E take() throws InterruptedException;

   //与offer对应
    E poll(long timeout, TimeUnit unit)
        throws InterruptedException;

     //查看队列还有的容量
    int remainingCapacity();

   // 删除操作，与add对应
    boolean remove(Object o);


    public boolean contains(Object o);


    int drainTo(Collection<? super E> c);


    int drainTo(Collection<? super E> c, int maxElements);
}

查看接口的继承关系图

接下来我们重点介绍 ArrayBlockingQueue、LinkedBlockingQueue、LinkedBlockingDeque

ArrayBlockingQueue的源码解析

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    //队列底层容器
    final Object[] items;

    /** 下一个 take, poll, peek or remove的数组下标位置 */
    int takeIndex;

    /** 下一个 next put, offer, or add的数组下标位置 */
    int putIndex;

    /** 数组的已有数量 */
    int count;

    /**  队列锁******/
    final ReentrantLock lock;

    /** 获取队列数据的的condition等待队列 */
    private final Condition notEmpty;

    /** 存入队列数据的condition 等待队列*/
    private final Condition notFull;

    transient Itrs itrs = null;

     //队列构造方法必须指定大小，有界队列
    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }

    // boolean fair ture为公平锁，严格的按照FIFO的原则获取锁，默认非公平锁
    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    代码省略。。。。。。。。
}

接下来看源码分析ArrayBlockingQueue是如何对队列进行add-remove put-take offer-poll

  //add方法实际是调用的offer方法如如果成功返回true 否则抛出异常   
   public boolean add(E e) {
        return super.add(e);
    }

 
    public boolean offer(E e) {
        //检查传入的数据是否为空
        checkNotNull(e);
         // 定义锁 final变量，构造器类初始化的时候就已经申请好，state变量初始化好内存。
        final ReentrantLock lock = this.lock;
        lock.lock();
        try { 
           //判断当前的count是否等于数组的容量
           // 是返回fasle 否则加入到队列中
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

      private void enqueue(E x) {
        final Object[] items = this.items;
         // putIndex 入队的下一个数组下标值
        items[putIndex] = x;
//需要将putIndex重新设置为0，这是因为当前队列执行元素获取时总是从队列头部获取，而添加元素从中//从队列尾部获取所以当队列索引（从0开始）与数组长度相等时，
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        //当队列数据不为空时唤醒取数队列的等待condtion表示可以去获取数据了
        notEmpty.signal();
    }


    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        //中断锁。当前线程可以响应中断 中断后，抛出异常
        lock.lockInterruptibly();
        try {
            while (count == items.length)
               //当队列满了以后，那么入队的condition等待队列 等待，将该线程加入的等待队列中
                notFull.await();
            //否则入队操作
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

LinkedBlockingQueue的源码解析

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
   
     //存储数据的节点
    static class Node<E> {
        E item;
        Node<E> next;
        Node(E x) { item = x; }
    }

    /** 容量大小 默认为 Integer.MAX_VALUE */
    private final int capacity;

    /** 当前队列的数量 */
    private final AtomicInteger count = new AtomicInteger();

     //头节点
    transient Node<E> head;

    //尾节点
    private transient Node<E> last;

    /** 取数据锁对象 */
    private final ReentrantLock takeLock = new ReentrantLock();

    /**  Wait queue for waiting takes  */
    private final Condition notEmpty = takeLock.newCondition();

    /** put数据锁对象 */
    private final ReentrantLock putLock = new ReentrantLock();

    /** 等待队列条件 */
    private final Condition notFull = putLock.newCondition();

由上面的LinkedBlockingQueue源码可以分析到 LinkedBlockingQueue 是一个基于链表的阻塞的无边界的队列。默认的队列容量为Integer.MAX_VALUE  当然这样容易引发的问题，当入列速率远超过出队速率时，可能会造成内存溢出、即超出容量的最大值。另外ArrayBlockingQueue入队与出队只有一把锁。但是LinkedBlockingQueue入队和出队是不同的锁对象、提高了队列的吞吐量

public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        final AtomicInteger count = this.count;
        if (count.get() == capacity)
            return false;
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            if (count.get() < capacity) {
                enqueue(node);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    // 如果容量还够。唤醒入队的condtion去写入数据
                    notFull.signal();
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0)
             // 如果容量为0表示队列还有一条数据 （原来的值为-1） 。唤醒入队的condtion去写入数据
            signalNotEmpty();
        return c >= 0;
    }

take阻塞获取的方法

public E take() throws InterruptedException {
        E x;
        int c = -1;
        //获取当前队列大小
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();//可中断
        try {
            //如果队列没有数据，挂机当前线程到条件对象的等待队列中
            while (count.get() == 0) {
                notEmpty.await();
            }
            //如果存在数据直接删除并返回该数据
            x = dequeue();
            c = count.getAndDecrement();//队列大小减1
            if (c > 1)
                notEmpty.signal();//还有数据就唤醒后续的消费线程
        } finally {
            takeLock.unlock();
        }
        //满足条件，唤醒条件对象上等待队列中的添加线程
        if (c == capacity)
            signalNotFull();
        return x;
    }

LinkedBlockingDeque是一个双向的阻塞的无边界的队列相比于LinkedBlockingDeque

可以任意的选择向链表的的头部添加数据还是向链表的尾部添加数据。当然也可以选择取数据的方向、

public interface BlockingDeque<E> extends BlockingQueue<E>, Deque<E> {

    void addFirst(E e);

    void addLast(E e);

    boolean offerFirst(E e);

    boolean offerLast(E e);

    。。。代码省略。。。
 }

    public E takeFirst() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E x;
           // 如果获取到的头结点中的数据为空 那么消费数据的线程挂起 等待
            while ( (x = unlinkFirst()) == null)
                notEmpty.await();
            return x;
        } finally {
            lock.unlock();
        }
    }


  private E unlinkFirst() {
      
        Node<E> f = first;
        if (f == null)
            return null;
        Node<E> n = f.next;
        E item = f.item;
        f.item = null;
      //头结点的下一结点赋值给自己 方便GC
        f.next = f; // help GC
  // 将头结点的的下一个结点值变成头结点
        first = n;
        if (n == null)
            last = null;
        else
          //操作成功将头结点的的前面前结点设置为null
            n.prev = null;
     // 队列总数减一
        --count;
//  入队操作唤醒
        notFull.signal();
 // 并返回头结点的内容
        return item;
    }