深入理解队列：LinkedBlockingQueue 源码深度解析

static class Node<E> {
    E item;
    /**
     * One of:
     * - the real successor Node
     * - this Node, meaning the successor is head.next
     * - null, meaning there is no successor (this is the last node)
     */
    Node<E> next;
    Node(E x) { item = x; }
}

// 队列的容量，默认为Integer.MAX_VALUE
private final int capacity;
// 队列的元素数量，初始值为0
private final AtomicInteger count = new AtomicInteger();
// 链表（队列）的头节点
transient Node<E> head;
// 链表（队列）的尾节点
private transient Node<E> last;
// 这个可重入锁用于take，poll等获取数据的方法，即用于出队的锁
private final ReentrantLock takeLock = new ReentrantLock();
// 用于take, poll等获取数据的方法的阻塞条件
private final Condition notEmpty = takeLock.newCondition();
// 这个可重入锁用于put，offer等等添加数据的方法
// 设计两把锁的目的，主要为了take和put可以同时进行，互不影响
private final ReentrantLock putLock = new ReentrantLock();
// 用于put，offer等添加数据的方法的阻塞条件
private final Condition notFull = putLock.newCondition();
// 无参构造方法，队列默认容量Integer.MAX_VALUE
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}
// 指定链表容量大小，链表头尾相等，节点值（item）为null
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);
    // 获取到putLock锁
    final ReentrantLock putLock = this.putLock;
    // 加锁，保证线程安全
    putLock.lock(); // Never contended, but necessary for visibility
    try {
        int n = 0;
        for (E e : c) {
        	// 节点内的值不能为null
            if (e == null)
                throw new NullPointerException();
            // 判断队列是否满了
            if (n == capacity)
                throw new IllegalStateException("Queue full");
            // 将Node节点添加到队列的尾部，last = last.next = new Node<E>(e);
            enqueue(new Node<E>(e));
            ++n;
        }
        // 原子类设置Node节点个数，线程安全
        count.set(n);
    } finally {
    	// 解锁
        putLock.unlock();
    }
}

public boolean add(E e) {
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {
	// 如果存入的值为null，直接抛出空指针异常
    if (e == null) throw new NullPointerException();
    long nanos = unit.toNanos(timeout);
    // 预先设置c为 -1，约定负数为入队失败
    int c = -1;
    // 获取入队锁
    final ReentrantLock putLock = this.putLock;
    // 获取队列元素个数
    final AtomicInteger count = this.count;
    // 加锁
    putLock.lockInterruptibly();
    try {
        while (count.get() == capacity) {
        	// 如果超时时间过了队列仍然是满的话就直接返回false
            if (nanos <= 0)
                return false;
            // 否则调用awaitNanos等待，超时会返回<= 0L
            nanos = notFull.awaitNanos(nanos);
        }
        // 如果上述没有阻塞，也就是队列没有满，那么这里直接入队
        enqueue(new Node<E>(e));
        // 队列元素个数+1，此时c为元素入队前的个数，也就是比当前队列元素个数少1
        c = count.getAndIncrement();
        if (c + 1 < capacity)
        	// 如果添加数据后还队列还没有满，则继续调用notFull的signal方法唤醒其他等待在入队的线程
            notFull.signal();
    } finally {
    	// 解锁
        putLock.unlock();
    }
    // c==0说明队列中有一个元素了，那么就需要唤醒其他正在等待出队的线程
    // 这一点可能不好理解，c = count.getAndIncrement();理解了就差不多
    if (c == 0)
        signalNotEmpty();
    return true;
}

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -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) {
        	// put方法这里没有设置阻塞超时时间
            notFull.await();
        }
        enqueue(node);
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    E x = null;
    int c = -1;
    long nanos = unit.toNanos(timeout);
    final AtomicInteger count = this.count;
    // 获取take锁
    final ReentrantLock takeLock = this.takeLock;
    // 上锁
    takeLock.lockInterruptibly();
    try {
        while (count.get() == 0) {
        	// 如果队列为空，那么将进入等待，且如果阻塞时间过期，那么将返回null
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        // 在超时时间内返回，则调用dequeue获取队列中的数据
        x = dequeue();
        c = count.getAndDecrement();
        // 如果c > 1，说明队列中还有节点元素，那么继续唤醒其他出队线程
        if (c > 1)
            notEmpty.signal();
    } finally {
    	// 解锁
        takeLock.unlock();
    }
    // 如果c == capacity就是说队列中有一个空位，唤醒入队线程
    if (c == capacity)
        signalNotFull();
    return x;
}

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) {
        	// take方法这里没有设置阻塞超时时间
            notEmpty.await();
        }
        x = dequeue();
        c = count.getAndDecrement();
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();
    return x;
}