一、阻塞隊(duì)列

阻塞隊(duì)列：是一種特殊的隊(duì)列，也有先進(jìn)先出的特性。它是一種線程安全的隊(duì)列。
有以下兩個(gè)特性：

• 當(dāng)隊(duì)列滿的時(shí)候, 繼續(xù)入隊(duì)列就會(huì)阻塞, 直到有其他線程從隊(duì)列中取走元素。
• 當(dāng)隊(duì)列空的時(shí)候, 繼續(xù)出隊(duì)列也會(huì)阻塞, 直到有其他線程往隊(duì)列中插入元素。
  阻塞隊(duì)列的一個(gè)重要應(yīng)用場(chǎng)景就是：實(shí)現(xiàn)生產(chǎn)者消費(fèi)模型。

1.1 生產(chǎn)者消費(fèi)者模型

生產(chǎn)者消費(fèi)者模型：是多線程編程中的一種典型的編碼技巧。用來(lái)降低生產(chǎn)者與消費(fèi)者之間的耦合度。生產(chǎn)者和消費(fèi)者之間的交易場(chǎng)所就是一個(gè)阻塞隊(duì)列。

這樣的模型的優(yōu)勢(shì)有以下兩個(gè)：

1. 解耦合，降低代碼耦合度：
   像如果是A B兩個(gè)服務(wù)器，之間直接進(jìn)行交互，如果對(duì)A或者B中的數(shù)據(jù)進(jìn)行修改操作，大概率就會(huì)影響到另一個(gè)服務(wù)器。而使用阻塞隊(duì)列作為交易平臺(tái)，我們修改服務(wù)器的數(shù)據(jù)時(shí)，由于阻塞隊(duì)列中的結(jié)構(gòu)固定，兩個(gè)服務(wù)器之間的耦合度就降低。
2. 削峰削谷：
   在服務(wù)器中，波峰就是請(qǐng)求量高的時(shí)候，波谷就是請(qǐng)求量低的時(shí)候。
   如果是AB兩個(gè)服務(wù)器之間進(jìn)行交互，當(dāng)上游服務(wù)器A經(jīng)歷波峰，將大量請(qǐng)求傳給服務(wù)器B的時(shí)候，服務(wù)器就有可能掛掉。
   因?yàn)樯嫌畏?wù)器，干的活簡(jiǎn)單，消耗的資源少；而下游服務(wù)器，干的活復(fù)雜，消耗的的資源就多。
   但是如果我們將阻塞隊(duì)列作為交易平臺(tái)，那么服務(wù)器B就可以依據(jù)自己的節(jié)奏從隊(duì)列中拿請(qǐng)求。

但是這樣的模型也會(huì)付出代價(jià)：

1. 引入阻塞隊(duì)列之后整體結(jié)構(gòu)會(huì)更加復(fù)雜。比如本來(lái)是AB兩個(gè)服務(wù)器之間的交互，但引入一個(gè)作為阻塞隊(duì)列的服務(wù)器（這種稱為消息隊(duì)列），就需要部署這個(gè)服務(wù)器，還要與AB實(shí)現(xiàn)交互。
2. 效率也會(huì)有影響。

1.2 Java提供的阻塞隊(duì)列

提供了一個(gè)BlockingDeque的接口（需要導(dǎo)java.util.concurrent.BlockingQueue包）：

主要使用下面3個(gè)實(shí)現(xiàn)了BlockingDeque接口的來(lái)實(shí)例化阻塞隊(duì)列：

1. 鏈表實(shí)現(xiàn)的，LinkedBlockingDeque（需要導(dǎo)java.util.concurrent.LinkedBlockingDeque包）:

   

2. 數(shù)組實(shí)現(xiàn)的，ArrayBlockingDeque需要導(dǎo)`java.util.concurrent.ArrayBlockingDeque包）:

   

3. 小根堆實(shí)現(xiàn)的，PriorityBlockingDeque需要導(dǎo)java.util.concurrent.PriorityBlockingDeque包）:

   

在阻塞隊(duì)列中我們雖然可以使用隊(duì)列中常用的出隊(duì)列入隊(duì)列方法，但是那些方法不帶阻塞效果。帶阻塞效果的入隊(duì)列方法是put，出隊(duì)列方法是take，這兩個(gè)方法都會(huì)拋出InterruptedException異常。

1.3 實(shí)現(xiàn)一個(gè)簡(jiǎn)單生產(chǎn)者消費(fèi)者模型

實(shí)現(xiàn)一個(gè)簡(jiǎn)單的生產(chǎn)者消費(fèi)者模型：

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 + "入隊(duì)列成功");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        Thread consumer = new Thread(() -> {
            while(true){
                try {
                    int x = blockingQueue.take();
                    System.out.println(x + "出隊(duì)列成功");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        producer.start();
        consumer.start();
    }
}

二、自己實(shí)現(xiàn)阻塞隊(duì)列

我們使用數(shù)組來(lái)實(shí)現(xiàn)一個(gè)循環(huán)隊(duì)列。

不知道循環(huán)隊(duì)列的實(shí)現(xiàn)的可以看下面這個(gè)鏈接：隊(duì)列

2.1 成員變量

• 使用capacity代表數(shù)組的最大長(zhǎng)度；
• 使用size表示數(shù)組中元素的個(gè)數(shù)；
• head表示隊(duì)頭元素的下標(biāo)；
• tail表示隊(duì)尾元素的下標(biāo)。

	private int capacity = 0xffff;
    private String[] elem ;//存儲(chǔ)數(shù)組
    private int size;//存儲(chǔ)元素個(gè)數(shù)
    private int head;//隊(duì)頭
    private int tail;//隊(duì)尾

2.2 構(gòu)造方法

提供兩個(gè)構(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方法都涉及到修改判斷等操作，為避免原子性問(wèn)題帶來(lái)線程安全問(wèn)題對(duì)該這些操作都要加鎖。

在put方法中我們需要在隊(duì)列滿的時(shí)候發(fā)生阻塞，使用wait來(lái)等待。而在Java官方文檔給出了建議我們使用循環(huán)語(yǔ)句來(lái)使用wait。

因?yàn)閣ait是除了notify喚醒外，還有可能被interrupt方法喚醒拋出異常，如果只要if，不用while，拋出異常后就會(huì)繼續(xù)執(zhí)行下面的邏輯，帶來(lái)bug。而使用循環(huán)就不會(huì)，拋出異常后，會(huì)再次判斷循環(huán)條件。

最后在入隊(duì)成功后發(fā)出一個(gè)通知notify來(lái)喚醒由于隊(duì)列空而阻塞等待的線程。

    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)隊(duì)列為空的時(shí)候，跟put一樣使用wait來(lái)阻塞。

最后在出隊(duì)成功后發(fā)出一個(gè)通知notify來(lái)喚醒由于隊(duì)列滿而阻塞等待的線程。

 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 最終代碼

最終我們自己實(shí)現(xiàn)的一個(gè)簡(jiǎn)單的阻塞隊(duì)列就如下：

public class MyBlockingQueue {
    private int capacity = 0xffff;
    private String[] elem ;//存儲(chǔ)數(shù)組
    private int size;//存儲(chǔ)元素個(gè)數(shù)
    private int head;//隊(duì)頭
    private int tail;//隊(duì)尾


    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多線程阻塞隊(duì)列實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)JavaSE多線程阻塞隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論