使用場景
由于公司業務需求，需要對接socket、MQTT等消息隊列。
眾所周知 socket 是雙向通信，socket的回復是人為定義的，客戶端推送消息給服務端，服務端的回復是兩條線。無法像http請求有回復。
下發指令給硬件時，需要校驗此次數據下發是否成功。
用戶體驗而言，點擊按鈕就要知道此次的下發成功或失敗。
如上圖模型，
第一種方案使用Tread.sleep
優點:占用資源小，放棄當前cpu資源
缺點: 回復速度快，休眠時間過長，仍然需要等待休眠結束才能返回，響應速度是固定的，無法及時響應第二種方案使用CountDownLatch
|
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
|
package com.lzy.demo.delay;import java.util.Map;import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.CountDownLatch;import java.util.concurrent.DelayQueue;import java.util.concurrent.Delayed;import java.util.concurrent.ExecutorService;import java.util.concurrent.ThreadPoolExecutor;import java.util.concurrent.TimeUnit;public class CountDownLatchPool { //countDonw池 private final static Map<Integer, CountDownLatch> countDownLatchMap = new ConcurrentHashMap<>(); //延遲隊列 private final static DelayQueue<MessageDelayQueueUtil> delayQueue = new DelayQueue<>(); private volatile static boolean flag =false; //單線程池 private final static ExecutorService t = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1)); public static void addCountDownLatch(Integer messageId) { CountDownLatch countDownLatch = countDownLatchMap.putIfAbsent(messageId,new CountDownLatch(1) ); if(countDownLatch == null){ countDownLatch = countDownLatchMap.get(messageId); } try { addDelayQueue(messageId); countDownLatch.await(3L, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("阻塞等待結束~~~~~~"); } public static void removeCountDownLatch(Integer messageId){ CountDownLatch countDownLatch = countDownLatchMap.get(messageId); if(countDownLatch == null) return; countDownLatch.countDown(); countDownLatchMap.remove(messageId); System.out.println("清除Map數據"+countDownLatchMap); } private static void addDelayQueue(Integer messageId){ delayQueue.add(new MessageDelayQueueUtil(messageId)); clearMessageId(); } private static void clearMessageId(){ synchronized (CountDownLatchPool.class){ if(flag){ return; } flag = true; } t.execute(()->{ while (delayQueue.size() > 0){ System.out.println("進入線程并開始執行"); try { MessageDelayQueueUtil take = delayQueue.take(); Integer messageId1 = take.getMessageId(); removeCountDownLatch(messageId1); System.out.println("清除隊列數據"+messageId1); } catch (InterruptedException e) { e.printStackTrace(); } } flag = false; System.out.println("結束end----"); }); } public static void main(String[] args) throws InterruptedException { /* 測試超時清空map new Thread(()->addCountDownLatch(1)).start(); new Thread(()->addCountDownLatch(2)).start(); new Thread(()->addCountDownLatch(3)).start(); */ //提前創建線程，清空countdown new Thread(()->{ try { Thread.sleep(500L); removeCountDownLatch(1); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); //開始阻塞 addCountDownLatch(1); //通過調整上面的sleep我們發現阻塞市場取決于countDownLatch.countDown()執行時間 System.out.println("阻塞結束----"); }}class MessageDelayQueueUtil implements Delayed { private Integer messageId; private long avaibleTime; public Integer getMessageId() { return messageId; } public void setMessageId(Integer messageId) { this.messageId = messageId; } public long getAvaibleTime() { return avaibleTime; } public void setAvaibleTime(long avaibleTime) { this.avaibleTime = avaibleTime; } public MessageDelayQueueUtil(Integer messageId){ this.messageId = messageId; //avaibleTime = 當前時間+ delayTime //重試3次，每次3秒+1秒的延遲 this.avaibleTime=3000*3+1000 + System.currentTimeMillis(); } @Override public long getDelay(TimeUnit unit) { long diffTime= avaibleTime- System.currentTimeMillis(); return unit.convert(diffTime,TimeUnit.MILLISECONDS); } @Override public int compareTo(Delayed o) { //compareTo用在DelayedUser的排序 return (int)(this.avaibleTime - ((MessageDelayQueueUtil) o).getAvaibleTime()); }} |
由于socket并不確定每次都會有數據返回，所以map的數據會越來越大，最終導致內存溢出
需定時清除map內的無效數據。
可以使用DelayedQuene延遲隊列來處理，相當于給對象添加一個過期時間
使用方法 addCountDownLatch 等待消息,異步回調消息清空removeCountDownLatch
到此這篇關于詳解Java中CountDownLatch異步轉同步工具類的文章就介紹到這了,更多相關CountDownLatch異步轉同步工具類內容請搜索服務器之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持服務器之家！
原文鏈接：https://blog.csdn.net/qq_37256345/article/details/117808156
