JUC第八讲：Condition源码分析

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
 * Condition接口与示例
 * @param <T>
 */
public class BoundedQueue<T> {
    // 线程池
    private static ExecutorService THREAD_POOL = new ThreadPoolExecutor(2, 4, 60, TimeUnit.MINUTES,
            new LinkedBlockingQueue<Runnable>(1000), Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());
  // 对象数组
    private Object[] items; 
    // 添加的下标，删除的下标和数组当前数量
    private int addIndex, removeIndex, count;
    private Lock lock = new ReentrantLock(); // 定义一个可重入锁
    private Condition notEmpty = lock.newCondition(); // 添加一个Condition
    private Condition notFull = lock.newCondition(); // 添加一个Condition
    public BoundedQueue(int size) {
        items = new Object[size];
    }
    /**
     * 添加一个元素，如果数组满，则添加线程进入等待状态，直到有"空位"
     * @param t
     * @throws InterruptedException
     */
    public void add(T t) throws InterruptedException {
        lock.lock(); // 获取锁
        try {
            // 场景1：如果数组满了，notFull进入等待
            while (count == items.length) { 
                System.out.println("items满了，add方法进入等待.");
                notFull.await(); // 等待remove方法里的notFull.signal()
            }
      // 场景2：item添加对象
            items[addIndex] = t;
            if (++addIndex == items.length) // 调整数组索引，避免越界
                addIndex = 0;
            ++count; // count+1，代表添加了一个对象
            notEmpty.signal(); // 走到这里，数组里至少有1个对象，必不为空，因此唤醒notEmpty
        } finally {
            System.out.println("add: " + t);
            lock.unlock(); // 释放锁
        }
    }
    /**
     * 由头部删除一个元素，如果数组空，则删除线程进入等待状态，
     * 直到有新添加元素（注意这里并没有真的删除元素，只是把count-1当作是删除）
     * @return
     * @throws InterruptedException
     */
    @SuppressWarnings("unchecked")
    public T remove() throws InterruptedException {
        lock.lock(); // 获取锁
        try {
            //场景1：如果数组为空，notEmpty进入等待
            while (count == 0) { 
                System.out.println("items为空，remove方法进入等待.");
                notEmpty.await(); // 等待add方法里的notEmpty.signal()
            }
      //场景2：如果数组非空，item移除对象
            Object x = items[removeIndex]; // item移除对象（假移除）
            if (++removeIndex == items.length) // 调整数组索引，避免越界
                removeIndex = 0;
            --count; // count-1，代表移除了一个对象
            notFull.signal(); // 走到这里，数组里至少有1个空位，必不为满，因此唤醒notFull
            return (T) x;
        } finally {
            System.out.println("remove");
            lock.unlock(); // 释放锁
        }
    }
    public static void main(String args[]) throws InterruptedException {
        int count = 3; // 可以加大数组的size来看更多的过程
        BoundedQueue<Integer> bq = new BoundedQueue<Integer>(count);
        // 开启一个线程执行添加操作
        THREAD_POOL.submit(new Callable<Object>() {
            public Object call() throws InterruptedException {
                for (int i = 0; i < count * 2; i++) {
                    bq.add(i);
                    Thread.sleep(200); // 通过睡眠来制造添加和移除的速度差
                }
                return null;
            }
        });
        // 开启一个线程执行移除操作
        THREAD_POOL.submit(new Callable<Object>() {
            public Object call() throws InterruptedException {
                Thread.sleep(1000);
                for (int i = 0; i < count * 2; i++) {
                    bq.remove();
                    Thread.sleep(50); // 通过睡眠来制造添加和移除的速度差
                }
                return null;
            }
        });
    }
}

private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. */
private transient Node firstWaiter; // 条件队列的头节点
/** Last node of condition queue. */    
private transient Node lastWaiter;  // 条件队列的尾节点
/**
 * Creates a new {@code ConditionObject} instance.
 */
public ConditionObject() { }

// 阻塞当前线程，直接被唤醒或被中断
public final void await() throws InterruptedException {
    // 如果当前线程被中断过，则抛出中断异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 添加一个waitStatus为CONDITION的节点到条件队列尾部
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);    // 释放操作。我们知道只有在拥有锁（acquire成功）的时候才能调用await()方法，因此，调用await()方法的线程的节点必然是同步队列的头节点。所以，当调用await()方法时，相当于同步队列的首节点（获取了锁的节点）移动到Condition的条件队列中。
    // 0为正常，被中断值为THROW_IE或REINTERRUPT
    int interruptMode = 0;
    // isOnSyncQueue：判断node是否在同步队列（注意和条件队列区分。调用signal方法会将节点从条件队列移动到同步队列，因此这边就可以跳出while循环）
    while (!isOnSyncQueue(node)) {
        // node如果不在同步队列则进行park（阻塞当前线程）
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)    // 检查线程被唤醒是否是因为被中断，如果是则跳出循环，否则会进行下一次循环，因为被唤醒前提是进入同步队列，所以下一次循环也必然会跳出循环
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)   // acquireQueued返回true代表被中断过，如果中断模式不是THROW_IE，则必然为REINTERRUPT（见上面的checkInterruptWhileWaiting方法）
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();   // 移除waitStatus为CANCELLED的节点
    if (interruptMode != 0) // 如果跳出while循环是因为被中断
        reportInterruptAfterWait(interruptMode);    // 则根据interruptMode，选择抛出InterruptedException 或 重新中断当前线程
}

