JavaSE多線程阻塞隊列實現(xiàn)代碼
一、阻塞隊列
阻塞隊列：是一種特殊的隊列，也有先進先出的特性。它是一種線程安全的隊列。
有以下兩個特性：
- 當(dāng)隊列滿的時候, 繼續(xù)入隊列就會阻塞, 直到有其他線程從隊列中取走元素。
- 當(dāng)隊列空的時候, 繼續(xù)出隊列也會阻塞, 直到有其他線程往隊列中插入元素。
阻塞隊列的一個重要應(yīng)用場景就是：實現(xiàn)生產(chǎn)者消費模型。
1.1 生產(chǎn)者消費者模型
生產(chǎn)者消費者模型：是多線程編程中的一種典型的編碼技巧。用來降低生產(chǎn)者與消費者之間的耦合度。生產(chǎn)者和消費者之間的交易場所就是一個阻塞隊列。
這樣的模型的優(yōu)勢有以下兩個：
- 解耦合，降低代碼耦合度：
像如果是A B兩個服務(wù)器，之間直接進行交互，如果對A或者B中的數(shù)據(jù)進行修改操作，大概率就會影響到另一個服務(wù)器。而使用阻塞隊列作為交易平臺，我們修改服務(wù)器的數(shù)據(jù)時，由于阻塞隊列中的結(jié)構(gòu)固定，兩個服務(wù)器之間的耦合度就降低。 - 削峰削谷：
在服務(wù)器中，波峰就是請求量高的時候，波谷就是請求量低的時候。
如果是AB兩個服務(wù)器之間進行交互，當(dāng)上游服務(wù)器A經(jīng)歷波峰，將大量請求傳給服務(wù)器B的時候，服務(wù)器就有可能掛掉。
因為上游服務(wù)器，干的活簡單，消耗的資源少；而下游服務(wù)器，干的活復(fù)雜，消耗的的資源就多。
但是如果我們將阻塞隊列作為交易平臺，那么服務(wù)器B就可以依據(jù)自己的節(jié)奏從隊列中拿請求。
但是這樣的模型也會付出代價：
- 引入阻塞隊列之后整體結(jié)構(gòu)會更加復(fù)雜。比如本來是AB兩個服務(wù)器之間的交互，但引入一個作為阻塞隊列的服務(wù)器（這種稱為消息隊列），就需要部署這個服務(wù)器，還要與AB實現(xiàn)交互。
- 效率也會有影響。
1.2 Java提供的阻塞隊列
提供了一個BlockingDeque的接口（需要導(dǎo)java.util.concurrent.BlockingQueue包）：
主要使用下面3個實現(xiàn)了BlockingDeque接口的來實例化阻塞隊列：
- 鏈表實現(xiàn)的，LinkedBlockingDeque（需要導(dǎo)
java.util.concurrent.LinkedBlockingDeque包）:
- 數(shù)組實現(xiàn)的，ArrayBlockingDeque需要導(dǎo)`java.util.concurrent.ArrayBlockingDeque包）:
- 小根堆實現(xiàn)的，PriorityBlockingDeque需要導(dǎo)
java.util.concurrent.PriorityBlockingDeque包）:
在阻塞隊列中我們雖然可以使用隊列中常用的出隊列入隊列方法，但是那些方法不帶阻塞效果。帶阻塞效果的入隊列方法是put，出隊列方法是take，這兩個方法都會拋出InterruptedException異常。
1.3 實現(xiàn)一個簡單生產(chǎn)者消費者模型
實現(xiàn)一個簡單的生產(chǎn)者消費者模型：
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
public class Demo {
public static void main(String[] args) {
BlockingQueue<Integer> blockingQueue = new LinkedBlockingDeque<>(1000);
Thread producer = new Thread(() -> {
int i = 0;
while(true) {
try {
blockingQueue.put(i++);
System.out.println(i + "入隊列成功");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread consumer = new Thread(() -> {
while(true){
try {
int x = blockingQueue.take();
System.out.println(x + "出隊列成功");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
producer.start();
consumer.start();
}
}
二、自己實現(xiàn)阻塞隊列
我們使用數(shù)組來實現(xiàn)一個循環(huán)隊列。
不知道循環(huán)隊列的實現(xiàn)的可以看下面這個鏈接：隊列
2.1 成員變量
- 使用capacity代表數(shù)組的最大長度；
- 使用size表示數(shù)組中元素的個數(shù)；
- head表示隊頭元素的下標(biāo)；
- tail表示隊尾元素的下標(biāo)。
private int capacity = 0xffff;
private String[] elem ;//存儲數(shù)組
private int size;//存儲元素個數(shù)
private int head;//隊頭
private int tail;//隊尾
2.2 構(gòu)造方法
提供兩個構(gòu)造方法：
- 使用默認(rèn)最大值初始化數(shù)組；
- 使用傳的參初始化數(shù)組。
public MyBlockingQueue(int capacity) {
this.capacity = capacity;
elem = new String[this.capacity];
}
public MyBlockingQueue() {
elem = new String[this.capacity];
}
2.3 put方法
由于put和take方法都涉及到修改判斷等操作，為避免原子性問題帶來線程安全問題對該這些操作都要加鎖。
在put方法中我們需要在隊列滿的時候發(fā)生阻塞，使用wait來等待。而在Java官方文檔給出了建議我們使用循環(huán)語句來使用wait。
因為wait是除了notify喚醒外，還有可能被interrupt方法喚醒拋出異常，如果只要if，不用while，拋出異常后就會繼續(xù)執(zhí)行下面的邏輯，帶來bug。而使用循環(huán)就不會，拋出異常后，會再次判斷循環(huán)條件。
最后在入隊成功后發(fā)出一個通知notify來喚醒由于隊列空而阻塞等待的線程。
public void put(String s) throws InterruptedException {
synchronized (this) {
while(size == elem.length) {
this.wait();
}
elem[tail] = s;
tail = (tail+1) % elem.length;
size++;
this.notify();
}
}
2.4 take方法
當(dāng)隊列為空的時候，跟put一樣使用wait來阻塞。
最后在出隊成功后發(fā)出一個通知notify來喚醒由于隊列滿而阻塞等待的線程。
public String take() throws InterruptedException {
synchronized (this) {
while(size == 0) {
this.wait();
}
String ret = elem[head];
head = (head+1) % capacity;
size--;
this.notify();
return ret;
}
}
2.5 最終代碼
最終我們自己實現(xiàn)的一個簡單的阻塞隊列就如下：
public class MyBlockingQueue {
private int capacity = 0xffff;
private String[] elem ;//存儲數(shù)組
private int size;//存儲元素個數(shù)
private int head;//隊頭
private int tail;//隊尾
public MyBlockingQueue(int length) {
elem = new String[length];
}
public MyBlockingQueue() {
elem = new String[this.capacity];
}
public void put(String s) throws InterruptedException {
synchronized (this) {
while(size == elem.length) {
this.wait();
}
elem[tail] = s;
tail = (tail+1) % capacity;
size++;
this.notify();
}
}
public String take() throws InterruptedException {
synchronized (this) {
while(size == 0) {
this.wait();
}
String ret = elem[head];
head = (head+1) % capacity;
size--;
this.notify();
return ret;
}
}
}
總結(jié)
到此這篇關(guān)于JavaSE多線程阻塞隊列實現(xiàn)的文章就介紹到這了,更多相關(guān)JavaSE多線程阻塞隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
java調(diào)用python腳本引入第三方庫失敗的實現(xiàn)
本文主要介紹了java調(diào)用python腳本引入第三方庫失敗的實現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-07-07
java后臺調(diào)用HttpURLConnection類模擬瀏覽器請求實例(可用于接口調(diào)用)
這篇文章主要介紹了java后臺調(diào)用HttpURLConnection類模擬瀏覽器請求實例,該實例可用于接口調(diào)用,具有一定的實用價值,需要的朋友可以參考下2014-10-10
本篇文章主要介紹了springMVC攔截器和過濾器總結(jié)，可以用來對訪問的url進行攔截處理，有興趣的可以了解一下。2017-01-01
Java利用Netty時間輪實現(xiàn)延時任務(wù)
時間輪是一種可以執(zhí)行定時任務(wù)的數(shù)據(jù)結(jié)構(gòu)和算法。本文將為大家詳細(xì)講解一下Java如何利用Netty時間輪算法實現(xiàn)延時任務(wù)，感興趣的小伙伴可以了解一下2022-08-08
SpringBoot+mybatis+thymeleaf實現(xiàn)登錄功能示例
這篇文章主要介紹了SpringBoot+mybatis+thymeleaf實現(xiàn)登錄功能示例，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-07-07
這篇文章主要為大家詳細(xì)介紹了Java實現(xiàn)短信驗證碼功能的完整代碼,文中的示例代碼講解詳細(xì),具有一定的借鑒價值,感興趣的小伙伴可以了解下2025-12-12
這篇文章介紹了SpringMVC的異常處理和攔截器功能,在異常處理部分,詳細(xì)說明了如何通過DispatcherServlet捕獲并處理異常,本文通過實例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友參考下吧2025-09-09
Spring Security 中細(xì)化權(quán)限粒度的方法
這篇文章主要介紹了Spring Security 中細(xì)化權(quán)限粒度的方法,本文給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2020-09-09
這篇文章主要為大家介紹了RocketMQ?源碼分析Broker消息刷盤服務(wù)示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-05-05
SpringBoot2整合JTA組件實現(xiàn)多數(shù)據(jù)源事務(wù)管理
這篇文章主要介紹了SpringBoot2整合JTA組件實現(xiàn)多數(shù)據(jù)源事務(wù)管理，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03