Java并发LinkedBlockingQueue源码分析

/**
 * 链表中的节点类
 */
static class Node<E> {
    //节点中的元素
    E item;
    //下一个节点
    Node<E> next;
    Node(E x) { item = x; }
}
/** 链表队列的容量大小,如果没有指定则使用Integer最大值 */
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();
/** 添加节点到链表队列中防止多个线程同时操作所产生数据安全问题时所加的锁 */
private final ReentrantLock putLock = new ReentrantLock();
/** Wait queue for waiting puts */
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();
    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();
    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) {
            notFull.await();
        }
        //将新节点添加到链表的尾部并设置为尾节点
        enqueue(node);
        //获取没有添加当前节点时链表中的节点数量
        //并更新链表中的节点数量
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            //唤醒等待添加节点的线程
            //可能当前线程在等待队列中等待的时候
            //有新的线程要执行添加节点的操作
            //但是链表的容量已经到达最大,所以新的线程也会进行等待
            //当前线程被唤醒了并且链表的容量没有到达最大则尝试去唤醒等待的线程
            notFull.signal();
    } finally {
        //释放锁
        putLock.unlock();
    }
    if (c == 0)
        //c等于0说明添加当前节点的时候链表中没有节点
        //可能有线程在获取节点,但是链表中没有节点
        //从而一直进行等待,当添加了节点的时候就需要唤醒获取节点的线程
        signalNotEmpty();
}

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;
    }
    //acquireQueued 指定节点中的线程被中断了或者被唤醒了则会尝试去获取锁
    //如果还未到指定节点中的线程获取锁的时候则会继续挂起
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null)
        //指定节点的线程已经获取到了锁并且节点关联的下一个节点不为空
        //此时就需要将已经获取到锁的节点从等待队列中移除
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        //指定节点还在等待队列中此时就需要继续等待
        return false;
    if (node.next != null)
        //指定节点已经不在等待队列中了
        return true;
    //从等待队列中的尾节点开始向头节点遍历,校验指定的节点是否在其中
    return findNodeFromTail(node);
}

public final void signal() {
    if (!isHeldExclusively())
        //加锁的线程不是当前线程则抛出异常
        throw new IllegalMonitorStateException();
    //头节点
    Node first = firstWaiter;
    if (first != null)
        //唤醒头节点
        doSignal(first);
}
/**
 * 唤醒等待队列中的头节点
 * 如果等待队列中的头节点被取消等待或已经被唤醒了
 * 此时就需要唤醒头节点的后续的一个节点
 * 直到成功的唤醒一个节点中的线程
 */
private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) && (first = firstWaiter) != null);
}
/**
* 将指定的节点添加到同步等待队列中
* 并根据前一个节点的等待状态来决定是否需要立刻唤醒指定节点
*/
final boolean transferForSignal(Node node) {
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        //更改节点状态失败说明该节点已经被唤醒了
        return false;
    //将要唤醒的节点添加到同步等待队列中
    //并返回前一个节点
    Node p = enq(node);
    //前一个节点的等待状态
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        //如果前一个节点的等待状态大于0则说明已经被取消加锁,此时就需要唤醒后续的节点,就是当前节点
        //前一个节点的等待状态不大于0但是更改前一个节点的等待状态时失败则说明前一个节点已经被唤醒了并更改了状态
        //此时就需要尝试将当前节点中的线程唤醒
        LockSupport.unpark(node.thread);
    return true;
}