轉載請注明出處：https://www.cnblogs.com/wenjunwei/p/10411444.html
前言
本文通過一個簡單的例子，來展現如何使用阻塞隊列(BlockingQueue)來實現異步通信功能。(這是筆者在做日志記錄時，用來做異步操作的)
BlockingQueue的核心方法
放入數據：
- offer(anObject):表示如果可能的話,將anObject加到BlockingQueue里,即如果BlockingQueue可以容納,則返回true,否則返回false.（本方法不阻塞當前執行方法的線程）。
- offer(E o, long timeout, TimeUnit unit):可以設定等待的時間，如果在指定的時間內，還不能往隊列中加入BlockingQueue，則返回失敗。
- put(anObject):把anObject加到BlockingQueue里,如果BlockQueue沒有空間,則調用此方法的線程被阻斷直到BlockingQueue里面有空間再繼續。
獲取數據：
- poll(time):取走BlockingQueue里排在首位的對象,若不能立即取出,則可以等time參數規定的時間,取不到時返回null。
- poll(long timeout, TimeUnit unit)：從BlockingQueue取出一個隊首的對象，如果在指定時間內，隊列一旦有數據可取，則立即返回隊列中的數據。否則知道時間超時還沒有數據可取，返回失敗。
- take():取走BlockingQueue里排在首位的對象,若BlockingQueue為空,阻斷進入等待狀態直到BlockingQueue有新的數據被加入。
- drainTo():一次性從BlockingQueue獲取所有可用的數據對象（還可以指定獲取數據的個數），通過該方法，可以提升獲取數據效率；不需要多次分批加鎖或釋放鎖。
實現
這里使用Springboot搭建一個微服務作為啟動程序。 並定時對消息進行消費。
通過阻塞隊列的方式，實現消息的異步處理。
阻塞隊列
BlockingQueueMessage.java
定義一個LinkedBlockingQueue隊列，並設置隊列長度為5.
public class BlockingQueueMessage { public static BlockingQueue<String> queue = new LinkedBlockingQueue<>(5); }
隊列生產者
這里定時器用於模擬日志的產生。
隊列生產者：每一秒鍾產生一條日志消息並放入隊列中
Producer.java
@Component public class Producer { private static AtomicInteger count = new AtomicInteger(); @Scheduled(fixedDelay = 1000) public void producerMessage1() { offer("張三"); } @Scheduled(fixedDelay = 1000) public void producerMessage2() { offer("李四"); } @Scheduled(fixedDelay = 1000) public void producerMessage3() { offer("王五"); } private void offer(String name){ try { String str = String.valueOf(count.incrementAndGet()); System.out.println(name + "生產消息：" + str); BlockingQueueMessage.queue.offer(str, 2, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } } }
隊列消費者
這里定時器用於模擬消費產生的日志信息，獲取到后可以進行入庫等操作。
隊列消費者：此處使用了whiile循環，如果能獲取到隊列，則消費完后繼續獲取，當獲取到隊列的消息msg為null時，代表現在隊列中已經已經沒有消息，則跳出循環，3秒之后定時器再繼續進行消費。BlockingQueueMessage.queue.poll(2, TimeUnit.SECONDS)設置了阻塞隊列獲取時等待超時時間為2秒。
Consumer.java
@Component public class Consumer { @Scheduled(fixedDelay = 3000) public void consumerMessage() { boolean isRunning = true; while (isRunning) { try { String msg = BlockingQueueMessage.queue.poll(2, TimeUnit.SECONDS); if (null != msg) { System.out.println("接收到的消息：" + msg); } else { isRunning = false; } } catch (InterruptedException e) { e.printStackTrace(); } } } }
springboot啟動類
啟動類上面需要增加@EnableScheduling注解，否則定時器不生效。
QueueApplication.java
@SpringBootApplication @EnableScheduling public class QueueApplication { public static void main(String[] args) { SpringApplication.run(QueueApplication.class); } }
maven依賴只需引入
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
輸出結果
張三生產消息：1 李四生產消息：2 王五生產消息：3 張三生產消息：4 李四生產消息：5 王五生產消息：6 張三生產消息：7 李四生產消息：8 ---接收到的消息---：1 ---接收到的消息---：2 ---接收到的消息---：3 ---接收到的消息---：4 ---接收到的消息---：5 王五生產消息：9 張三生產消息：10 李四生產消息：11 王五生產消息：12 張三生產消息：13 李四生產消息：14 王五生產消息：15 張三生產消息：16 ---接收到的消息---：9 ---接收到的消息---：10 ---接收到的消息---：11 ---接收到的消息---：12 ---接收到的消息---：13
一看結果，立馬發現不對，收到的消息丟失了呢？
因為這里設置了LinkedBlockingQueue長度為5，當隊列中消息大於長度時，則無法繼續寫入隊列，只有當隊列被消費后才可以繼續存入。所以這個消息長度最好是可以設置的，比如從配置文件中獲取。
優化
springboot靜態變量賦值
Constants.java
@Component public class Constants { //這里可以給默認值，也可以不給 public static int queueNum = 5; //set方法一定不能是static的，否則無效,因為靜態方法是在編譯時就初始化的 @Value("${queueNum}") public void setQueueNum(int queueNum) { Constants.queueNum = queueNum; } }
application.yml
queueNum: 10
BlockingQueueMessage.java
將隊列數量改為從變量中獲取.
public class BlockingQueueMessage { public static BlockingQueue<String> queue = new LinkedBlockingQueue<>(Constants.queueNum); }
重新啟動項目，獲得結果，收到消息完整。
張三生產消息：1 李四生產消息：2 王五生產消息：3 張三生產消息：4 李四生產消息：5 王五生產消息：6 張三生產消息：7 李四生產消息：8 王五生產消息：9 ---接收到的消息---：1 ---接收到的消息---：2 ---接收到的消息---：3 ---接收到的消息---：4 ---接收到的消息---：5 ---接收到的消息---：6 ---接收到的消息---：7 ---接收到的消息---：8 ---接收到的消息---：9
常見的BlockingQueue
1. ArrayBlockingQueue
基於數組的阻塞隊列實現，在ArrayBlockingQueue內部，維護了一個定長數組，以便緩存隊列中的數據對象，這是一個常用的阻塞隊列，除了一個定長數組外，ArrayBlockingQueue內部還保存着兩個整形變量，分別標識着隊列的頭部和尾部在數組中的位置。
ArrayBlockingQueue在生產者放入數據和消費者獲取數據，都是共用同一個鎖對象，由此也意味着兩者無法真正並行運行，這點尤其不同於LinkedBlockingQueue；按照實現原理來分析，ArrayBlockingQueue完全可以采用分離鎖，從而實現生產者和消費者操作的完全並行運行。Doug Lea之所以沒這樣去做，也許是因為ArrayBlockingQueue的數據寫入和獲取操作已經足夠輕巧，以至於引入獨立的鎖機制，除了給代碼帶來額外的復雜性外，其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue間還有一個明顯的不同之處在於，前者在插入或刪除元素時不會產生或銷毀任何額外的對象實例，而后者則會生成一個額外的Node對象。這在長時間內需要高效並發地處理大批量數據的系統中，其對於GC的影響還是存在一定的區別。而在創建ArrayBlockingQueue時，我們還可以控制對象的內部鎖是否采用公平鎖，默認采用非公平鎖。
2. LinkedBlockingQueue
基於鏈表的阻塞隊列，同ArrayListBlockingQueue類似，其內部也維持着一個數據緩沖隊列（該隊列由一個鏈表構成），當生產者往隊列中放入一個數據時，隊列會從生產者手中獲取數據，並緩存在隊列內部，而生產者立即返回；只有當隊列緩沖區達到最大值緩存容量時（LinkedBlockingQueue可以通過構造函數指定該值），才會阻塞生產者隊列，直到消費者從隊列中消費掉一份數據，生產者線程會被喚醒，反之對於消費者這端的處理也基於同樣的原理。而LinkedBlockingQueue之所以能夠高效的處理並發數據，還因為其對於生產者端和消費者端分別采用了獨立的鎖來控制數據同步，這也意味着在高並發的情況下生產者和消費者可以並行地操作隊列中的數據，以此來提高整個隊列的並發性能。
作為開發者，我們需要注意的是，如果構造一個LinkedBlockingQueue對象，而沒有指定其容量大小，LinkedBlockingQueue會默認一個類似無限大小的容量（Integer.MAX_VALUE），這樣的話，如果生產者的速度一旦大於消費者的速度，也許還沒有等到隊列滿阻塞產生，系統內存就有可能已被消耗殆盡了。
ArrayBlockingQueue和LinkedBlockingQueue是兩個最普通也是最常用的阻塞隊列，一般情況下，在處理多線程間的生產者消費者問題，使用這兩個類足以。
3. DelayQueue
DelayQueue中的元素只有當其指定的延遲時間到了，才能夠從隊列中獲取到該元素。DelayQueue是一個沒有大小限制的隊列，因此往隊列中插入數據的操作（生產者）永遠不會被阻塞，而只有獲取數據的操作（消費者）才會被阻塞。
使用場景：
DelayQueue使用場景較少，但都相當巧妙，常見的例子比如使用一個DelayQueue來管理一個超時未響應的連接隊列。
4. PriorityBlockingQueue
基於優先級的阻塞隊列（優先級的判斷通過構造函數傳入的Compator對象來決定），但需要注意的是PriorityBlockingQueue並不會阻塞數據生產者，而只會在沒有可消費的數據時，阻塞數據的消費者。因此使用的時候要特別注意，生產者生產數據的速度絕對不能快於消費者消費數據的速度，否則時間一長，會最終耗盡所有的可用堆內存空間。在實現PriorityBlockingQueue時，內部控制線程同步的鎖采用的是公平鎖。
5. SynchronousQueue
一種無緩沖的等待隊列，類似於無中介的直接交易，有點像原始社會中的生產者和消費者，生產者拿着產品去集市銷售給產品的最終消費者，而消費者必須親自去集市找到所要商品的直接生產者，如果一方沒有找到合適的目標，那么對不起，大家都在集市等待。相對於有緩沖的BlockingQueue來說，少了一個中間經銷商的環節（緩沖區），如果有經銷商，生產者直接把產品批發給經銷商，而無需在意經銷商最終會將這些產品賣給那些消費者，由於經銷商可以庫存一部分商品，因此相對於直接交易模式，總體來說采用中間經銷商的模式會吞吐量高一些（可以批量買賣）；但另一方面，又因為經銷商的引入，使得產品從生產者到消費者中間增加了額外的交易環節，單個產品的及時響應性能可能會降低。
聲明一個SynchronousQueue有兩種不同的方式，它們之間有着不太一樣的行為。公平模式和非公平模式的區別:
- 如果采用公平模式：SynchronousQueue會采用公平鎖，並配合一個FIFO隊列來阻塞多余的生產者和消費者，從而體系整體的公平策略；
- 但如果是非公平模式（SynchronousQueue默認）：SynchronousQueue采用非公平鎖，同時配合一個LIFO隊列來管理多余的生產者和消費者，而后一種模式，如果生產者和消費者的處理速度有差距，則很容易出現飢渴的情況，即可能有某些生產者或者是消費者的數據永遠都得不到處理。
感謝您的閱讀，如果您覺得閱讀本文對您有幫助，請點一下“推薦”按鈕。本文歡迎各位轉載，但是轉載文章之后必須在文章開頭給出原文鏈接