1 package com.thread.test.thread;
2 import java.util.Random;
3 import java.util.concurrent.*;
4
5 /**
6 * Created by windwant on 2016/5/26.
7 */
8 public class MyBlockingQueue {
9 public static void main(String[] args) throws InterruptedException {
10 testArrayBlockingQueue();
11 }
12
13 /**
14 * 公平性 構造函數 true
15 */
16 public static void testArrayBlockingQueue(){
17 BlockingQueue<String> abq = new ArrayBlockingQueue<String>(5);
18 ExecutorService es = Executors.newCachedThreadPool();
19 es.execute(new MyPro(abq, 1000));
20 es.execute(new MyCus(abq, 5000));
21 es.shutdown();
22 }
23
24 /**
25 * 基於鏈表節點的可設置容量的隊列，先進先出，隊尾插入元素，隊首獲取元素。
26 * 鏈表隊列比基於數據的隊列有更高的存取效率，但是在並發應用中效率無法預測。
27 */
28 public static void testLinkedBlockingQueue(){
29 BlockingQueue<String> abq = new LinkedBlockingQueue<String>(5);
30 ExecutorService es = Executors.newCachedThreadPool();
31 es.execute(new MyPro(abq, 20));
32 es.execute(new MyCus(abq, 2000));
33 es.shutdown();
34 }
35
36 /**
37 * DelayQueue
38 * 無容量限制的阻塞隊列，元素包含延遲時限，只有到達時限，元素才能被取出。
39 * 隊列頂部是距離到期時間最遠的元素。
40 * 如果所有的元素都未到期，將會返回null。
41 * 元素在執行getDelay()方法返回值小於等於0時過期，即使沒有被通過take或者poll執行提取，它們也會被當作一般元素對待。
42 * 隊列size方法返回所有元素的數量。
43 * 隊列不能包含null元素。
44 */
45 public static void testDelayQueue() throws InterruptedException {
46 DelayQueue<MyDelayItem> dq = new DelayQueue<MyDelayItem>();
47 ExecutorService es = Executors.newFixedThreadPool(5);
48 es.execute(new MyDelayPro(dq, 1000));
49 es.execute(new MyDelayCus(dq, 10000));
50 es.shutdown();
51 }
52
53 /**
54 * 無容量限制的阻塞隊列，元素順序維持策略同PriorityQueue一樣，支持阻塞獲取
55 * 不允許添加null元素
56 * 元素必須支持排序
57 * 支持集合遍歷，排序
58 */
59 public static void testPriorityBlockingQueue() throws InterruptedException {
60 PriorityBlockingQueue<MyPriorityItem> pbq = new PriorityBlockingQueue<MyPriorityItem>();
61 ExecutorService es = Executors.newFixedThreadPool(5);
62 es.execute(new MyPriorityBlockingQueuePro(pbq, 1000));
63 es.execute(new MyPriorityBlockingQueueCus(pbq, 10000));
64 es.shutdown();
65 }
66
67 /**
68 * 阻塞隊列，插入元素和提取元素必須同步。
69 * 異步隊列沒有容量的概念。
70 * 無法使用peek，因為只有當你嘗試移除時，元素才會存在。
71 * 無法插入元素，除非有另外一個線程同時嘗試獲取元素。
72 * 不支持遍歷操作，因為隊列中根本沒有元素。
73 * 隊列的頂部就是嘗試插入元素的線程要插入的元素。
74 * 如果沒有嘗試插入元素的線程，那麼就不存在能夠提取的元素，poll會返回null。
75 * 集合操作contains返null
76 * 不允許插入null元素
77 * */
78 public static void testSynchronousQueue() throws InterruptedException {
79 SynchronousQueue<String> sq = new SynchronousQueue<String>();
80 ExecutorService es = Executors.newFixedThreadPool(5);
81 es.execute(new MySynchronousQueuePro(sq, 1000));
82 es.execute(new MySynchronousQueueCus(sq, 2000));
83 es.shutdown();
84 }
85 }
86
87 /**
88 * 測試生產者
89 */
90 class MyPro implements Runnable{
91
92 private BlockingQueue<String> bq;
93
94 private int period = 1000;
95
96 private Random r = new Random();
97 MyPro(BlockingQueue bq, int period){
98 this.bq = bq;
99 this.period = period;
100 }
101
102 public void run() {
103 try{
104 while (true){
105 Thread.sleep(period);
106 String value = String.valueOf(r.nextInt(100));
107 if(bq.offer(value)){ //offer 能夠插入就返回true，否則返回false
108 System.out.println("pro make value: " + value + " queue : " + bq.toString());
109 System.out.println("******************************************************");
110 }
111 }
112 }catch (InterruptedException e){
113 e.printStackTrace();
114 }
115 }
116 }
117
118 /**
119 * 測試消費者
120 */
121 class MyCus implements Runnable{
122
123 private BlockingQueue<String> bq;
124
125 private int period = 1000;
126
127 private Random r = new Random();
128 MyCus(BlockingQueue bq, int period){
129 this.bq = bq;
130 this.period = period;
131 }
132
133 public void run() {
134 try{
135 while (true){
136 Thread.sleep(period);
137 String value = bq.take(); //獲取隊列頭部元素，無元素則阻塞
138 System.out.println("cus take value: " + value + " queue : " + bq.toString());
139 System.out.println("======================================================");
140 }
141 }catch (InterruptedException e){
142 e.printStackTrace();
143 }
144 }
145 }
146
147 /**
148 * 延遲隊列元素 實現排序
149 */
150 class MyDelayItem implements Delayed{
151
152 private long liveTime;
153
154 private long removeTime;
155
156 MyDelayItem(long liveTime, long removeTime){
157 this.liveTime = liveTime;
158 this.removeTime = TimeUnit.MILLISECONDS.convert(liveTime, TimeUnit.MILLISECONDS) + System.nanoTime();
159 }
160
161 public long getDelay(TimeUnit unit) {
162 return unit.convert(removeTime - System.nanoTime(), unit);
163 }
164
165 public int compareTo(Delayed o) {
166 if(o == null) return -1;
167 if(o == this) return 0;
168 if(o instanceof MyDelayItem){
169 MyDelayItem tmp = (MyDelayItem) o;
170 if(liveTime > tmp.liveTime){
171 return 1;
172 }else if(liveTime == tmp.liveTime){
173 return 0;
174 }else{
175 return -1;
176 }
177 }
178 long diff = getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
179 return diff > 0 ? 1 : diff == 0 ? 0 : -1;
180 }
181
182 public String toString(){
183 return "{livetime: " + String.valueOf(liveTime) + ", removetime: " + String.valueOf(removeTime) + "}";
184 }
185 }
186
187 /**
188 * 延遲隊列測試生產者
189 */
190 class MyDelayPro implements Runnable{
191
192 private DelayQueue<MyDelayItem> dq;
193
194 private int period = 1000;
195
196 private Random r = new Random();
197
198 MyDelayPro(DelayQueue dq, int period){
199 this.dq = dq;
200 this.period = period;
201 }
202 public void run() {
203 try{
204 while (true){
205 Thread.sleep(period);
206 if(dq.size() > 5){
207 continue;
208 }
209 MyDelayItem di = new MyDelayItem(r.nextInt(10), r.nextInt(10));
210 dq.offer(di);
211 System.out.println("delayqueue: add---" + di.toString() + "size: " + dq.size());
212 System.out.println("*************************************");
213 }
214 }catch (InterruptedException e){
215 e.printStackTrace();
216 }
217 }
218 }
219
220 /**
221 * 延遲隊列測試消費者
222 */
223 class MyDelayCus implements Runnable{
224
225 private DelayQueue<MyDelayItem> dq;
226
227 private int period = 1000;
228
229 MyDelayCus(DelayQueue dq, int period){
230 this.dq = dq;
231 this.period = period;
232 }
233 public void run() {
234 try{
235 while (true){
236 Thread.sleep(period);
237 MyDelayItem di = dq.take();
238 System.out.println("delayqueue: remove---" + di.toString());
239 System.out.println("delayqueue: ---" + dq.toString());
240 System.out.println("======================================");
241 }
242 }catch (InterruptedException e){
243 e.printStackTrace();
244 }
245 }
246 }
247
248 /**
249 * 延遲隊列元素 時限排序對比延遲隊列
250 */
251 class MyPriorityItem implements Comparable<MyPriorityItem> {
252
253 private int priority;
254
255 MyPriorityItem(int priority){
256 this.priority = priority;
257 }
258
259 /**
260 * 數字大優先級高
261 * @param o
262 * @return
263 */
264 public int compareTo(MyPriorityItem o) {
265 if(o == null) return -1;
266 if(o == this) return 0;
267 if(priority > o.priority){
268 return -1;
269 }else if(priority == o.priority){
270 return 0;
271 }else{
272 return 1;
273 }
274 }
275
276 public String toString(){
277 return "{priority: " + String.valueOf(priority) + "}";
278 }
279 }
280
281 /**
282 * 優先隊列測試生產者
283 */
284 class MyPriorityBlockingQueuePro implements Runnable{
285
286 private PriorityBlockingQueue<MyPriorityItem> pbq;
287
288 private int period = 1000;
289
290 private Random r = new Random();
291
292 MyPriorityBlockingQueuePro(PriorityBlockingQueue pbq, int period){
293 this.pbq = pbq;
294 this.period = period;
295 }
296 public void run() {
297 try{
298 while (true){
299 Thread.sleep(period);
300 if(pbq.size() > 5){
301 continue;
302 }
303 MyPriorityItem pi = new MyPriorityItem(r.nextInt(10));
304 pbq.offer(pi);
305 System.out.println("PriorityBlockingQueue: add---" + pi.toString() + " size: " + pbq.size());
306 System.out.println("PriorityBlockingQueue: " + pbq.toString());
307 System.out.println("*************************************");
308 }
309 }catch (InterruptedException e){
310 e.printStackTrace();
311 }
312 }
313 }
314
315 /**
316 * 優先隊列測試消費者
317 */
318 class MyPriorityBlockingQueueCus implements Runnable{
319
320 private PriorityBlockingQueue<MyPriorityItem> pbq;
321
322 private int period = 1000;
323
324 private Random r = new Random();
325
326 MyPriorityBlockingQueueCus(PriorityBlockingQueue pbq, int period){
327 this.pbq = pbq;
328 this.period = period;
329 }
330 public void run() {
331 try{
332 while (true){
333 Thread.sleep(period);
334 MyPriorityItem di = pbq.take();
335 System.out.println("PriorityBlockingQueue: remove---" + di.toString());
336 System.out.println("PriorityBlockingQueue: ---" + pbq.toString());
337 System.out.println("======================================");
338 }
339 }catch (InterruptedException e){
340 e.printStackTrace();
341 }
342 }
343 }
344
345 /**
346 * 阻塞隊列測試生產者
347 */
348 class MySynchronousQueuePro implements Runnable{
349
350 private SynchronousQueue<String> sq;
351
352 private int period = 1000;
353
354 private Random r = new Random();
355 MySynchronousQueuePro(SynchronousQueue sq, int period){
356 this.sq = sq;
357 this.period = period;
358 }
359
360 public void run() {
361 try{
362 while (true){
363 Thread.sleep(period);
364 String value = String.valueOf(r.nextInt(100));
365 if(sq.offer(value)) {
366 System.out.println("pro make value: " + value + " synchronous :" + sq.toString());
367 System.out.println("******************************************************");
368 }
369 }
370 }catch (InterruptedException e){
371 e.printStackTrace();
372 }
373 }
374 }
375
376 /**
377 * 阻塞隊列測試消費者
378 */
379 class MySynchronousQueueCus implements Runnable{
380
381 private BlockingQueue<String> sq;
382
383 private int period = 1000;
384
385 MySynchronousQueueCus(BlockingQueue sq, int period){
386 this.sq = sq;
387 this.period = period;
388 }
389
390 public void run() {
391 try{
392 while (true){
393 Thread.sleep(period);
394 String value = sq.take();
395 System.out.println("cus take value: " + value + " synchronous :" + sq.toString());
396 System.out.println("======================================================");
397 }
398 }catch (InterruptedException e){
399 e.printStackTrace();
400 }
401 }
402 }
項目地址：https://github.com/windwant/threadtest
