AQS源码解析 4.Condition条件队列入门_手写BrokingQueue

AQS源码解析—Condition条件队列入门_手写BrokingQueue

Condition 条件队列介绍

• AQS 中还有另一个非常重要的内部类 ConditionObject，它实现了 Condition 接口，主要用于实现条件锁。
• ConditionObject 中也维护了一个队列，这个队列主要用于等待条件的成立，当条件成立时，其它线程将 signal 这个队列中的元素，将其移动到 AQS 的队列中，等待占有锁的线程释放锁后被唤醒。
• Condition 典型的运用场景是在 BlockingQueue 中的实现，当队列为空时，获取元素的线程阻塞在 notEmpty 条件上，一旦队列中添加了一个元素，将通知 notEmpty 条件，将其队列中的元素移动到 AQS 队列中等待被唤醒。

管程模型（Monitor）

• The Owner：The Owner 中只能有一个线程，就是持锁线程。
• Entry Set：理解为竞争队列，抢锁失败后会进入这个队列中阻塞，当 The Owner 线程释放锁后，Entry Set 中所有的线程一起去竞争锁。
• Wait Set：等待 (阻塞) 队列，如果线程在执行期间被 wait() 方式挂起，会释放锁 并进入 Wait Set 中等待，当满足条件时 等待队列中的线程则会被 notify() 随机唤醒，唤醒后该线程同样可以去竞争锁。

这里只是简单分析一下管程模型（Monitor）=> wait() / notify()，后续关于 synchronized 的文章会详细分析；

而 Condition 条件队列就是在 Java 语言层面实现了这样类似的功能 => await() / signal() 。

两者区别：

• Object => Monitor => Wait Set，一个 Object 只有一个 Monitor，同样也只有一个 Wait Set，notify() 只能随机唤醒 Wait Set 中的一个线程。
• Condition => Lock.newCondition，ReentrantLock 的条件变量比 synchronized 强大之处在于支持多个条件变量，每个 Condition 对象都包含一个等待队列，可以创建 多个 Condition 对象，从而调用某个 conditionxx.signal()，唤醒该条件队列中的第一个线程，更细粒度的划分。

但不管是执行 wait() / notify() 还是 await() / signal()，前提都需要获得锁。

手写一个入门的 BrokingQueue

• 通过手写阻塞队列的方式，来初步认识 AQS 中的 ConditionObject 对象，以及 Condition 的使用。
• 当队列中元素满了就阻塞生产者线程并唤醒消费者线程，当队列为空就阻塞消费者线程并唤醒生产者线程。

BrokingQueue 接口

/**
 * @author lyc
 * @date 2022-10-26
 */
public interface BrokingQueue<T> {
    /**
     * 插入数据
     */
    void put(T element) throws InterruptedException;

    /**
     * 获取数据
     */
    T take() throws InterruptedException;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

MiniArrayBrokingQueue 类

/**
 * @author lyc
 * @date 2022-10-26
 */
public class MiniArrayBrokingQueue implements BrokingQueue {

    /**
     * ReentrantLock锁：用于线程并发控制
     */
    private Lock lock = new ReentrantLock();

    /**
     * 条件队列1：notFull 存放生产者线程
     * 生产者线程生产数据时，会先检查当前queues是否已经满了，如果已满，则需要令当前生产者线程调用notFull.await()
     * 进入到notFull条件队列挂起，等待消费者线程消费数据时唤醒。
     */
    private Condition notFull = lock.newCondition();

    /**
     * 条件队列2：notEmpty 存放消费者线程
     * 消费者线程消费数据时，会先检查当前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;

    /**
     * 初始化构造方法
     * @param size：数组队列初始化大小
     */
    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
            count++;

            // 当向队列中成功放入一个元素之后，需要做什么呢？
            // 需要给notEmpty一个唤醒信号，告诉消费者去消费
            notEmpty.signal();
        } finally {
            // 释放锁
            lock.unlock();
        }
    }

    /**
     * 获取数据 消费者
     * @return Object
     */
    @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
            count--;

            // 当从队列中成功消费一个元素之后，需要做什么呢？
            // 需要给notFull一个唤醒信号，告诉生产者去生产
            notFull.signal();

            // 返回消费的element
            return element;
        } finally {
            // 释放锁
            lock.unlock();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

测试

public class MiniArrayBrokingQueueTest {
    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();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

输出

生产者线程生产数据：1
消费者线程消费数据：1
生产者线程生产数据：2
消费者线程消费数据：2
生产者线程生产数据：3
消费者线程消费数据：3
生产者线程生产数据：4
消费者线程消费数据：4
生产者线程生产数据：5
消费者线程消费数据：5
生产者线程生产数据：6
消费者线程消费数据：6
生产者线程生产数据：7
消费者线程消费数据：7
生产者线程生产数据：8
消费者线程消费数据：8
生产者线程生产数据：9
消费者线程消费数据：9
生产者线程生产数据：0
消费者线程消费数据：0
生产者线程生产数据：1
消费者线程消费数据：1

Process finished with exit code -1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

很完美的实现了 BrokingQueue 阻塞队列的 插入数据 和 获取数据 的功能。

---

参考

• 视频参考
  • b站_小刘讲源码付费课
• 文章参考
  • 兴趣使然的草帽路飞_AQS源码探究_05 Conditon条件队列(手写一个入门的BrokingQueue)