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

Posted 2021-05-21 兴趣使然の草帽路飞

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