Java多線程案例之阻塞隊列詳解

更新時間：2022年10月18日 14:55:48 作者：Moon?Bay

阻塞隊列是一種特殊的隊列.?也遵守?“先進先出”?的原則.阻塞隊列能是一種線程安全的數(shù)據(jù)結(jié)構(gòu)。本文將通過一些示例為大家詳細講講阻塞隊列的原理與使用，感興趣的小伙伴可以學習一下

一.阻塞隊列介紹

1.1阻塞隊列特性

阻塞隊列特性:

一.安全性

二.產(chǎn)生阻塞效果

阻塞隊列是一種特殊的隊列. 也遵守 “先進先出” 的原則.阻塞隊列能是一種線程安全的數(shù)據(jù)結(jié)構(gòu), 并且具有以下特性:

當隊列滿的時候, 繼續(xù)入隊列就會阻塞, 直到有其他線程從隊列中取走元素.
當隊列空的時候, 繼續(xù)出隊列也會阻塞, 直到有其他線程往隊列中插入元素.

阻塞隊列的一個典型應用場景就是 “生產(chǎn)者消費者模型”. 這是一種非常典型的開發(fā)模型.

1.2阻塞隊列的優(yōu)點

我們可以將阻塞隊列比做成"生產(chǎn)者"和"消費者"的"交易平臺".

我們可以把這個模型來比做成"包餃子"

A 的作用是搟餃子皮,也就是"生產(chǎn)者"

B 的作用是包餃子,也就是"消費者"

X 的作用一個當作放搟好餃子皮的一個盤中,也就是阻塞隊列

這樣我們根據(jù)A,B,X可以想象以下場景

場景一:

當A搟餃子皮的速度過快,X被A的桿好餃子皮放滿了,這樣A就需要停止搟餃子皮這一個操作,這時只能等待B來利用A提供的餃子皮包餃子后X所空出的空間,來給A提供生產(chǎn)的環(huán)境

場景二:

當B包餃子的速度過快,X被B的包餃子所用的餃子皮用空,這樣B就需要停止包餃子這一個操作,這時只能等待A提供的餃子皮包餃子后X所存在餃子皮,來給B提供消費的環(huán)境

二.生產(chǎn)者消費者模型

生產(chǎn)者消費者模式就是通過一個容器來解決生產(chǎn)者和消費者的強耦合問題

生產(chǎn)者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，所以生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產(chǎn)者要數(shù)據(jù)，而是直接從阻塞隊列里取

(1) 阻塞隊列就相當于一個緩沖區(qū)，平衡了生產(chǎn)者和消費者的處理能力.

比如在 “秒殺” 場景下, 服務(wù)器同一時刻可能會收到大量的支付請求. 如果直接處理這些支付請求,服務(wù)器可能扛不住(每個支付請求的處理都需要比較復雜的流程). 這個時候就可以把這些請求都放到一個阻塞隊列中, 然后再由消費者線程慢慢的來處理每個支付請求.這樣做可以有效進行 “削峰”, 防止服務(wù)器被突然到來的一波請求直接沖垮.

(2) 阻塞隊列也能使生產(chǎn)者和消費者之間解耦.

比如過年一家人一起包餃子. 一般都是有明確分工, 比如一個人負責搟餃子皮, 其他人負責包. 搟餃子皮的人就是 “生產(chǎn)者”, 包餃子的人就是 “消費者”.搟餃子皮的人不關(guān)心包餃子的人是誰(能包就行, 無論是手工包, 借助工具, 還是機器包), 包餃子的人也不關(guān)心搟餃子皮的人是誰(有餃子皮就行, 無論是用搟面杖搟的, 還是拿罐頭瓶搟, 還是直接從超市買的).

2.1阻塞隊列對生產(chǎn)者的優(yōu)化

優(yōu)化一:能夠讓多個服務(wù)器程序之間更充分的解耦合:

如果不使用生產(chǎn)者和消費者模型,此時A和B的耦合性比較強,如果A線程出現(xiàn)一些狀況B就會掛,B線程出現(xiàn)一些狀況A就會掛,這時當我們引入阻塞隊列后我們就可以將A,B線程分開,如果A,B線程掛了有阻塞隊列的存在下,是不會影響別的線程

優(yōu)化二:能夠?qū)τ谡埱筮M行"削峰填谷":

我們可以聯(lián)想到我國的三峽大壩,三峽大壩就相當于阻塞隊列,當我們遇到雨水大的季節(jié),我們就可以關(guān)閉三峽大壩,利用三峽大壩來存水;當我們遇到干旱期,我們就可以打開三峽大壩的門,來放水解決干旱問題

三.標準庫中的阻塞隊列

3.1Java提供阻塞隊列實現(xiàn)的標準類

java官方也提供了阻塞隊列的標準類，主要有下面幾個：

標準類	說明
ArrayBlockingQueue	一個由數(shù)組結(jié)構(gòu)組成的有界阻塞隊列
LinkedBlockingQueue	一個由鏈表結(jié)構(gòu)組成的有界阻塞隊列
PriorityBlockingQueue	一個支持優(yōu)先級排序的無界阻塞隊列
DelayQueue	一個使用優(yōu)先級隊列實現(xiàn)的無界阻塞隊列
SynchronousQueue	一個不存儲元素的阻塞隊列
LinkedTransferQueue	一個由鏈表結(jié)構(gòu)組成的無界阻塞隊列
LinkedBlockingDeque	一個由鏈表結(jié)構(gòu)組成的雙向阻塞隊列
BlockingQueue接口	單向阻塞隊列實現(xiàn)了該接口
BlockingDeque接口	雙向阻塞隊列實現(xiàn)了該接口

3.2Blockingqueue基本使用

在 Java 標準庫中內(nèi)置了阻塞隊列. 如果我們需要在一些程序中使用阻塞隊列, 直接使用標準庫中的即可.

BlockingQueue 是一個接口. 真正實現(xiàn)的類是 LinkedBlockingQueue.

put 方法用于阻塞式的入隊列, take 用于阻塞式的出隊列.

BlockingQueue 也有 offer, poll, peek 等方法, 但是這些方法不帶有阻塞特性.

BlockingQueue<String> queue = new LinkedBlockingQueue<>();
// 入隊列
queue.put("abc");
// 出隊列. 如果沒有 put 直接 take, 就會阻塞.
String elem = queue.take();

四.阻塞隊列實現(xiàn)

4.1阻塞隊列的代碼實現(xiàn)

我們通過 “循環(huán)隊列” 的方式來實現(xiàn)

使用 synchronized 進行加鎖控制.put 插入元素的時候, 判定如果隊列滿了, 就進行 wait. (注意, 要在循環(huán)中進行 wait. 被喚醒時不一定隊列就不滿了, 因為同時可能是喚醒了多個線程).take 取出元素的時候, 判定如果隊列為空, 就進行 wait. (也是循環(huán) wait)

我們在設(shè)計阻塞隊列的時候可以將隊列聯(lián)想成一個圓

class BlockingQueue{
    //隊列里存放的個數(shù)
   volatile private int size = 0;
    //隊列的頭節(jié)點
    private int head = 0;
    //隊列的尾節(jié)點
    private int prov = 0;
    //創(chuàng)建一個數(shù)組,我們來給這個數(shù)組的容量設(shè)置為100
    private int[] array = new int[100];
    //創(chuàng)建一個專業(yè)的鎖對象
    private Object object = new Object();
    //實現(xiàn)阻塞隊列中的put方法
    public void put(int value) throws InterruptedException {
        synchronized (object) {
            //當數(shù)組已經(jīng)滿了
            if (size == array.length) {
                object.wait();
            } else {
                //我們可以優(yōu)化成prov = (prov + 1) % items.length
                array[prov] = value;
                prov ++;
               if (prov >= array.length) {
                   prov = 0;
               }
            }
            size++;
            object.notify();
        }
    }
    //實現(xiàn)阻塞隊列中的take方法
    public int take() throws InterruptedException {
        synchronized (object) {
            if (size == 0) {
                object.wait();
            }
            int x = array[head];
            head++;
            if (head >= array.length) {
                head = 0;
            }
            size--;
            object.notify();
            return x;
        }
    }
}

4.2阻塞隊列搭配生產(chǎn)者與消費者的代碼實現(xiàn)

class BlockingQueue{
    //隊列里存放的個數(shù)
   volatile private int size = 0;
    //隊列的頭節(jié)點
    private int head = 0;
    //隊列的尾節(jié)點
    private int prov = 0;
    //創(chuàng)建一個數(shù)組,我們來給這個數(shù)組的容量設(shè)置為100
    private int[] array = new int[100];
    //創(chuàng)建一個專業(yè)的鎖對象
    private Object object = new Object();
    //實現(xiàn)阻塞隊列中的put方法
    public void put(int value) throws InterruptedException {
        synchronized (object) {
            //當數(shù)組已經(jīng)滿了
            if (size == array.length) {
                object.wait();
            } else {
                //我們可以優(yōu)化成prov = (prov + 1) % items.length
                array[prov] = value;
                prov ++;
               if (prov >= array.length) {
                   prov = 0;
               }
            }
            size++;
            object.notify();
        }
    }
    //實現(xiàn)阻塞隊列中的take方法
    public int take() throws InterruptedException {
        synchronized (object) {
            if (size == 0) {
                object.wait();
            }
            int x = array[head];
            head++;
            if (head >= array.length) {
                head = 0;
            }
            size--;
            object.notify();
            return x;
        }
    }
}
public class Test {
    public static void main(String[] args) {
        BlockingQueue blockingQueue = new BlockingQueue();
        Thread thread1 = new Thread(()-> {
            while (true) {
                for (int i = 0; i < 100; i++) {
                    try {
                        blockingQueue.put(i);
                        System.out.println("生產(chǎn)了"+i);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        Thread thread2 = new Thread(()->{
                while (true) {
                    try {
                        int b = blockingQueue.take();
                        System.out.println("消耗了"+b);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
        });
        thread1.start();
        thread2.start();
    }
}