實現Java 阻塞隊列

xiaoxiaozi 發布于2019-08-19 10:23 / 513人閱讀

摘要：實現阻塞隊列在自己實現之前先搞清楚阻塞隊列的幾個特點基本隊列特性先進先出。消費隊列空時會阻塞直到寫入線程寫入了隊列數據后喚醒消費線程。

實現Java 阻塞隊列

在自己實現之前先搞清楚阻塞隊列的幾個特點：
基本隊列特性：先進先出。
寫入隊列空間不可用時會阻塞。
獲取隊列數據時當隊列為空時將阻塞。

實現隊列的方式多種，總的來說就是數組和鏈表；其實我們只需要搞清楚其中一個即可，不同的特性主要表現為數組和鏈表的區別。

這里的 ArrayBlockingQueue 看名字很明顯是由數組實現。

我們先根據它這三個特性嘗試自己實現試試。

初始化隊列

我這里自定義了一個類：ArrayBlockQueue，它的構造函數如下：

//隊列參數
    public int size;

    private volatile Object[] items;

    private volatile int pollPoint = 0;

    private volatile int addPoint = 0;

    private volatile int count = 0;

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

寫入操作

有幾個需要注意的點：

隊列滿的時候，寫入的線程需要被阻塞。

寫入過隊列的數量大于隊列大小時需要從第一個下標開始寫。

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

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

初始化兩個鎖

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

    private Object pollLock = new Object();

寫入操作代碼

public void add(Object item) {

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

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

            //當隊列從0個元素添加到1個元素，則說明從空狀態轉非空狀態可以通知一次取元素線程
            if (count == 1)
                synchronized (pollLock) {
                    pollLock.notifyAll();
                }
        }
    }

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

在寫入數據成功后需要使用 pollLock.notifyAll()，這樣的目的是當獲取隊列為空時，一旦寫入數據成功就可以把消費隊列的線程喚醒。

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

消費隊列

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

寫入隊列滿時會阻塞直到獲取線程消費了隊列數據后喚醒寫入線程。

消費隊列空時會阻塞直到寫入線程寫入了隊列數據后喚醒消費線程。

public Object poll() {

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

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

測試

每一秒出隊1個

//消費者
//每一秒出隊1個
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() + " 出隊一個數據:   " + queue.poll());
        }
    }
}

每5秒入隊2個，可見生產明顯慢于消費

//生產者
//每5秒入隊2個
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() + "【入隊一個數據:   " + 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
消費者線程-1出隊一個數據:   0
生成者線程-3入隊一個數據:    0
消費者線程-3出隊一個數據:   0
消費者線程-2出隊一個數據:   0
生成者線程-2入隊一個數據:    0
生成者線程-1入隊一個數據:    0
生成者線程-2入隊一個數據:    1
生成者線程-3入隊一個數據:    1
生成者線程-1入隊一個數據:    1
消費者線程-3出隊一個數據:   1
消費者線程-1出隊一個數據:   1
消費者線程-2出隊一個數據:   1
消費者線程-2出隊一個數據:   2
生成者線程-2入隊一個數據:    2
消費者線程-1出隊一個數據:   2
生成者線程-3入隊一個數據:    2
生成者線程-1入隊一個數據:    2
消費者線程-3出隊一個數據:   2
生成者線程-1入隊一個數據:    3
生成者線程-3入隊一個數據:    3
生成者線程-2入隊一個數據:    3
消費者線程-1出隊一個數據:   3
消費者線程-3出隊一個數據:   3
消費者線程-2出隊一個數據:   3

引用

云服務器 GPU云服務器 java實現鏈式隊列開源消息隊列Java實現并發阻塞隊列阻塞隊列線程集

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://m.specialneedsforspecialkids.com/yun/75935.html

什么是阻塞隊列？如何使用阻塞隊列來實現生產者-消費者模型？

摘要：什么是阻塞隊列阻塞隊列是一個在隊列基礎上又支持了兩個附加操作的隊列。阻塞隊列的應用場景阻塞隊列常用于生產者和消費者的場景，生產者是向隊列里添加元素的線程，消費者是從隊列里取元素的線程。由鏈表結構組成的無界阻塞隊列。什么是阻塞隊列？阻塞隊列是一個在隊列基礎上又支持了兩個附加操作的隊列。 2個附加操作：支持阻塞的插入方法：隊列滿時，隊列會阻塞插入元素的線程，直到隊列不滿。支持阻塞的...

jemygraw 2019-08-15 14:44 評論0 收藏0
java 隊列

摘要：是基于鏈接節點的線程安全的隊列。通過這些高效并且線程安全的隊列類，為我們快速搭建高質量的多線程程序帶來極大的便利。隊列內部僅允許容納一個元素。該隊列的頭部是延遲期滿后保存時間最長的元素。隊列簡述 Queue：基本上，一個隊列就是一個先入先出（FIFO）的數據結構Queue接口與List、Set同一級別，都是繼承了Collection接口。LinkedList實現了Deque接口。...

goji 2019-08-16 17:47 評論0 收藏0
Java知識點總結（Java容器-Queue）

摘要：知識點總結容器知識點總結容器接口與是在同一級別，都是繼承了接口。另一種隊列則是雙端隊列，支持在頭尾兩端插入和移除元素，主要包括。一個由鏈表結構組成的無界阻塞隊列。是一個阻塞的線程安全的隊列，底層實現也是使用鏈式結構。 Java知識點總結（Java容器-Queue） @(Java知識點總結)[Java, Java容器] Queue Queue接口與List、Set是在同一級別，都是繼承了...

hedzr 2019-08-16 12:37 評論0 收藏0
Java阻塞隊列的實現

摘要：盡管中已經包含了阻塞隊列的官方實現，但是熟悉其背后的原理還是很有幫助的。阻塞隊列的實現阻塞隊列的實現類似于帶上限的的實現。下面是阻塞隊列的一個簡單實現必須注意到，在和方法內部，只有隊列的大小等于上限或者下限時，才調用方法。阻塞隊列與普通隊列的區別在于，當隊列是空的時，從隊列中獲取元素的操作將會被阻塞，或者當隊列是滿時，往隊列里添加元素的操作會被阻塞。試圖從空的阻塞隊列中獲取元素的線程...

付倫 2019-08-14 11:39 評論0 收藏0
Java 隊列

摘要：隊列中有元素時，就說明有過期了，線程繼續執行，然后元素出隊，根據相應的移除緩存。所以嚴格來說，雖然實現了隊列接口，但是它的目的卻并不是隊列，而是將生產者消費者線程配對。轉移隊列鏈式轉移隊列。引言本周在編寫短信驗證碼頻率限制切面的時候，經潘老師給的實現思路，使用隊列進行實現。看了看java.util包下的Queue接口，發現還從來沒用過呢！ Collection集合類接口，由它派生...

Pocher 2019-08-16 17:00 評論0 收藏0
Java并發編程筆記（一）

摘要：并發編程實戰水平很高，然而并不是本好書。一是多線程的控制，二是并發同步的管理。最后，使用和來關閉線程池，停止其中的線程。當線程調用或等阻塞時，對這個線程調用會使線程醒來，并受到，且線程的中斷標記被設置。《Java并發編程實戰》水平很高，然而并不是本好書。組織混亂、長篇大論、難以消化，中文翻譯也較死板。這里是一篇批評此書的帖子，很是貼切。俗話說：看到有這么多人罵你，我就放心了。然而知...

cnsworder 2019-08-15 14:19 評論0 收藏0