AQS源码探究_05 Conditon条件队列(手写一个入门的BrokingQueue)

ゞ浴缸里的玫瑰 2022-10-21 03:51 106阅读 0赞

## AQS源码探究\_05 Conditon条件队列(手写一个入门的BrokingQueue) ##

### 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();
        }
    }