// 添加一个waitStatus为CONDITION的节点到条件队列尾部
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();   // 移除waitStatus不为 CONDITION的节点（条件队列里的节点 waitStatus 都为 CONDITION）
        t = lastWaiter; // 将t赋值为移除了waitStatus不为CONDITION后的尾节点（上面进行了移除操作，因此尾节点可能会发生变化）
    }
    // 以当前线程新建一个waitStatus为CONDITION的节点
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // t为空，代表条件队列为空
    if (t == null)
        // 将头节点赋值为node
        firstWaiter = node;
    else
        // 否则，队列不为空。将t（原尾节点）的后继节点赋值为node
        t.nextWaiter = node;
    lastWaiter = node;  // 将node赋值给尾节点，即将node放到条件队列的尾部。这里没有用CAS来保证原子性，原因在于调用await()方法的线程必定是获取了锁的线程，也就是说该过程是由锁来保证线程安全的
    return node;
}

//　从条件队列移除所有waitStatus不为CONDITION的节点
private void unlinkCancelledWaiters() {
    // t赋值为条件队列的尾节点  
    Node t = firstWaiter; 
    Node trail = null;
    while (t != null) {
        // 向下遍历
        Node next = t.nextWaiter;
        // 如果t的waitStatus不为CONDITION
        if (t.waitStatus != Node.CONDITION) {
            // 断开t与t后继节点的关联
            t.nextWaiter = null;
            if (trail == null)  // 如果trail为null，则将firstWaiter赋值为next节点，此时还没有遍历到waitStatus为CONDITION的节点，因此直接移动firstWaiter的指针即可移除前面的节点
                firstWaiter = next; 
            else
                trail.nextWaiter = next;    // 否则将trail的后继节点设为next节点。此时，trail节点到next节点中的所有节点被移除（包括t节点，但可能不止t节点。因为，trail始终指向遍历过的最后一个waitStatus为CONDITION，因此只需要将trail的后继节点设置为next，即可将trail之后到next之前的所有节点移除）
            if (next == null)
                lastWaiter = trail;
        }
        else
            trail = t;  // 如果t的waitStatus为CONDITION，则将trail赋值为t，trail始终指向遍历过的最后一个waitStatus为CONDITION
        t = next;   // t指向下一个节点
    }
}

// 释放锁
final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        // 当前的同步状态
        int savedState = getState();
        // 独占模式下release（一般指释放锁）
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            // 如果release失败则将该节点的waitStatus设置为CANCELLED
            node.waitStatus = Node.CANCELLED;
    }
}

// 判断node是否再同步队列中
final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null) // 如果waitStatus为CONDITION 或 node没有前驱节点，则必然不在同步队列，直接返回false
        return false;
    if (node.next != null) // If has successor, it must be on queue 如果有后继节点，必然是在同步队列中，返回true
        return true;
    /*
     * node.prev can be non-null, but not yet on queue because
     * the CAS to place it on queue can fail. So we have to
     * traverse from tail to make sure it actually made it.  It
     * will always be near the tail in calls to this method, and
     * unless the CAS failed (which is unlikely), it will be
     * there, so we hardly ever traverse much.
     */
    return findNodeFromTail(node);  // 返回node是否为同步队列节点，如果是返回true，否则返回false
}

// 从同步队列的尾节点开始向前遍历，如果node为同步队列节点则返回true，否则返回false
private boolean findNodeFromTail(Node node) {
    Node t = tail;
    for (;;) {
        if (t == node)
            return true;
        if (t == null)
            return false;
        t = t.prev;
    }
}

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)  // 将first节点赋值为first节点的后继节点（相当于移除first节点），如果first节点的后继节点为空，则将lastWaiter赋值为null
            lastWaiter = null;
        // 断开first节点对first节点后继节点的关联
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&   // transferForSignal：将first节点从条件队列移动到同步队列
             (first = firstWaiter) != null);    // 如果transferForSignal失败，并且first节点不为null，则向下遍历条件队列的节点，直到节点成功移动到同步队列 或者 firstWaiter为null
}

//　将node节点从条件队列移动到同步队列，如果成功则返回true。
final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))  // 如果不能更改节点的waitStatus，则表示该节点已被取消，返回false
        return false;
    /*
     * Splice onto queue and try to set waitStatus of predecessor to
     * indicate that thread is (probably) waiting. If cancelled or
     * attempt to set waitStatus fails, wake up to resync (in which
     * case the waitStatus can be transiently and harmlessly wrong).
     */
    Node p = enq(node); // 否则，调用enq方法将node添加到同步队列，注意：enq方法返回的节点是node的前驱节点
    // 将ws赋值为node前驱节点的等待状态
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) // 如果node前驱节点的状态为CANCELLED（ws>0） 或 使用CAS将waitStatus修改成SIGNAL失败，则代表node的前驱节点无法来唤醒node节点，因此直接调用LockSupport.unpark方法唤醒node节点
        LockSupport.unpark(node.thread);
    return true;
}

public final void signalAll() {
    // 检查当前线程是否为独占模式同步器的所有者
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        // 唤醒条件队列的所有节点
        doSignalAll(first);
}

// 将条件队列的所有节点移到同步队列
private void doSignalAll(Node first) {
    // 因为要移除条件队列的所有节点到同步队列，因此这边直接将firstWaiter和lastWaiter赋值为null
    lastWaiter = firstWaiter = null;
    do {
        // next赋值为first节点的后继节点
        Node next = first.nextWaiter;
        // 断开first节点对first节点后继节点的关联
        first.nextWaiter = null;
        // transferForSignal：将first节点从条件队列移动到同步队列
        transferForSignal(first);
        // first赋值为next节点
        first = next;
    } while (first != null);    // 循环遍历，将条件队列的所有节点移动到同步队列
}