JAVA并发编程系列(10)Condition条件队列-并发协作者

package lading.java.mutithread;

import cn.hutool.core.date.DateTime;

import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 模拟生产者-消费者，生产者生产了产品后消费者才能消费，消费者消费完产品后才能继续生产产品
 * 每次随机最多生产4件产品，消费者消费完才能继续生产
 */
public class Demo011Condition {
    //可消费的产品数量
    public static AtomicInteger productNum = new AtomicInteger(0);
    //可重入锁
    public static ReentrantLock lock = new ReentrantLock();
    //消费者信号
    public static Condition consumer = lock.newCondition();
    //生产者信号
    public static Condition producer = lock.newCondition();

    public static void main(String[] args) {
        new Thread(() -> {
            while (true) {
                lock.lock();
                try {
                    //如果有产品可以消费
                    while (productNum.get() > 0) {
                        //生产者线程阻塞等待,并释放锁
                        producer.await();
                    }
                    //随机生产最多4件产品，耗时3s
                    Thread.sleep(3000);
                    int total = new Random().nextInt(4) + 1;
                    for (int i = 0; i < total; i++) {
                        System.out.println(DateTime.now().toString("YYYY-MM-dd hh:mm:ss SSS") + " 【" + Thread.currentThread().getName() + "】新生产了产品:" + productNum.incrementAndGet());
                    }
                    //通知消费者来消费
                    System.out.println("生产者 --> 已生产" + total + "件产品，可以过来消费。");
                    consumer.signalAll();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }

        }, "生产者").start();

        new Thread(() -> {
            while (true) {
                lock.lock();
                try {
                    //没有可以消费，消费者进入阻塞等待，释放锁
                    while (productNum.get() == 0) {
                        consumer.await();
                    }
                    //消费者消费产品,耗时2s
                    Thread.sleep(2000);
                    int total = productNum.get();
                    while (productNum.get() != 0) {
                        System.out.println(DateTime.now().toString("YYYY-MM-dd hh:mm:ss SSS") + " 【" + Thread.currentThread().getName() + "】消费了产品:" + productNum.getAndDecrement());
                    }
                    //唤醒生产者继续生产
                    System.out.println("消费者 --> " + total + "件产品已清空，请继续生产。");
                    producer.signalAll();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }
        }, "消费者").start();
    }
}

        public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            //将自己加入到AQS的FIFO等待队列尾部，进入等待    
            Node node = addConditionWaiter();
            //释放锁资源
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            //检测本线程节点是否在AQS同步队列中，如果不在就继续检测，直到发现在AQS同步队列中
            while (!isOnSyncQueue(node)) {
                //如果在队列中，就挂起本线程
                LockSupport.park(this);
                //如果被中断，则跳出循环
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            //尝试获取锁-毕竟这期间有可能被唤醒了，以及是否发生过中断
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

        public final void signal() {
            //判断当前线程是否持有锁，只有持有锁才能唤醒其他线程，不持有锁，就抛异常
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignal(first);
        }
        //具体唤醒过程
        private void doSignal(Node first) {
            //尝试将头节点NODE唤醒，从等待队列中，挪到AQS队列，并通过CAS操作更新NODE状态
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }