1. Condition队列介绍
AQS中还有另一个非常重要的内部类ConditionObject，它实现了Condition接口，主要用于实现条件锁。
ConditionObject中也维护了一个队列，这个队列主要用于等待条件的成立，当条件成立时，其它线程将signal这个队列中的元素，将其移动到AQS的队列中，等待占有锁的线程释放锁后被唤醒。
Condition典型的运用场景是在BlockingQueue中的实现，当队列为空时，获取元素的线程阻塞在notEmpty条件上，一旦队列中添加了一个元素，将通知notEmpty条件，将其队列中的元素移动到AQS队列中等待被唤醒。
2. 手写一个入门的BrokingQueue
2.1 自定义BrokingQueue接口
/** * @author csp * @date 2021-05-03 */ public interface BrokingQueue<T> { /** * 插入数据 */ void put(T element) throws InterruptedException; /** * 获取数据 */ T take() throws InterruptedException; }
2.2 自定义MiniArrayBrokingQueue类
/** * @author csp * @date 2021-05-03 */ public class MiniArrayBrokingQueue implements BrokingQueue { /** * ReentrantLock锁：用于线程并发控制 */ private Lock lock = new ReentrantLock(); /** * 生产者线程生产数据时，会先检查当前queues是否已经满了，如果已满，则需要令当前生产者线程调用notFull.await() * 进入到notFull条件队列挂起，等待消费者线程消费一个数据时才可以唤醒~ */ private Condition notFull = lock.newCondition(); /** * 消费者线程消费数据时，会先检查当前queues中是否有数据，如果没有数据，则需要令当前消费者线程调用notEmpty.await() * 进入到notEmpty条件队列挂起，等待生产者生产出数据时才能唤醒！ */ private Condition notEmpty = lock.newCondition(); /** * 存储元素的数组队列 */ private Object[] queues; /** * 数组队列的长度 */ private int size; /** * count：当前队列中，可以被消费的数据量 * putptr：记录生产者存放数据的下一次位置，每个生产者生产完一个数据后，会将 putptr++ * takeptr：记录消费者消费数据的下一个位置，每个消费者消费完一个数据后，会将takeptr++ */ private int count, putptr, takeptr; public MiniArrayBrokingQueue(int size) { this.size = size; this.queues = new Object[size]; } /** * 插入数据 * * @param element */ @Override public void put(Object element) throws InterruptedException { // 获取锁 lock.lock(); try { // 判断当前queues数组队列是否已满 if (count == this.size) { // 如果已满，则进入到notFull条件队列挂起，等待消费者线程消费一个数据时才可以唤醒~ notFull.await(); } // 执行到这里，说明数组队列queues未满，可以向队列中存放数据~ this.queues[putptr] = element; putptr++; // 新存放数据后，判断是否达到了数组队列的最大值 if (putptr == this.size) { // putptr恢复到0，为什么可以归零呢？ // 因为：多线程执行条件下，总会有消费者线程在不断消费数据，即使当前生产者线程已经将putptr移动到size的位置， // 但是扔有可能有消费者线程将queues数组队列size位置之前的数据给消费掉了，那么此时queues其实并不算已经满了， // 那么将putptr恢复到0，就是为了让其不断去寻找queues数组中的空位，并用来存放element putptr = 0; } // 当前队列中，可以被消费的数据量++ count++; // 当向队列中成功放入一个元素之后，需要做什么呢？ // 需要给notEmpty一个唤醒信号，告诉消费者去消费 notEmpty.signal(); } finally { // 释放锁 lock.unlock(); } } /** * 获取数据 * * @return */ @Override public Object take() throws InterruptedException { // 获取锁 lock.lock(); try { // 判断当前queues数组队列是否是空的 if (count == 0) { // 如果queues是空的，则进入到notEmpty队列进行挂起，等待生产者线程生产数据时才可以唤醒~ notEmpty.await(); } // 执行到这里，说明队列中有数据可以被消费了~ Object element = this.queues[takeptr]; takeptr++; // 新消费数据后，判断是否达到了数组队列的最大值 if (takeptr == this.size) { // takeptr恢复到0，为什么可以归零呢？ // 因为：多线程执行条件下，总会有生产者线程在不断生产数据，即使当前消费者线程已经将takeptr移动到size的位置， // 但是扔有可能有生产者线程向queues数组队列size之前的位置put数据，那么此时queues其实并不算空了， // 那么将takeptr恢复到0，就是为了让其不断去寻找queues数组中能被消费的数据 takeptr = 0; } // 当前队列中，可以被消费的数据量-- count--; // 当从队列中成功消费一个元素之后，需要做什么呢？ // 需要给notFull一个唤醒信号，告诉生产者去生产 notFull.signal(); // 返回消费的element return element; } finally { // 释放锁 lock.unlock(); } } // 测试程序： public static void main(String[] args) { BrokingQueue<Integer> queue = new MiniArrayBrokingQueue(10); Thread producer = new Thread(() -> { int i = 0; while (true) { i++; if (i == 10) { i = 0; } try { System.out.println("生产者线程生产数据：" + i); queue.put(Integer.valueOf(i)); TimeUnit.MILLISECONDS.sleep(200); } catch (InterruptedException e) { } } }); producer.start(); Thread consumer = new Thread(() -> { while (true) { try { Integer result = queue.take(); System.out.println("消费者线程消费数据：" + result); TimeUnit.MILLISECONDS.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); } } }); consumer.start(); } }
