實(shí)現(xiàn)Java 阻塞隊(duì)列

xiaoxiaozi 發(fā)布于2019-08-19 10:23 / 553人閱讀

摘要：實(shí)現(xiàn)阻塞隊(duì)列在自己實(shí)現(xiàn)之前先搞清楚阻塞隊(duì)列的幾個(gè)特點(diǎn)基本隊(duì)列特性先進(jìn)先出。消費(fèi)隊(duì)列空時(shí)會阻塞直到寫入線程寫入了隊(duì)列數(shù)據(jù)后喚醒消費(fèi)線程。

實(shí)現(xiàn)Java 阻塞隊(duì)列

在自己實(shí)現(xiàn)之前先搞清楚阻塞隊(duì)列的幾個(gè)特點(diǎn)：
基本隊(duì)列特性：先進(jìn)先出。
寫入隊(duì)列空間不可用時(shí)會阻塞。
獲取隊(duì)列數(shù)據(jù)時(shí)當(dāng)隊(duì)列為空時(shí)將阻塞。

實(shí)現(xiàn)隊(duì)列的方式多種，總的來說就是數(shù)組和鏈表；其實(shí)我們只需要搞清楚其中一個(gè)即可，不同的特性主要表現(xiàn)為數(shù)組和鏈表的區(qū)別。

這里的 ArrayBlockingQueue 看名字很明顯是由數(shù)組實(shí)現(xiàn)。

我們先根據(jù)它這三個(gè)特性嘗試自己實(shí)現(xiàn)試試。

初始化隊(duì)列

我這里自定義了一個(gè)類：ArrayBlockQueue，它的構(gòu)造函數(shù)如下：

//隊(duì)列參數(shù)
    public int size;

    private volatile Object[] items;

    private volatile int pollPoint = 0;

    private volatile int addPoint = 0;

    private volatile int count = 0;

//初始化隊(duì)列容量
    public ArrayBlockQueue(int size) {
        this.size = size;
        this.items = new Object[size];
    }

寫入操作

有幾個(gè)需要注意的點(diǎn)：

隊(duì)列滿的時(shí)候，寫入的線程需要被阻塞。

寫入過隊(duì)列的數(shù)量大于隊(duì)列大小時(shí)需要從第一個(gè)下標(biāo)開始寫。

先看第一個(gè)隊(duì)列滿的時(shí)候，寫入的線程需要被阻塞，先來考慮下如何才能使一個(gè)線程被阻塞，看起來的表象線程卡住啥事也做不了。

其實(shí)這樣的一個(gè)特點(diǎn)很容易讓我們想到 Java 的等待通知機(jī)制來實(shí)現(xiàn)線程間通信。
所以我這里的做法是，一旦隊(duì)列滿時(shí)就將寫入線程調(diào)用 addLock.wait() 進(jìn)入 waiting 狀態(tài)，直到空間可用時(shí)再進(jìn)行喚醒。

初始化兩個(gè)鎖

//初始化兩個(gè)鎖
    private Object addLock = new Object();

    private Object pollLock = new Object();

寫入操作代碼

