自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

JUC之阻塞隊(duì)列BlockingQueue竟然有8種類型？

作者：一角錢技術(shù)
開發(fā) 前端
隊(duì)列是一種特殊的線性表，是一種先進(jìn)先出（FIFO）的數(shù)據(jù)結(jié)構(gòu)。它只允許在表的前端（front）進(jìn)行刪除操作，而在表的后端（rear）進(jìn)行插入操作。進(jìn)行插入操作的端稱為隊(duì)尾，進(jìn)行刪除操作的端稱為隊(duì)頭。隊(duì)列中沒有元素時(shí)，稱為空隊(duì)列。

 隊(duì)列是一種特殊的線性表，是一種先進(jìn)先出(FIFO)的數(shù)據(jù)結(jié)構(gòu)。它只允許在表的前端(front)進(jìn)行刪除操作，而在表的后端(rear)進(jìn)行插入操作。進(jìn)行插入操作的端稱為隊(duì)尾，進(jìn)行刪除操作的端稱為隊(duì)頭。隊(duì)列中沒有元素時(shí)，稱為空隊(duì)列。

下面是Queue類的繼承關(guān)系圖：


Queue


Queue：隊(duì)列的上層接口，提供了插入、刪除、獲取元素這3種類型的方法，而且對(duì)每一種類型都提供了兩種方式，先來看看插入方法：

  • add(E e)：插入元素到隊(duì)尾，插入成功返回true，沒有可用空間拋出異常 IllegalStateException。
  • offer(E e)： 插入元素到隊(duì)尾，插入成功返回true，否則返回false。

add和offer作為插入方法的唯一不同就在于隊(duì)列滿了之后的處理方式。add拋出異常，而offer返回false。

再來看看刪除和獲取元素方法(和插入方法類似)：

  • remove()：獲取并移除隊(duì)首的元素，該方法和poll方法的不同之處在于，如果隊(duì)列為空該方法會(huì)拋出異常，而poll不會(huì)。
  • poll()：獲取并移除隊(duì)首的元素，如果隊(duì)列為空，返回null。
  • element()：獲取隊(duì)列首的元素，該方法和peek方法的不同之處在于，如果隊(duì)列為空該方法會(huì)拋出異常，而peek不會(huì)。
  • peek()：獲取隊(duì)列首的元素，如果隊(duì)列為空，返回null。

如果隊(duì)列是空，remove和element方法會(huì)拋出異常，而poll和peek返回null。

  • Queue 是單向隊(duì)列，為了提供更強(qiáng)大的功能，JDK在1.6的時(shí)候新增了一個(gè)雙向隊(duì)列
  • Deque，用來實(shí)現(xiàn)更靈活的隊(duì)列操作。

Deque

Deque在Queue的基礎(chǔ)上，增加了以下幾個(gè)方法：

  • addFirst(E e)：在前端插入元素，異常處理和add一樣;
  • addLast(E e)：在后端插入元素，和add一樣的效果;
  • offerFirst(E e)：在前端插入元素，異常處理和offer一樣;
  • offerLast(E e)：在后端插入元素，和offer一樣的效果;
  • removeFirst()：移除前端的一個(gè)元素，異常處理和remove一樣;
  • removeLast()：移除后端的一個(gè)元素，和remove一樣的效果;
  • pollFirst()：移除前端的一個(gè)元素，和poll一樣的效果;
  • pollLast()：移除后端的一個(gè)元素，異常處理和poll一樣;
  • getFirst()：獲取前端的一個(gè)元素，和element一樣的效果;
  • getLast()：獲取后端的一個(gè)元素，異常處理和element一樣;
  • peekFirst()：獲取前端的一個(gè)元素，和peek一樣的效果;
  • peekLast()：獲取后端的一個(gè)元素，異常處理和peek一樣;
  • removeFirstOccurrence(Object o)：從前端開始移除第一個(gè)是o的元素;
  • removeLastOccurrence(Object o)：從后端開始移除第一個(gè)是o的元素;
  • push(E e)：和addFirst一樣的效果;
  • pop()：和removeFirst一樣的效果。

可以發(fā)現(xiàn)，其實(shí)很多方法的效果都是一樣的，只不過名字不同。比如Deque為了實(shí)現(xiàn)Stack的語義，定義了push和pop兩個(gè)方法。

BlockingQueue阻塞隊(duì)列

BlockingQueue(阻塞隊(duì)列)，在Queue的基礎(chǔ)上實(shí)現(xiàn)了阻塞等待的功能。它是JDK 1.5中加入的接口，它是指這樣的一個(gè)隊(duì)列：當(dāng)生產(chǎn)者向隊(duì)列添加元素但隊(duì)列已滿時(shí)，生產(chǎn)者會(huì)被阻塞;當(dāng)消費(fèi)者從隊(duì)列移除元素但隊(duì)列為空時(shí)，消費(fèi)者會(huì)被阻塞。

BlockingQueue ，是java.util.concurrent 包提供的用于解決并發(fā) 生產(chǎn)者 — 消費(fèi)者 問題的最有用的類，很好的解決了多線程中，如何高效安全“傳輸”數(shù)據(jù)的問題。它的特性是在任意時(shí)刻只有一個(gè)線程可以進(jìn)行take或者put操作，并且 BlockingQueue 提供類超時(shí) return null 的機(jī)制，在許多生產(chǎn)場(chǎng)景里都可以看到這個(gè)工具的身影。

總體認(rèn)識(shí)

一般我們用到的阻塞隊(duì)列有哪些?可以通過下面一個(gè)類圖來總體看下：


可以看到 BlockingQueue 是一個(gè)接口，繼承它的另外還有兩個(gè)接口 BlockingDeque(雙端隊(duì)列)、TransferQueue(兩個(gè)線程之間傳遞元素)。

阻塞隊(duì)列的成員如下：


隊(duì)列類型

  1. 無限隊(duì)列(unbounded queue)— 幾乎可以無限增長
  2. 有限隊(duì)列(bounded queue)— 定義了最大容量

隊(duì)列數(shù)據(jù)結(jié)構(gòu)

隊(duì)列實(shí)質(zhì)就是一種存儲(chǔ)數(shù)據(jù)的結(jié)構(gòu)

  • 通常用鏈表或者數(shù)組實(shí)現(xiàn)
  • 一般而言隊(duì)列具備FIFO先進(jìn)先出的特性，當(dāng)然也有雙端隊(duì)列(Deque)優(yōu)先級(jí)隊(duì)列
  • 主要操作：入隊(duì)(Enqueue)與 出對(duì)(Dequeue)

常見的5種阻塞隊(duì)列

  • ArrayBlockingQueue：一個(gè)由數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列。
  • LinkedBlockingQueue：一個(gè)由鏈表結(jié)構(gòu)組成的有界阻塞隊(duì)列。
  • PriorityBlockingQueue：一個(gè)支持優(yōu)先級(jí)排序的無界阻塞隊(duì)列。
  • SynchronousQueue：一個(gè)不存儲(chǔ)元素的阻塞隊(duì)列。
  • DelayQueue：一個(gè)使用優(yōu)先級(jí)隊(duì)列實(shí)現(xiàn)的無界阻塞隊(duì)列。

BlockingQueue API

BlockingQueue的核心方法

  1. public interface BlockingQueue<E> extends Queue<E> { 
  2.  
  3.     //將給定元素設(shè)置到隊(duì)列中，如果設(shè)置成功返回true, 否則返回false。如果是往限定了長度的隊(duì)列中設(shè)置值，推薦使用offer()方法。 
  4.     boolean add(E e); 
  5.  
  6.     //將給定的元素設(shè)置到隊(duì)列中，如果設(shè)置成功返回true, 否則返回false. e的值不能為空，否則拋出空指針異常。 
  7.     boolean offer(E e); 
  8.  
  9.     //將元素設(shè)置到隊(duì)列中，如果隊(duì)列中沒有多余的空間，該方法會(huì)一直阻塞，直到隊(duì)列中有多余的空間。 
  10.     void put(E e) throws InterruptedException; 
  11.  
  12.     //將給定元素在給定的時(shí)間內(nèi)設(shè)置到隊(duì)列中，如果設(shè)置成功返回true, 否則返回false. 
  13.     boolean offer(E e, long timeout, TimeUnit unit) 
  14.         throws InterruptedException; 
  15.  
  16.     //從隊(duì)列中獲取值，如果隊(duì)列中沒有值，線程會(huì)一直阻塞，直到隊(duì)列中有值，并且該方法取得了該值。 
  17.     E take() throws InterruptedException; 
  18.  
  19.     //在給定的時(shí)間里，從隊(duì)列中獲取值，時(shí)間到了直接調(diào)用普通的poll方法，為null則直接返回null。 
  20.     E poll(long timeout, TimeUnit unit) 
  21.         throws InterruptedException; 
  22.  
  23.     //獲取隊(duì)列中剩余的空間。 
  24.     int remainingCapacity(); 
  25.  
  26.     //從隊(duì)列中移除指定的值。 
  27.     boolean remove(Object o); 
  28.  
  29.     //判斷隊(duì)列中是否擁有該值。 
  30.     public boolean contains(Object o); 
  31.  
  32.     //將隊(duì)列中值，全部移除，并發(fā)設(shè)置到給定的集合中。 
  33.     int drainTo(Collection<? super E> c); 
  34.  
  35.     //指定最多數(shù)量限制將隊(duì)列中值，全部移除，并發(fā)設(shè)置到給定的集合中。 
  36.     int drainTo(Collection<? super E> c, int maxElements); 
  37. } 

 BlockingQueue 接口的所有方法可以分為兩大類：負(fù)責(zé)向隊(duì)列添加元素的方法和 檢索這些元素的方法。在隊(duì)列滿/空的情況下，來自這兩個(gè)組的每個(gè)方法的行為都不同。

添加元素


檢索元素 


BlockingQueue最重要的也就是關(guān)于阻塞等待的幾個(gè)方法，而這幾個(gè)方法正好可以用來實(shí)現(xiàn)生產(chǎn)-消費(fèi)的模型。

ArrayBlockingQueue

ArrayBlockingQueue 由數(shù)組支持的有界阻塞隊(duì)列，隊(duì)列基于數(shù)組實(shí)現(xiàn)，容量大小在創(chuàng)建 ArrayBlockingQueue 對(duì)象時(shí)已經(jīng)定義好。 此隊(duì)列按照先進(jìn)先出(FIFO)的原則對(duì)元素進(jìn)行排序。支持公平鎖和非公平鎖，默認(rèn)采用非公平鎖。

ArrayBlockingQueue 內(nèi)部由ReentrantLock來實(shí)現(xiàn)線程安全，由Condition的await和signal來實(shí)現(xiàn)等待喚醒的功能。它的數(shù)據(jù)結(jié)構(gòu)是數(shù)組，準(zhǔn)確的說是一個(gè)循環(huán)數(shù)組(可以類比一個(gè)圓環(huán))，所有的下標(biāo)在到達(dá)最大長度時(shí)自動(dòng)從0繼續(xù)開始。

深入理解ArrayBlockingQueue可以閱讀《阻塞隊(duì)列 — ArrayBlockingQueue源碼分析》


LinkedBlockingQueue

LinkedBlockingQueue 由鏈表節(jié)點(diǎn)支持的可選有界隊(duì)列，是一個(gè)基于鏈表的無界隊(duì)列(理論上有界)，隊(duì)列按照先進(jìn)先出的順序進(jìn)行排序。LinkedBlockingQueue不同于ArrayBlockingQueue，它如果不指定容量，默認(rèn)為 Integer.MAX_VALUE，也就是無界隊(duì)列。所以為了避免隊(duì)列過大造成機(jī)器負(fù)載或者內(nèi)存爆滿的情況出現(xiàn)，我們?cè)谑褂玫臅r(shí)候建議手動(dòng)傳一個(gè)隊(duì)列的大小。

LinkedBlockingQueue 內(nèi)部由單鏈表實(shí)現(xiàn)，只能從head取元素，從tail添加元素。添加元素和獲取元素都有獨(dú)立的鎖，也就是說LinkedBlockingQueue是讀寫分離的，讀寫操作可以并行執(zhí)行。LinkedBlockingQueue采用可重入鎖(ReentrantLock)來保證在并發(fā)情況下的線程安全。

向無限隊(duì)列添加元素的所有操作都將永遠(yuǎn)不會(huì)阻塞，[注意這里不是說不會(huì)加鎖保證線程安全]，因此它可以增長到非常大的容量。

使用無限 BlockingQueue 設(shè)計(jì)生產(chǎn)者 - 消費(fèi)者模型時(shí)最重要的是 消費(fèi)者應(yīng)該能夠像生產(chǎn)者向隊(duì)列添加消息一樣快地消費(fèi)消息。否則，內(nèi)存可能會(huì)填滿，然后就會(huì)得到一個(gè) OutOfMemory 異常。

深入理解LinkedBlockingQueue可以閱讀《阻塞隊(duì)列 — LinkedBlockingQueue源碼分析》


PriorityBlockingQueue

PriorityBlockingQueue 優(yōu)先級(jí)隊(duì)列，線程安全(添加、讀取都進(jìn)行了加鎖)、無界、讀阻塞的隊(duì)列，底層采用的堆結(jié)構(gòu)實(shí)現(xiàn)(二叉樹)，默認(rèn)是小根堆，最小的或者最大的元素會(huì)一直置頂，每次獲取都取最頂端的數(shù)據(jù)?？梢詫?shí)現(xiàn)優(yōu)先出隊(duì)。最特別的是它只有一個(gè)鎖，入隊(duì)操作永遠(yuǎn)成功，而出隊(duì)只有在空隊(duì)列的時(shí)候才會(huì)進(jìn)行線程阻塞?？梢哉f有一定的應(yīng)用場(chǎng)景吧，比如：有任務(wù)要執(zhí)行，可以對(duì)任務(wù)加一個(gè)優(yōu)先級(jí)的權(quán)重，這樣隊(duì)列會(huì)識(shí)別出來，對(duì)該任務(wù)優(yōu)先進(jìn)行出隊(duì)。

深入理解PriorityBlockingQueue可以閱讀《阻塞隊(duì)列 —PriorityBlockingQueue源碼分析》


DelayQueue

DelayQueue 由優(yōu)先級(jí)支持的、基于時(shí)間的調(diào)度隊(duì)列，內(nèi)部使用非線程安全的優(yōu)先隊(duì)列(PriorityQueue)實(shí)現(xiàn)，而無界隊(duì)列基于數(shù)組的擴(kuò)容實(shí)現(xiàn)。在創(chuàng)建元素時(shí)，可以指定多久才能從隊(duì)列中獲取當(dāng)前元素。只有延時(shí)期滿后才能從隊(duì)列中獲取元素。

深入理解DelayQueue可以閱讀《阻塞隊(duì)列 — DelayQueue源碼分析》

SynchronousQueue

SynchronousQueue 一個(gè)不存儲(chǔ)元素的阻塞隊(duì)列，每一個(gè) put 操作必須等待 take 操作，否則不能繼續(xù)添加元素。支持公平鎖和非公平鎖2種策略來訪問隊(duì)列。默認(rèn)是采用非公平性策略訪問隊(duì)列。公平性策略底層使用了類似隊(duì)列的數(shù)據(jù)結(jié)構(gòu)，而非公平策略底層使用了類似棧的數(shù)據(jù)結(jié)構(gòu)。SynchronousQueue的吞吐量高于LinkedBlockingQueue和ArrayBlockingQueue。

深入理解SynchronousQueue可以閱讀《阻塞隊(duì)列 — SynchronousQueue源碼分析》


LinkedTransferQueue

LinkedTransferQueue 是一個(gè)由鏈表結(jié)構(gòu)組成的無界阻塞傳輸隊(duì)列，它是一個(gè)很多隊(duì)列的結(jié)合體(ConcurrentLinkedQueue，LinkedBlockingQueue，SynchronousQueue)，在除了有基本阻塞隊(duì)列的功能(但是這個(gè)阻塞隊(duì)列沒有使用鎖)之外;隊(duì)列實(shí)現(xiàn)了TransferQueue接口重寫了transfer 和 tryTransfer 方法，這組方法和SynchronousQueue公平模式的隊(duì)列類似，具有匹配的功能。

深入理解LinkedTransferQueue可以閱讀《阻塞隊(duì)列 — LinkedTransferQueue源碼分析》


LinkedBlockingDeque

LinkedBlockingDeque 一個(gè)由于鏈表結(jié)構(gòu)組成的雙向阻塞隊(duì)列，隊(duì)列頭部和尾部都可以添加和移除元素，多線程并發(fā)時(shí)，可以將鎖的競(jìng)爭(zhēng)對(duì)多降到一半。

深入理解LinkedBlockingDeque可以閱讀《阻塞隊(duì)列 — LinkedBlockingDeque源碼分析》


DelayedWorkQueue

DelayedWorkQueue 也是一種設(shè)計(jì)為定時(shí)任務(wù)的延遲隊(duì)列，其實(shí)現(xiàn)原理和DelayQueue 基本一樣，核心數(shù)據(jù)結(jié)構(gòu)是二叉最小堆的優(yōu)先隊(duì)列，隊(duì)列滿時(shí)會(huì)自動(dòng)擴(kuò)容，不過是將優(yōu)先級(jí)隊(duì)列和DelayQueue的實(shí)現(xiàn)過程遷移到本身方法體中，從而可以在該過程當(dāng)中靈活的加入定時(shí)任務(wù)特有的方法調(diào)用。

深入理解DelayedWorkQueue可以閱讀《阻塞隊(duì)列 — DelayedWorkQueue源碼分析》


對(duì)比分析

LinkedBlockingQueue與ArrayBlockingQueue區(qū)別

  • 隊(duì)列大小有所不同，ArrayBlockingQueue是有界的初始化必須指定大小，而LinkedBlockingQueue可以是有界的也可以是無界的(Integer.MAX_VALUE)，對(duì)于后者而言，當(dāng)添加速度大于移除速度時(shí)，在無界的情況下，可能會(huì)造成內(nèi)存溢出等問題。
  • 數(shù)據(jù)存儲(chǔ)容器不同，ArrayBlockingQueue采用的是數(shù)組作為數(shù)據(jù)存儲(chǔ)容器，而LinkedBlockingQueue采用的則是以Node節(jié)點(diǎn)作為連接對(duì)象的鏈表。
  • 由于ArrayBlockingQueue采用的是數(shù)組的存儲(chǔ)容器，因此在插入或刪除元素時(shí)不會(huì)產(chǎn)生或銷毀任何額外的對(duì)象實(shí)例，而LinkedBlockingQueue則會(huì)生成一個(gè)額外的Node對(duì)象。這可能在長時(shí)間內(nèi)需要高效并發(fā)地處理大批量數(shù)據(jù)的時(shí)，對(duì)于GC可能存在較大影響。
  • 兩者的實(shí)現(xiàn)隊(duì)列添加或移除的鎖不一樣，ArrayBlockingQueue實(shí)現(xiàn)的隊(duì)列中的鎖是沒有分離的，即添加操作和移除操作采用的同一個(gè)ReenterLock鎖，而LinkedBlockingQueue實(shí)現(xiàn)的隊(duì)列中的鎖是分離的，其添加采用的是putLock，移除采用的則是takeLock，這樣能大大提高隊(duì)列的吞吐量，也意味著在高并發(fā)的情況下生產(chǎn)者和消費(fèi)者可以并行地操作隊(duì)列中的數(shù)據(jù)，以此來提高整個(gè)隊(duì)列的并發(fā)性能。

LinkedTransferQueue和SynchronousQueue(公平模式)區(qū)別

  • LinkedTransferQueue 和SynchronousQueue 其實(shí)基本是差不多的，兩者都是無鎖帶阻塞功能的隊(duì)列，都是使用的雙重隊(duì)列;
  • SynchronousQueue 通過內(nèi)部類Transferer 來實(shí)現(xiàn)公平和非公平隊(duì)列，在LinkedTransferQueue 中沒有公平與非公平的區(qū)分;
  • LinkedTransferQueue 實(shí)現(xiàn)了TransferQueue接口，該接口定義的是帶阻塞操作的操作，相比SynchronousQueue 中的Transferer 功能更豐富。
  • SynchronousQueue 中放數(shù)據(jù)操作和取數(shù)據(jù)操作都是阻塞的，當(dāng)隊(duì)列中的操作和本次操作不匹配時(shí)，線程會(huì)阻塞，直到匹配的操作到來。LinkedTransferQueue 是無界隊(duì)列，放數(shù)據(jù)操作不會(huì)阻塞，取數(shù)據(jù)操作如果沒有匹配操作可能會(huì)阻塞，通過參數(shù)決定是否阻塞(ASYNC,SYNC,NOW,TIMED)。

LinkedBlockingDeque與LinkedList區(qū)別

  1. package com.niuh.deque; 
  2.  
  3. import java.util.Iterator; 
  4. import java.util.LinkedList; 
  5. import java.util.Queue; 
  6. import java.util.concurrent.LinkedBlockingDeque; 
  7.  
  8. /* 
  9.  *   LinkedBlockingDeque是“線程安全”的隊(duì)列，而LinkedList是非線程安全的。 
  10.  * 
  11.  *   下面是“多個(gè)線程同時(shí)操作并且遍歷queue”的示例 
  12.  *   (1) 當(dāng)queue是LinkedBlockingDeque對(duì)象時(shí)，程序能正常運(yùn)行。 
  13.  *   (2) 當(dāng)queue是LinkedList對(duì)象時(shí)，程序會(huì)產(chǎn)生ConcurrentModificationException異常。 
  14.  * 
  15.  */ 
  16. public class LinkedBlockingDequeRunner { 
  17.  
  18.     // TODO: queue是LinkedList對(duì)象時(shí)，程序會(huì)出錯(cuò)。 
  19.     // private static Queue<String> queue = new LinkedList<String>(); 
  20.     private static Queue<String> queue = new LinkedBlockingDeque<String>(); 
  21.  
  22.     public static void main(String[] args) { 
  23.  
  24.         // 同時(shí)啟動(dòng)兩個(gè)線程對(duì)queue進(jìn)行操作！ 
  25.         new MyThread("A").start(); 
  26.         new MyThread("B").start(); 
  27.     } 
  28.  
  29.     private static void printAll() { 
  30.         String value; 
  31.         Iterator iter = queue.iterator(); 
  32.         while (iter.hasNext()) { 
  33.             value = (String) iter.next(); 
  34.             System.out.print(value + ", "); 
  35.         } 
  36.         System.out.println(); 
  37.     } 
  38.  
  39.     private static class MyThread extends Thread { 
  40.         MyThread(String name) { 
  41.             super(name); 
  42.         } 
  43.  
  44.         @Override 
  45.         public void run() { 
  46.             int i = 0; 
  47.             while (i++ < 6) { 
  48.                 // “線程名” + "-" + "序號(hào)" 
  49.                 String val = Thread.currentThread().getName() + i; 
  50.                 queue.add(val); 
  51.                 // 通過“Iterator”遍歷queue。 
  52.                 printAll(); 
  53.             } 
  54.         } 
  55.     } 
  56. } 

 輸出結(jié)果： 

  1. A1,  
  2. A1, A2,  
  3. A1, A2, A3,  
  4. A1, A2, A3, A4,  
  5. A1, A2, A3, A4, A5,  
  6. A1, A2, A3, A4, A5, A6,  
  7. A1, A2, A3, A4, A5, A6, B1,  
  8. A1, A2, A3, A4, A5, A6, B1, B2,  
  9. A1, A2, A3, A4, A5, A6, B1, B2, B3,  
  10. A1, A2, A3, A4, A5, A6, B1, B2, B3, B4,  
  11. A1, A2, A3, A4, A5, A6, B1, B2, B3, B4, B5,  
  12. A1, A2, A3, A4, A5, A6, B1, B2, B3, B4, B5, B6,  

 結(jié)果說明：示例程序中，啟動(dòng)兩個(gè)線程(線程A和線程B)分別對(duì)LinkedBlockingDeque進(jìn)行操作:

以線程A而言，它會(huì)先獲取“線程名”+“序號(hào)”，然后將該字符串添加到LinkedBlockingDeque中;

接著，遍歷并輸出LinkedBlockingDeque中的全部元素。

線程B的操作和線程A一樣，只不過線程B的名字和線程A的名字不同。

當(dāng)queue是LinkedBlockingDeque對(duì)象時(shí)，程序能正常運(yùn)行。

如果將queue改為LinkedList時(shí)，程序會(huì)產(chǎn)生ConcurrentModificationException異常。

BlockingQueue應(yīng)用

多線程生產(chǎn)者-消費(fèi)者示例

接下來我們創(chuàng)建一個(gè)由兩部分組成的程序： 生產(chǎn)者 ( Producer ) 和消費(fèi)者 ( Consumer ) 。

生產(chǎn)者(Producer)

生產(chǎn)者將生成一個(gè) 0 到 100 的隨機(jī)數(shù)(十全大補(bǔ)丸的編號(hào))，并將該數(shù)字放在 BlockingQueue 中。我們將創(chuàng)建 16 個(gè)線程(潘金蓮)用于生成隨機(jī)數(shù)并使用 put() 方法阻塞，直到隊(duì)列中有可用空間。

需要記住的重要一點(diǎn)是，我們需要阻止我們的消費(fèi)者線程無限期地等待元素出現(xiàn)在隊(duì)列中。

從生產(chǎn)者(潘金蓮)向消費(fèi)者(武大郎)發(fā)出信號(hào)的好方法是，不需要處理消息，而是發(fā)送稱為毒 ( poison ) 丸 ( pill ) 的特殊消息。 我們需要發(fā)送盡可能多的毒 ( poison ) 丸 ( pill ) ，因?yàn)槲覀冇邢M(fèi)者(武大郎)。然后當(dāng)消費(fèi)者從隊(duì)列中獲取特殊的毒 ( poison ) 丸 ( pill )消息時(shí)，它將優(yōu)雅地完成執(zhí)行。

以下生產(chǎn)者的代碼：

  1. package com.niuh.queue; 
  2.  
  3. import lombok.extern.slf4j.Slf4j; 
  4.  
  5. import java.util.concurrent.BlockingQueue; 
  6. import java.util.concurrent.ThreadLocalRandom; 
  7.  
  8. /** 
  9.  * 生產(chǎn)者（Producer） 
  10.  **/ 
  11. @Slf4j 
  12. public class NumbersProducer implements Runnable { 
  13.     private BlockingQueue<Integer> numbersQueue; 
  14.     private final int poisonPill; 
  15.     private final int poisonPillPerProducer; 
  16.  
  17.     public NumbersProducer(BlockingQueue<Integer> numbersQueue, int poisonPill, int poisonPillPerProducer) { 
  18.         this.numbersQueue = numbersQueue; 
  19.         this.poisonPill = poisonPill; 
  20.         this.poisonPillPerProducer = poisonPillPerProducer; 
  21.     } 
  22.  
  23.     public void run() { 
  24.         try { 
  25.             generateNumbers(); 
  26.         } catch (InterruptedException e) { 
  27.             Thread.currentThread().interrupt(); 
  28.         } 
  29.     } 
  30.  
  31.     private void generateNumbers() throws InterruptedException { 
  32.         for (int i = 0; i < 100; i++) { 
  33.             numbersQueue.put(ThreadLocalRandom.current().nextInt(100)); 
  34.             log.info("潘金蓮-{}號(hào),給武大郎的泡藥！", Thread.currentThread().getId()); 
  35.         } 
  36.  
  37.         /*while (true) { 
  38.             numbersQueue.put(ThreadLocalRandom.current().nextInt(100)); 
  39.             if (false) { 
  40.                 break; 
  41.             } 
  42.         }*/ 
  43.  
  44.         for (int j = 0; j < poisonPillPerProducer; j++) { 
  45.             numbersQueue.put(poisonPill); 
  46.             log.info("潘金蓮-{}號(hào),往武大郎的藥里放入第{}顆毒丸！", Thread.currentThread().getId(), j + 1); 
  47.         } 
  48.     } 
  49. } 

 我們的生成器構(gòu)造函數(shù)將 BlockingQueue 作為參數(shù)，用于協(xié)調(diào)生產(chǎn)者和使用者之間的處理，我們看到方法generateNumbers() 將 100 個(gè)元素(生產(chǎn)100副藥給武大郎吃)放入隊(duì)列中。它還需要有毒 ( poison ) 丸 ( pill ) (潘金蓮給武大郎下毒)消息，以便知道在執(zhí)行完成時(shí)放入隊(duì)列的消息類型。該消息需要將 poisonPillPerProducer 次放入隊(duì)列中。

消費(fèi)者(Consumer)

每個(gè)消費(fèi)者將使用 take() 方法從 BlockingQueue 獲取一個(gè)元素，因此它將阻塞，直到隊(duì)列中有一個(gè)元素。從隊(duì)列中取出一個(gè) Integer 后，它會(huì)檢查該消息是否是毒 ( poison ) 丸 ( pill )(武大郎看潘金蓮有沒有下毒) ，如果是，則完成一個(gè)線程的執(zhí)行。否則，它將在標(biāo)準(zhǔn)輸出上打印出結(jié)果以及當(dāng)前線程的名稱。

  1. package com.niuh.queue; 
  2.  
  3. import lombok.extern.slf4j.Slf4j; 
  4.  
  5. import java.util.concurrent.BlockingQueue; 
  6.  
  7. /** 
  8.  * 消費(fèi)者（Consumer） 
  9.  **/ 
  10. @Slf4j 
  11. public class NumbersConsumer implements Runnable { 
  12.     private BlockingQueue<Integer> queue; 
  13.     private final int poisonPill; 
  14.  
  15.     public NumbersConsumer(BlockingQueue<Integer> queue, int poisonPill) { 
  16.         this.queue = queue; 
  17.         this.poisonPill = poisonPill; 
  18.     } 
  19.  
  20.     public void run() { 
  21.         try { 
  22.             while (true) { 
  23.                 Integer number = queue.take(); 
  24.                 if (number.equals(poisonPill)) { 
  25.                     return; 
  26.                 } 
  27.                 log.info("武大郎-{}號(hào),喝藥-編號(hào):{}", Thread.currentThread().getId(), number); 
  28.             } 
  29.         } catch (InterruptedException e) { 
  30.             Thread.currentThread().interrupt(); 
  31.         } 
  32.     } 
  33. } 

 需要注意的重要事項(xiàng)是隊(duì)列的使用。與生成器構(gòu)造函數(shù)中的相同，隊(duì)列作為參數(shù)傳遞。我們可以這樣做，是因?yàn)?BlockingQueue 可以在線程之間共享而無需任何顯示同步。

驗(yàn)證測(cè)試

既然我們有生產(chǎn)者和消費(fèi)者，我們就可以開始我們的計(jì)劃。我們需要定義隊(duì)列的容量，并將其設(shè)置為 10個(gè)元素。 我們創(chuàng)建4 個(gè)生產(chǎn)者線程，并且創(chuàng)建等于可用處理器數(shù)量的消費(fèi)者線程：

  1. package com.niuh.queue; 
  2.  
  3. import java.util.concurrent.ArrayBlockingQueue; 
  4. import java.util.concurrent.BlockingQueue; 
  5. import java.util.concurrent.LinkedBlockingQueue; 
  6.  
  7. /** 
  8.  * 多線程生產(chǎn)者-消費(fèi)者示例 
  9.  **/ 
  10. public class Main { 
  11.  
  12.     public static void main(String[] args) { 
  13.         int BOUND = 10; 
  14.         int N_PRODUCERS = 16; 
  15.         int N_CONSUMERS = Runtime.getRuntime().availableProcessors(); //=8 
  16.         int poisonPill = Integer.MAX_VALUE; 
  17.         int poisonPillPerProducer = N_CONSUMERS / N_PRODUCERS; // =0 
  18.         int mod = N_CONSUMERS % N_PRODUCERS;//0+8=8 
  19.  
  20.         BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(BOUND); 
  21.  
  22.         //潘金蓮給武大郎熬藥 
  23.         for (int i = 1; i < N_PRODUCERS; i++) { 
  24.             new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer)).start(); 
  25.         } 
  26.  
  27.         //武大郎開始喝藥 
  28.         for (int j = 0; j < N_CONSUMERS; j++) { 
  29.             new Thread(new NumbersConsumer(queue, poisonPill)).start(); 
  30.         } 
  31.  
  32.         try { 
  33.             Thread.sleep(5000); 
  34.         } catch (InterruptedException e) { 
  35.             e.printStackTrace(); 
  36.         } 
  37.  
  38.         //潘金蓮開始投毒，武大郎喝完毒藥GG 
  39.         new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer + mod)).start(); 
  40.     } 
  41.  
  42. } 

 BlockingQueue 是使用具有容量的構(gòu)造創(chuàng)建的。我們正在創(chuàng)造 4 個(gè)生產(chǎn)者和 N 個(gè)消費(fèi)者(武大郎)。我們將我們的毒 ( poison ) 丸 ( pill )消息指定為 Integer.MAX_VALUE，因?yàn)槲覀兊纳a(chǎn)者在正常工作條件下永遠(yuǎn)不會(huì)發(fā)送這樣的值。這里要注意的最重要的事情是 BlockingQueue 用于協(xié)調(diào)它們之間的工作。

PS：以上代碼提交在 Github ：

https://github.com/Niuh-Study/niuh-juc-final.git

 

責(zé)任編輯：姜華 來源： 今日頭條
BlockingQueue
相關(guān)推薦

2020-07-07 07:37:36

Integer源碼Java

2025-01-14 00:00:00

Blocking隊(duì)列元素

2022-09-04 12:43:03

算法裁員Meta

2022-02-14 12:04:43

前綴SpringJpa

2021-07-05 18:05:40

SpringBean方法

2020-11-03 06:57:10

MyBatis數(shù)據(jù)庫

2019-06-14 08:48:46

Tomcat日志SpringBoot

2019-09-18 15:20:16

MyBatisSQL數(shù)據(jù)庫

2015-07-20 15:26:56

WiFi感知

2022-06-22 12:54:58

人工智能機(jī)器人元宇宙

2021-06-07 14:05:53

物聯(lián)網(wǎng)IOT物聯(lián)網(wǎng)技術(shù)

2023-06-30 08:27:20

2021-12-08 08:30:55

Java AQS機(jī)制 Java 基礎(chǔ)

2022-03-11 15:40:49

PaaS云服務(wù)

2012-01-10 14:43:48

程序員

2011-04-15 10:03:57

2020-10-12 13:37:19

安全黑客技術(shù)

2021-03-18 14:35:36

爬蟲技術(shù)開發(fā)

2019-09-16 13:51:34

物聯(lián)網(wǎng)平臺(tái)物聯(lián)網(wǎng)IOT

2021-07-08 09:17:07

物聯(lián)網(wǎng)人工智能IoT
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)