Java中的BlockingQueue阻塞队列原理以及实现详解

public interface BlockingQueue<E> extends Queue<E>

BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(BOUND);

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(); //条件对象
}

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock; //生产者加锁
    lock.lockInterruptibly();
    try {
        while (count == items.length) //如果队列满了，则会阻塞
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock(); //释放锁
    }
}

public final void await() throws InterruptedException {
    //线程是否被中断，如果被中断，直接抛异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //条件等待队列，会构建一个新的队列
    Node node = addConditionWaiter();
    //释放锁，并对对应的结点进行唤醒操作
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    //判断当前结点是在条件队列里面还是在同步队列里面
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

private E dequeue() {
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    //队列中有空位，通知生产者生产产品
    notFull.signal();
    return x;
}

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }