java之阻塞隊列BlockingQueue解析
更新時間：2023年10月24日 10:18:11 作者：強強愛java
這篇文章主要介紹了java之阻塞隊列BlockingQueue解析,阻塞隊列是一種線程安全的數(shù)據(jù)結(jié)構(gòu),用于在多線程環(huán)境下進行數(shù)據(jù)交換,它提供了一種阻塞的機制,當(dāng)隊列為空時,消費者線程將被阻塞,直到隊列中有數(shù)據(jù)可供消費,需要的朋友可以參考下
一、阻塞隊列概念
阻塞隊列，顧名思義，首先它是一個隊列（先進先出），而一個阻塞隊列在數(shù)據(jù)結(jié)構(gòu)所起到的作用大致如下圖：
- 線程1往阻塞隊列中添加元素，而線程2從阻塞隊列中移除元素
- 當(dāng)阻塞隊列是空是，從隊列中獲取元素的操作會被阻塞
- 當(dāng)阻塞隊列是滿時，從隊列中添加元素的操作會被阻塞
- 試圖從空的阻塞隊列中獲取元素的線程將會被阻塞，直到其他的線程往空的隊列插入新的元素。
- 試圖網(wǎng)已滿的阻塞隊列中添加新元素的線程同樣會被阻塞，直到其他的線程從列中移除一個或者多個元素或者完全清空隊列后使隊列重新變得空閑起來并后續(xù)新增
二、阻塞隊列的好處
1.在多線程領(lǐng)域：所謂阻塞，在某些情況下會掛起線程，一旦滿足條件，被掛起的線程又會自動被喚醒。
2.我們不需要關(guān)心什么時候需要阻塞線程，什么時候需要喚醒線程，因為這一切BlockingQueue（阻塞隊列）都給你一手包辦了
PS：在concurrent包發(fā)布以前，在多線程環(huán)境下，我們每個程序員都必須自己控制這些細節(jié)，尤其還要兼顧效率和線程安全，而這回給我們程序帶來不小的復(fù)雜度
三、阻塞隊列種類
- ArrayBlockingQueue：由數(shù)據(jù)結(jié)構(gòu)組成的有界阻塞隊列
- LinkedBlockingQueue：由鏈表結(jié)構(gòu)組成的有界（但大小默認(rèn)值為 Integer.MAX_VALUE )阻塞隊列
- PriorityBlockingQueue：支持優(yōu)先級排序的無界阻塞隊列
- DelayQueue：使用優(yōu)先級隊列實現(xiàn)的延遲無界阻塞隊列
- SynchronousQueue：不存儲元素的阻塞隊列，也即單個元素的隊列
- LinkedTransferQueue：由鏈表結(jié)構(gòu)組成的無界阻塞隊列
- LinkedBlockingDeque：由歷覽表結(jié)構(gòu)組成的雙向阻塞隊列
PS：重點掌握ArrayBlockingQueue、LinkedBlockingQueue、SychronousQueue三種
四、BlockingQueue的核心方法
| 方法類型 | 拋出異常 | 特殊值 | 一直阻塞 | 超時退出 |
| 插入 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 移除 | remove() | poll() | take | poll(time,unit) |
| 檢查 | element() | peek() | 不可用 | 不可用 |
說明：
| 方法類型 | 情況 |
| 拋出異常 | 當(dāng)阻塞隊列滿時，再往隊列中add會拋 IllegalStateException: Queue full； 當(dāng)阻塞隊列空時，再從隊列里remove會拋 NoSuchElementException |
| 特殊值 | offer(e)插入方法，成功true失敗false poll() 移除方法，成功返回出隊列的元素，隊列里沒有就返回null |
| 一直阻塞 | 當(dāng)阻塞隊列滿時，生產(chǎn)者線程繼續(xù)往隊列里put元素，隊列會一直阻塞線程直到take數(shù)據(jù)或響應(yīng)中斷退出 當(dāng)阻塞隊列空時，消費者線程試圖從隊列take元素，隊列會一直阻塞消費者線程直到隊列可用 |
| 超時退出 | 當(dāng)阻塞隊列滿時，隊列會阻塞生產(chǎn)者線程一定時間，超過限時后生產(chǎn)者線程會退出 |
五、示例代碼
1.使用SychronousQueue隊列
package com.jian8.juc.queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
/*
* 阻塞隊列SynchronousQueue演示
* */
public class SynchronousQueueDemo {
public static void main(String[] args) {
BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + "\t put 1");
blockingQueue.put("1");
System.out.println(Thread.currentThread().getName() + "\t put 2");
blockingQueue.put("2");
System.out.println(Thread.currentThread().getName() + "\t put 3");
blockingQueue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "AAA").start();
new Thread(() -> {
try {
TimeUnit.SECONDS.sleep(2);
System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());
TimeUnit.SECONDS.sleep(2);
System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());
TimeUnit.SECONDS.sleep(2);
System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "BBB").start();
}
}
運行結(jié)果：
2.傳統(tǒng)版生產(chǎn)者消費者模式
package com.jian8.juc.queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 一個初始值為零的變量，兩個線程對其交替操作，一個加1一個減1，來5輪
* 1. 線程 操作 資源類
* 2. 判斷 干活 通知
* 3. 防止虛假喚起機制
*/
public class ProdConsumer_TraditionDemo {
public static void main(String[] args) {
ShareData shareData = new ShareData();
for (int i = 1; i <= 5; i++) {
new Thread(() -> {
try {
shareData.increment();
} catch (Exception e) {
e.printStackTrace();
}
}, "ProductorA " + i).start();
}
for (int i = 1; i <= 5; i++) {
new Thread(() -> {
try {
shareData.decrement();
} catch (Exception e) {
e.printStackTrace();
}
}, "ConsumerA " + i).start();
}
for (int i = 1; i <= 5; i++) {
new Thread(() -> {
try {
shareData.increment();
} catch (Exception e) {
e.printStackTrace();
}
}, "ProductorB " + i).start();
}
for (int i = 1; i <= 5; i++) {
new Thread(() -> {
try {
shareData.decrement();
} catch (Exception e) {
e.printStackTrace();
}
}, "ConsumerB " + i).start();
}
}
}
//資源類
class ShareData {
private int number = 0;
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
//生產(chǎn)
public void increment() throws Exception {
lock.lock();
try {
//1.判斷
while (number != 0) {
//等待中，不能生產(chǎn)
condition.await();
}
//2.干活
number++;
System.out.println(Thread.currentThread().getName() + "\t" + number);
//3.通知
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
//消費
public void decrement() throws Exception {
lock.lock();
try {
//1.判斷
while (number == 0) {
//等待中，不能消費
condition.await();
}
//2.消費
number--;
System.out.println(Thread.currentThread().getName() + "\t" + number);
//3.通知
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
運行結(jié)果（部分）：
3.阻塞隊列版（以ArrayBlockingQueue為例）生產(chǎn)者消費者模式
package com.jian8.juc.queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class ProdConsumer_BlockQueueDemo {
public static void main(String[] args) {
MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "\t生產(chǎn)線程啟動");
try {
myResource.myProd();
} catch (Exception e) {
e.printStackTrace();
}
}, "Prod").start();
new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "\t消費線程啟動");
try {
myResource.myConsumer();
} catch (Exception e) {
e.printStackTrace();
}
}, "Consumer").start();
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("5s后main叫停，線程結(jié)束");
try {
myResource.stop();
} catch (Exception e) {
e.printStackTrace();
}
}
}
//資源類
class MyResource {
private volatile boolean flag = true;//默認(rèn)開啟，進行“生產(chǎn)+消費”活動
private AtomicInteger atomicInteger = new AtomicInteger();
BlockingQueue<String> blockingQueue = null;//消息隊列
public MyResource(BlockingQueue<String> blockingQueue) {
this.blockingQueue = blockingQueue;
System.out.println(blockingQueue.getClass().getName());
}
//生產(chǎn)
public void myProd() throws Exception {
String data = null;
boolean retValue;
while (flag) {
data = atomicInteger.incrementAndGet() + "";
//超過2s沒生產(chǎn)成功，退出生產(chǎn)
retValue = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
if (retValue) {
System.out.println(Thread.currentThread().getName() + "\t插入隊列" + data + "成功");
} else {
System.out.println(Thread.currentThread().getName() + "\t插入隊列" + data + "失敗");
}
TimeUnit.SECONDS.sleep(1);
}
System.out.println(Thread.currentThread().getName() + "\t大老板叫停了，flag=false，生產(chǎn)結(jié)束");
}
//消費
public void myConsumer() throws Exception {
String result = null;
while (flag) {
//超過2s沒從隊列獲取數(shù)據(jù)，消費退出
result = blockingQueue.poll(2, TimeUnit.SECONDS);
if (null == result || result.equalsIgnoreCase("")) {
flag = false;
System.out.println(Thread.currentThread().getName() + "\t超過2s沒有取到蛋糕，消費退出");
System.out.println();
return;
}
System.out.println(Thread.currentThread().getName() + "\t消費隊列" + result + "成功");
}
}
public void stop() throws Exception {
flag = false;
}
}
運行結(jié)果：
到此這篇關(guān)于java之阻塞隊列BlockingQueue解析的文章就介紹到這了,更多相關(guān)java阻塞隊列BlockingQueue內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Springboot如何實現(xiàn)代理服務(wù)器問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-06-06
Java?靜態(tài)機制之靜態(tài)變量、方法、代碼塊與類加載深入理解
本文全面解析了Java中static關(guān)鍵字的特性與應(yīng)用,靜態(tài)機制使成員與類本身關(guān)聯(lián),包括靜態(tài)變量（存儲在方法區(qū),所有對象共享）、靜態(tài)方法（可通過類名直接調(diào)用,不能訪問實例成員）和靜態(tài)代碼塊（類加載時執(zhí)行）,感興趣的朋友跟隨小編一起看看吧2025-11-11
這篇文章主要介紹了JAVA設(shè)計模式之命令模式的的相關(guān)資料，文中示例代碼非常詳細，供大家參考和學(xué)習(xí)，感興趣的朋友可以了解2021-11-11
深入學(xué)習(xí)java8?中的CompletableFuture
本文主要介紹了java8中的CompletableFuture,CompletableFuture實現(xiàn)了CompletionStage接口和Future接口,前者是對后者的一個擴展，增加了異步回調(diào)、流式處理、多個Future組合處理的能力，使Java在處理多任務(wù)的協(xié)同工作時更加順暢便利,下文需要的朋友可以參考一下2022-05-05
這篇文章主要介紹了如何實現(xiàn)Spring?Event(異步事件)問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-02-02
這篇文章主要介紹了SpringBoot自定義/error路徑失效的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-01-01
這篇文章主要為大家詳細介紹了Ubuntu16.04安裝部署solr7的圖文詳細教程，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-07-07
Spring Security和Shiro的相同點與不同點整理
在本篇文章里小編給大家整理的是關(guān)于Spring Security和Shiro的相同不同點整理，需要的朋友們可以參考下。2020-02-02
這篇文章主要介紹了Java多線程中的concurrent簡析,java.util.concurrent包提供了很多有用的類,方便我們進行并發(fā)程序的開發(fā),本文將會挑選其中常用的一些類來進行大概的說明,需要的朋友可以參考下2023-09-09
這篇文章主要介紹了SpringBoot中的yaml配置文件問題，本文給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2022-06-06