LinkedBlockingQueue

LinkedBlockingQueue是一个基于单链表实现的阻塞队列，通过构造器参数可以设定队列容量的大小，如果没有设定默认为Integer.MAX_VALUE,队列中元素顺序遵循FIFO先进先出的规则。

基本属性

/**
     * Linked list node class
     */
    static class Node<E> {
        E item;
        /**
         * 当前节点的后继节点
         */
        Node<E> next;

        Node(E x) { item = x; }
    }

    /** The capacity bound, or Integer.MAX_VALUE if none */
    private final int capacity;//队列容量大小

    /** Current number of elements */
    private final AtomicInteger count = new AtomicInteger();//元素数量

    /**
     * Head of linked list.
     * Invariant: head.item == null
     */
    transient Node<E> head;//链表头节点

    /**
     * Tail of linked list.
     * Invariant: last.next == null
     */
    private transient Node<E> last;//链表尾结点

    /** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();

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

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();
    public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

    /**
     * Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity.
     *
     * @param capacity the capacity of this queue
     * @throws IllegalArgumentException if {@code capacity} is not greater
     *         than zero
     */
    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }

入队

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;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        /*
         * Note that count is used in wait guard even though it is
         * not protected by lock. This works because count can
         * only decrease at this point (all other puts are shut
         * out by lock), and we (or some other waiting put) are
         * signalled if it ever changes from capacity. Similarly
         * for all other uses of count in other wait guards.
         */
        while (count.get() == capacity) {
            notFull.await();
        }
        enqueue(node);
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}

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

出队

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();
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();
    return x;
}
/**
 * 移除队头元素 head节点不存储数据 
 * head节点移除后将后继节点设置为head
 * 并将其item置空 并返回
 */
private E dequeue() {
    // assert takeLock.isHeldByCurrentThread();
    // assert head.item == null;
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

总结

相比于ArrayBlockingQueue
LinkedBlockingQueue出队和入队操作分别使用不同的锁 避免了相互阻塞的现象;ArrayBlockingQueue出入队使用的是同一把锁 出入队会互相阻塞 效率低下
两者都是有界队列，若队列长度固定 出队速度小于入队速度 会导致大量线程阻塞
LinkedBlockingQueue初始化时如果不指定容量，默认为Integer.MAX_VALUE，若出队速度小于入队速度 会导致队列特别长 占用大量内存