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

/**
 * @author csp
 * @date 2021-05-03
 */
public interface BrokingQueue<T> {
    /**
     * 插入数据
     */
    void put(T element) throws InterruptedException;
    /**
     * 获取数据
     */
    T take() throws InterruptedException;
}

/**
 * @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();
    }
}