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

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

  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
 

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

  

 

  
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
  
136
  
137
  
138
  
139
  
140
  
141
  
142
  
143
  
144
  
145
  
146
  
147
  
148
  
149
  
150
  
151
  
152
  
153
  
154
  
155
  
156
  
157
  
158
  
159
  
160
  
161
  
162
  
163
  
164
  
165
  
166
 

文章来源: csp1999.blog.csdn.net，作者：兴趣使然の草帽路飞，版权归原作者所有，如需转载，请联系作者。

原文链接：csp1999.blog.csdn.net/article/details/116381160