public void add(Object item) {

        synchronized (addLock) {
            //隊(duì)列滿則阻塞
            while (count >= size) {
                try {
                    addLock.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            items[addPoint] = item;
            addPoint = (addPoint + 1) % size;
            count++;

            //當(dāng)隊(duì)列從0個(gè)元素添加到1個(gè)元素，則說明從空狀態(tài)轉(zhuǎn)非空狀態(tài)可以通知一次取元素線程
            if (count == 1)
                synchronized (pollLock) {
                    pollLock.notifyAll();
                }
        }
    }

所以這里聲明了兩個(gè)對象用于隊(duì)列滿、空情況下的互相通知作用。

在寫入數(shù)據(jù)成功后需要使用 pollLock.notifyAll()，這樣的目的是當(dāng)獲取隊(duì)列為空時(shí)，一旦寫入數(shù)據(jù)成功就可以把消費(fèi)隊(duì)列的線程喚醒。

這里的 wait 和 notify 操作都需要對各自的對象使用 synchronized 方法塊，這是因?yàn)?wait 和 notifyAll 都需要獲取到各自的鎖。

消費(fèi)隊(duì)列

上文也提到了：當(dāng)隊(duì)列為空時(shí)，獲取隊(duì)列的線程需要被阻塞，直到隊(duì)列中有數(shù)據(jù)時(shí)才被喚醒。
代碼和寫入的非常類似，也很好理解；只是這里的等待、喚醒恰好是相反的。
總的來說就是：

寫入隊(duì)列滿時(shí)會阻塞直到獲取線程消費(fèi)了隊(duì)列數(shù)據(jù)后喚醒寫入線程。

消費(fèi)隊(duì)列空時(shí)會阻塞直到寫入線程寫入了隊(duì)列數(shù)據(jù)后喚醒消費(fèi)線程。

public Object poll() {

        synchronized (pollLock) {
            //隊(duì)列空則阻塞
            while (count == 0) {
                try {
                    pollLock.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            Object value = items[pollPoint];
            pollPoint = (pollPoint + 1) % size;
            count--;

            //當(dāng)隊(duì)列從滿到非滿，可以通知一次增添元素的線程
            if (count == (size - 1))
                synchronized (addLock) {
                    addLock.notifyAll();
                }
            return value;
        }
    }

測試

每一秒出隊(duì)1個(gè)

//消費(fèi)者
//每一秒出隊(duì)1個(gè)
class Counsum extends Thread {

    private ArrayBlockQueue queue;

    public Counsum(ArrayBlockQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {

        while (true) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " 出隊(duì)一個(gè)數(shù)據(jù):   " + queue.poll());
        }
    }
}

每5秒入隊(duì)2個(gè)，可見生產(chǎn)明顯慢于消費(fèi)

//生產(chǎn)者
//每5秒入隊(duì)2個(gè)
class Product extends Thread {

    private ArrayBlockQueue queue;

    public Product(ArrayBlockQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {

        int i = 0;
        while (true) {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            for (int j = 0; j < 2; j++, i++) {

                queue.add(i);
                System.out.println(Thread.currentThread().getName() + "【入隊(duì)一個(gè)數(shù)據(jù):   " + i+"】");
            }
        }
    }
}

main

public class Test extends Object {

    public static void main(String[] args) throws CloneNotSupportedException {
        
        ArrayBlockQueue queue = new ArrayBlockQueue(5);
        
        Thread counsum = new Counsum(queue);
        Thread counsum1 = new Counsum(queue);
        Thread counsum2 = new Counsum(queue);

        Thread product = new Product(queue);
        Thread product1 = new Product(queue);
        Thread product2 = new Product(queue);
        counsum.start();
        product.start();
        counsum1.start();
        product1.start();
        counsum2.start();
        product2.start();
    }
}
//output
消費(fèi)者線程-1出隊(duì)一個(gè)數(shù)據(jù):   0
生成者線程-3入隊(duì)一個(gè)數(shù)據(jù):    0
消費(fèi)者線程-3出隊(duì)一個(gè)數(shù)據(jù):   0
消費(fèi)者線程-2出隊(duì)一個(gè)數(shù)據(jù):   0
生成者線程-2入隊(duì)一個(gè)數(shù)據(jù):    0
生成者線程-1入隊(duì)一個(gè)數(shù)據(jù):    0
生成者線程-2入隊(duì)一個(gè)數(shù)據(jù):    1
生成者線程-3入隊(duì)一個(gè)數(shù)據(jù):    1
生成者線程-1入隊(duì)一個(gè)數(shù)據(jù):    1
消費(fèi)者線程-3出隊(duì)一個(gè)數(shù)據(jù):   1
消費(fèi)者線程-1出隊(duì)一個(gè)數(shù)據(jù):   1
消費(fèi)者線程-2出隊(duì)一個(gè)數(shù)據(jù):   1
消費(fèi)者線程-2出隊(duì)一個(gè)數(shù)據(jù):   2
生成者線程-2入隊(duì)一個(gè)數(shù)據(jù):    2
消費(fèi)者線程-1出隊(duì)一個(gè)數(shù)據(jù):   2
生成者線程-3入隊(duì)一個(gè)數(shù)據(jù):    2
生成者線程-1入隊(duì)一個(gè)數(shù)據(jù):    2
消費(fèi)者線程-3出隊(duì)一個(gè)數(shù)據(jù):   2
生成者線程-1入隊(duì)一個(gè)數(shù)據(jù):    3
生成者線程-3入隊(duì)一個(gè)數(shù)據(jù):    3
生成者線程-2入隊(duì)一個(gè)數(shù)據(jù):    3
消費(fèi)者線程-1出隊(duì)一個(gè)數(shù)據(jù):   3
消費(fèi)者線程-3出隊(duì)一個(gè)數(shù)據(jù):   3
消費(fèi)者線程-2出隊(duì)一個(gè)數(shù)據(jù):   3

引用