快速掌握并发编程---深入学习Condition

import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class Producer implements Runnable{

    private Queue<String> msg;
    private int maxSize;

    Lock lock;
    Condition condition;

    public Producer(Queue<String> msg, int maxSize, Lock lock, Condition condition) {
        this.msg = msg;
        this.maxSize = maxSize;
        this.lock = lock;
        this.condition = condition;
    }

    @Override
    public void run() {
        int i=0;
        while(true){
            i++;
            lock.lock();
                //队列中消息满了，此时生产者不能再生产了，因为装不下了
                //所以生产者就开始等待状态
                while(msg.size()==maxSize){
                    System.out.println("生产者队列满了，先等待");
                    try {
                        condition.await(); //阻塞线程并释放锁
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("生产消息："+i);
                msg.add("生产者的消息内容"+i);
                condition.signal(); //唤醒阻塞状态下的线程
            lock.unlock();
        }
    }
}

import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class Consumer implements Runnable{
    private Queue<String> msg;
    private int maxSize;

    Lock lock;
    Condition condition;

    public Consumer(Queue<String> msg, int maxSize, Lock lock, Condition condition) {
        this.msg = msg;
        this.maxSize = maxSize;
        this.lock = lock;
        this.condition = condition;
    }

    @Override
    public void run() {
        int i=0;
        while(true){
            i++;
            lock.lock(); //synchronized
            //消费者进来的时候需要判断是有可用的消息，
            //没有可用的消息就等待状态
            while(msg.isEmpty()){
                System.out.println("消费者队列空了，先等待");
                try {
                    condition.await(); //阻塞线程并释放锁   wait
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("消费消息："+msg.remove());
            condition.signal(); //唤醒阻塞状态下的线程
            lock.unlock();
        }
    }
}

public class TestCondition {
    public static void main( String[] args ){
        Queue<String> queue=new LinkedList<>();
        Lock lock=new ReentrantLock(); //重入锁
        Condition condition=lock.newCondition();
        int maxSize=5;

        Producer producer=new Producer(queue,maxSize,lock,condition);
        Consumer consumer=new Consumer(queue,maxSize,lock,condition);

        Thread t1=new Thread(producer);
        Thread t2=new Thread(consumer);
        t1.start();
        t2.start();

    }
}

void await() throws InterruptedException;

public final void await() throws InterruptedException {
    //表示 await 允许被中断
    if (Thread.interrupted()) throw new InterruptedException();
    //创建一个新的节点，节点状态为 condition，采用的数据结构仍然是单向链表
     Node node = addConditionWaiter();
    //释放当前的锁，得到锁的状态，并唤醒 AQS 队列中的一个线程
    //不管重入几次，都把state释放为0
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    //如果当前节点没有在同步队列上，即还没有被 signal，则将当前线程阻塞
    while (!isOnSyncQueue(node)) {
            ////通过 park 挂起当前线程
            LockSupport.park(this);
            // 当这个线程醒来,会尝试拿锁, 当 acquireQueued 返回 false 就是拿到锁了.
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
        }
    // 当这个线程醒来,会尝试拿锁, 当 acquireQueued 返回 false 就是拿到锁了.
    // interruptMode != THROW_IE -> 表示这个线程没有成功将 node 入队,但 signal 执行了 enq 方法让其入队了.
    // 将这个变量设置成 REINTERRUPT
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
    // 如果 node 的下一个等待者不是 null, 则进行清理,清理 Condition 队列上的节点. 
    // 如果是 null ,就没有什么好清理的了.
    if (node.nextWaiter != null) {
        //清理掉状态为cancelled状态的 
        nlinkCancelledWaiters();
    }
    // 如果线程被中断了,需要抛出异常.或者什么都不做
    if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
}

        /**
         * Adds a new waiter to wait queue.
         * @return its new wait node
         */
        private Node addConditionWaiter() {
            Node t = lastWaiter;
            // If lastWaiter is cancelled, clean out.
            //如果lastWaiter为canclled状态，则把他从链表中清理出去
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            //构建一个 Node，waitStatus=CONDITION。这里的链表是一个单向的，
            //所以相比 AQS 里的双向队来说简单了很多
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null){
                firstWaiter = node;
            } else{
                t.nextWaiter = node;
            }
            lastWaiter = node;
            return node;
        }

final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            //获取AQS中state值
            int savedState = getState();
            //释放锁并且唤醒下一个同步队列中的线程
            //注意这里处理的是同步队列
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }

    final boolean isOnSyncQueue(Node node) {
        //如果当前节点状态是CONDITION或node.prev是null，则证明当前节点在等待队列上而不是同步队列上。
        //之所以可以用node.prev来判断，是因为一个节点如果要加入同步队列，在加入前就会设置好prev字段。
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        //如果node.next不为null，则一定在同步队列上，
        //因为node.next是在节点加入同步队列后设置的
        if (node.next != null)
            return true;
        //前面的两个判断没有返回的话，就
        //从同步队列队尾遍历一个一个看是不是当前节点。
        return findNodeFromTail(node);
    } 
   //这个方法就相当简单了，就是从同步队列队尾遍历一个一个看是不是当前节点。
    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()){
          //如果同步状态不是被当前线程独占，直接抛出异常。从这里也能看出来，Condition只能配合独占类同步组件使用。
          throw new IllegalMonitorStateException();
       }
       // 拿到 Condition 队列上第一个节点
       Node first = firstWaiter;
       if (first != null){
           //通知等待队列队首的节点。
           doSignal(first);
        }
 }       

private void doSignal(Node first) {
      do {
             //从 Condition 队列中删除 first 节点
             if ( (firstWaiter = first.nextWaiter) == null){
                    // 将 next 节点设置成 null
                    lastWaiter = null;
              }
              first.nextWaiter = null;
         //transferForSignal方法尝试唤醒当前节点，如果唤醒失败，则继续尝试唤醒当前节点的后继节点。
        } while (!transferForSignal(first) &&(first = firstWaiter) != null);
}

final boolean transferForSignal(Node node) {
         //更新节点的状态为 0，如果更新失败，只有一种可能就是节点被 CANCELLED 了
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        //调用 enq，把当前节点添加到AQS 队列。并且返回返回按当前节点的上一个节点，也就是原tail 节点
        Node p = enq(node);
        int ws = p.waitStatus;
        // 如果上一个节点的状态被取消了, 或者尝试设置上一个节点的状态为 SIGNAL 
       //失败了(SIGNAL 表示: 他的 next 节点需要停止阻塞)
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            // 唤醒节点上的线程
            LockSupport.unpark(node.thread);
        //如果 node 的 prev 节点已经是signal 状态，那么被阻塞的 ThreadA 的唤醒工作由 AQS 队列来完成
        return true;
}