LinkedBlockingQueue<Integer> blockingQue = new LinkedBlockingQueue<>();
blockingQue.add(1);
blockingQue.remove(1);
blockingQue.remove(); // NoSuchElementException
blockingQue.element(); // NoSuchElementException

LinkedBlockingQueue<Integer> blockingQue = new LinkedBlockingQueue<>();
blockingQue.offer(1);
blockingQue.poll();
Integer peek = blockingQue.peek(); // 返回null

LinkedBlockingQueue<Integer> blockingQue = new LinkedBlockingQueue<>();
try {
    blockingQue.put(1);
    blockingQue.take();
} catch (InterruptedException e) {
    // 線程被中斷
    e.printStackTrace();
}

LinkedBlockingQueue<Integer> blockingQue = new LinkedBlockingQueue<>();
try {
    blockingQue.offer(1, 100, TimeUnit.MILLISECONDS);
    blockingQue.poll(100, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
    e.printStackTrace();
}

static class Node<E> {
    E item;
 
    Node<E> next;
 
    Node(E x) { item = x; }
}

    // 入隊
    private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        last = last.next = node; // last.next = node; last = node;
    }

 
    // 出隊
    private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == null;
        Node<E> h = head;
        Node<E> first = h.next;
        // 形成引用鏈閉環(huán)，JVM根據(jù)可達性分析時，GC root的引用鏈與該對象之間不可達，進行GC
        h.next = h; // help GC
        head = first;
        E x = first.item;
        first.item = null;
        return x;
    }

/** 容量, 初始化不設置時默認為Integer.MAX_VALUE*/
private final int capacity;
 
/** 當前隊列內(nèi)的元素數(shù)量 */
private final AtomicInteger count = new AtomicInteger();
 
/**
 * 隊首
 * 不變量: head.item == null
 */
transient Node<E> head;
 
/**
 * 隊尾
 * 不變量: last.next == null
 */
private transient Node<E> last;
 
/** 出隊操作公用鎖 */
private final ReentrantLock takeLock = new ReentrantLock();
 
/** 用于出隊操作的阻塞和喚醒 出隊的話，只需要考慮隊列是否為空 */
private final Condition notEmpty = takeLock.newCondition(); 
 
/** 入隊操作公用鎖 */
private final ReentrantLock putLock = new ReentrantLock();
 
/** 用于入隊操作的阻塞和喚醒 入隊只需要考慮隊列空間是否足夠*/
private final Condition notFull = putLock.newCondition();

    public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }
 
    // 自定義容量
    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }
 
    // 初始化時，批量添加集合中的元素
    public LinkedBlockingQueue(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);
        final ReentrantLock putLock = this.putLock;
        putLock.lock(); // Never contended, but necessary for visibility
        try {
            int n = 0;
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (n == capacity)
                    throw new IllegalStateException("Queue full");
                enqueue(new Node<E>(e));
                ++n;
            }
            count.set(n);
        } finally {
            putLock.unlock();
        }
    }

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    // Note: convention in all put/take/etc is to preset local var
    // holding count negative to indicate failure unless set.
    int c = -1; // -1代表操作異常
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    // 如果線程沒有被標記為中斷，則獲取鎖
    putLock.lockInterruptibly();
    try {
        while (count.get() == capacity) {
            // 這里是線程在執(zhí)行put操作時唯一一個執(zhí)行過程中釋放鎖的地方
            notFull.await(); // 容量已滿，等待被消費后喚醒
        }
        // 添加元素，更新容量
        enqueue(node);
        c = count.getAndIncrement();
        // 隊列容量有余時，在這里再次喚醒一個其他的生產(chǎn)者線程（或者說消費者消費速度大于生產(chǎn)）
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        // 釋放鎖
        putLock.unlock();
    }
    // 喚醒一個消費者
    if (c == 0)
        signalNotEmpty();
}

        while (count.get() == capacity) {
            notFull.await(); // 容量已滿，等待被消費后喚醒
        }

// put            
while (count.get() == capacity) {
     notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
 
 
 
// take
while (count.get() == 0) {
    notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
    notEmpty.signal();

// 喚醒一個等待中的消費者
private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

c = count.getAndIncrement();

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();
        // 生產(chǎn)速度大于消費速度，喚醒一個其他消費者進行消費
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    // 消費之前隊列已滿，消費后喚醒一個生產(chǎn)者
    if (c == capacity)
        signalNotFull();
    return x;
}
 
// 喚醒生產(chǎn)者
/**
 * Signals a waiting put. Called only from take/poll.
 */
private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}