您好,登錄后才能下訂單哦!
這篇“Java中阻塞隊列怎么使用”文章的知識點大部分人都不太理解,所以小編給大家總結了以下內容,內容詳細,步驟清晰,具有一定的借鑒價值,希望大家閱讀完這篇文章能有所收獲,下面我們一起來看看這篇“Java中阻塞隊列怎么使用”文章吧。
阻塞隊列是一種特殊的隊列,和數(shù)據(jù)結構中普通的隊列一樣,也遵守先進先出的原則同時,阻塞隊列是一種能保證線程安全的數(shù)據(jù)結構,并且具有以下兩種特性:當隊列滿的時候,繼續(xù)向隊列中插入元素就會讓隊列阻塞,直到有其他線程從隊列中取走元素;當隊列為空的時候,繼續(xù)出隊列也會讓隊列阻塞,直到有其他線程往隊列中插入元素
補充:線程阻塞的意思指代碼此時不會被執(zhí)行,即操作系統(tǒng)在此時不會把這個線程調度到CPU上去執(zhí)行了
import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.BlockingDeque; public class Test { public static void main(String[] args) throws InterruptedException { //不能直接newBlockingDeque,因為它是一個接口,要向上轉型 //LinkedBlockingDeque內部是基于鏈表方式來實現(xiàn)的 BlockingDeque<String> queue=new LinkedBlockingDeque<>(10);//此處可以指定一個具體的數(shù)字,這里的的10代表隊列的最大容量 queue.put("hello"); String elem=queue.take(); System.out.println(elem); elem=queue.take(); System.out.println(elem); } }
注意: put方法帶有阻塞功能,但是offer不具有,所以一般用put方法(能使用offer方法的原因是 BlockingDeque
繼承了Queue
)
打印結果如上所示,當打印了hello后,隊列為空,代碼執(zhí)行到elem=queue.take();
就不會繼續(xù)往下執(zhí)行了,此時線程進入阻塞等待狀態(tài),什么也不會打印了,直到有其他線程給隊列中放入新的元素為止
生產者消費者模型是在服務器開發(fā)和后端開發(fā)中比較常用的編程手段,一般用于解耦合和削峰填谷。
高耦合度:兩個代碼模塊的關聯(lián)關系比較高
高內聚:一個代碼模塊內各個元素彼此結合的緊密
因此,我們一般追求高內聚低耦合,這樣會加快執(zhí)行效率,而使用生產者消費者模型就可以解耦合
我們以實際生活中的情況為例,這里有兩臺服務器:A服務器和B服務器,當A服務器傳輸數(shù)據(jù)給B時,要是直接傳輸?shù)脑?,那么不是A向B推送數(shù)據(jù),就是B從A中拉取數(shù)據(jù),都是需要A和B直接交互,所以A和B存在依賴關系(A和B的耦合度比較高)。未來如果服務器需要擴展,比如加一個C服務器,讓A給C傳數(shù)據(jù),那么改動就比較復雜,且會降低效率。這時我們可以加一個隊列,這個隊列為阻塞隊列,如果A把數(shù)據(jù)寫到隊列里,B從中取,那么隊列相當于是中轉站(或者說交易場所),A相當于生產者(提供數(shù)據(jù)),B相當于消費者(接收數(shù)據(jù)),此時就構成了生產者消費者模型,這樣會讓代碼耦合度更低,維護更方便,執(zhí)行效率更高。
在計算機中,生產者充當其中一組線程,而消費者充當另一組線程,而交易場所就可以使用阻塞隊列了
在河道中大壩算是一個很重要的組成部分了,如果沒有大壩,大家試想一下結果:當汛期來臨后上游的水很大時,下游就會涌入大量的水發(fā)生水災讓莊稼被淹沒;而旱期的話下游的水會很少可能會引發(fā)旱災。若有大壩的話,汛期時大壩把多余的水存到大壩中,關閘蓄水,讓上游的水按一定速率往下流,避免突然一波大雨把下游淹了,這樣下游不至于出現(xiàn)水災。旱期時大壩把之前儲存好的水放出來,還是讓讓水按一定速率往下流,避免下流太缺水,這樣既可以避免汛期發(fā)生洪澇又可以避免旱期發(fā)生旱災了。
峰:相當于汛期
谷:相當于旱期
計算機中
這樣的情況在計算機中也是很典型的,尤其是在服務器開發(fā)中,網(wǎng)關通常會把互聯(lián)網(wǎng)中的請求轉發(fā)給業(yè)務服務器,比如一些商品服務器,用戶服務器,商家服務器(存放商家的信息),直播服務器。但因為互聯(lián)網(wǎng)過來的請求數(shù)量是多是少不可控,相當于上游的水,如果突然來了一大波請求,網(wǎng)關即使能扛得住,后續(xù)的很多服務器收到很多請求也就會崩潰(處理一個請求涉及到一系列的數(shù)據(jù)庫操作,因為數(shù)據(jù)庫相關操作效率本身比較低,這樣請求多了就處理不過來了,因此就會崩潰)
所以實際情況中網(wǎng)關和業(yè)務服務器之間往往用一個隊列來緩沖,這個隊列就是阻塞隊列(交易場所),用這個隊列來實現(xiàn)生產者(網(wǎng)關)消費者(業(yè)務服務器)模型,把請求緩存到隊列中,后面的消費者(業(yè)務服務器)按照自己固定的速率去讀請求。這樣當請求很多時,雖然隊列服務器可能會稍微受到一定壓力,但能保證業(yè)務服務器的安全。
import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; public class TestDemo { public static void main(String[] args) { // 使用一個 BlockingQueue 作為交易場所 BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(); // 此線程作為消費者 Thread customer = new Thread() { @Override public void run() { while (true) { // 取隊首元素 try { Integer value = queue.take(); System.out.println("消費元素: " + value); } catch (InterruptedException e) { e.printStackTrace(); } } } }; customer.start(); // 此線程作為生產者 Thread producer = new Thread() { @Override public void run() { for (int i = 1; i <= 10000; i++) { System.out.println("生產了元素: " + i); try { queue.put(i); Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } }; producer.start(); try { customer.join(); producer.join(); } catch (InterruptedException e) { e.printStackTrace(); } } }
打印如上(此代碼是讓生產者通過sleep每過1秒生產一個元素,而消費者不使用sleep,所以每當生產一個元素時,消費者都會立馬消費一個元素)
在學會如何使用BlockingQueue
后,那么如何自己去實現(xiàn)一個呢?
主要思路:
1.利用數(shù)組
2.head代表隊頭,tail代表隊尾
3.head和tail重合后到底是空的還是滿的判斷方法:專門定義一個size記錄當前隊列元素個數(shù),入隊列時size加1出隊列時size減1,當size為0表示空,為數(shù)組最大長度就是滿的(也可以浪費一個數(shù)組空間用head和tail重合表示空,用tail+1和head重合表示滿,但此方法較為麻煩,上一個方法較為直觀,因此我們使用上一個方法)
public class Test2 { static class BlockingQueue { private int[] items = new int[1000]; // 此處的1000相當于隊列的最大容量, 此處暫時不考慮擴容的問題. private int head = 0;//定義隊頭 private int tail = 0;//定義隊尾 private int size = 0;//數(shù)組大小 private Object locker = new Object(); // put 用來入隊列 public void put(int item) throws InterruptedException { synchronized (locker) { while (size == items.length) { // 隊列已經滿了,阻塞隊列開始阻塞 locker.wait(); } items[tail] = item; tail++; // 如果到達末尾, 就回到起始位置. if (tail >= items.length) { tail = 0; } size++; locker.notify(); } } // take 用來出隊列 public int take() throws InterruptedException { int ret = 0; synchronized (locker) { while (size == 0) { // 對于阻塞隊列來說, 如果隊列為空, 再嘗試取元素, 就要阻塞 locker.wait(); } ret = items[head]; head++; if (head >= items.length) { head = 0; } size--; // 此處的notify 用來喚醒 put 中的 wait locker.notify(); } return ret; } } public static void main(String[] args) throws InterruptedException { BlockingQueue queue = new BlockingQueue(); // 消費者線程 Thread consumer = new Thread() { @Override public void run() { while (true) { try { int elem = queue.take(); System.out.println("消費元素: " + elem); } catch (InterruptedException e) { e.printStackTrace(); } } } }; consumer.start(); // 生產者線程 Thread producer = new Thread() { @Override public void run() { for (int i = 1; i < 10000; i++) { System.out.println("生產元素: " + i); try { queue.put(i); Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } }; producer.start(); consumer.join(); producer.join(); } }
運行結果如上。
注意:
1.wait和notify的正確使用
2.put和take都會產生阻塞情況,但阻塞條件是對立的,wait不會同時觸發(fā)(put喚醒take阻塞,take喚醒put阻塞)
以上就是關于“Java中阻塞隊列怎么使用”這篇文章的內容,相信大家都有了一定的了解,希望小編分享的內容對大家有幫助,若想了解更多相關的知識內容,請關注億速云行業(yè)資訊頻道。
免責聲明:本站發(fā)布的內容(圖片、視頻和文字)以原創(chuàng)、轉載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關證據(jù),一經查實,將立刻刪除涉嫌侵權內容。