開(kāi)篇語(yǔ)

隊(duì)列在生活中隨處可見(jiàn)，醫(yī)院繳費(fèi)需要排隊(duì)、做核酸需要排隊(duì)、汽車等紅綠燈需要排隊(duì)等等。

隊(duì)列是一個(gè)按照先來(lái)到就排在前面，后來(lái)到排在后面的數(shù)據(jù)結(jié)構(gòu)，并且出隊(duì)的時(shí)候也是按照先來(lái)到先出隊(duì)。使用數(shù)組和鏈表進(jìn)行實(shí)現(xiàn)。通常用于協(xié)調(diào)任務(wù)的執(zhí)行和數(shù)據(jù)的交換。

介紹

ArrayBlockingQueue 是一個(gè)有界阻塞隊(duì)列，有界指的是隊(duì)列存在一個(gè)最大容量；阻塞指的是如果隊(duì)列已經(jīng)滿了，想要往隊(duì)列繼續(xù)添加元素的話，那么這個(gè)操作將會(huì)被暫停，直到隊(duì)列中有空位才會(huì)繼續(xù)完成添加操作。如果隊(duì)列已經(jīng)為空，想要從隊(duì)列中獲取元素，那么這個(gè)操作將會(huì)被暫停，直接隊(duì)列中存在元素才會(huì)繼續(xù)完成獲取操作。

它具有線程安全、性能好、公平鎖選項(xiàng)的特點(diǎn)：

• 線程安全：使用鎖和條件變量實(shí)現(xiàn)線程安全，無(wú)需額外的同步措施。
• 阻塞操作：當(dāng)隊(duì)列滿時(shí)，插入操作阻塞；當(dāng)隊(duì)列空時(shí)，刪除操作阻塞。這有助于避免忙等待和減少無(wú)意義的資源消耗。
• 公平鎖選項(xiàng)：支持是否使用公平鎖。避免鎖饑餓。
• 高性能：基于數(shù)組實(shí)現(xiàn)，內(nèi)存連續(xù)分配，訪問(wèn)性能較高。

但是同時(shí)也存在不靈活、無(wú)法支撐高并發(fā)的缺點(diǎn)

• 有界性：隊(duì)列的容量固定，不可動(dòng)態(tài)改變。因此在創(chuàng)建時(shí)分配多大容量將成為關(guān)鍵，分配過(guò)多會(huì)造成資源浪費(fèi)，分配過(guò)少會(huì)造成競(jìng)爭(zhēng)激烈。
• 鎖競(jìng)爭(zhēng)：在高并發(fā)情況下，鎖競(jìng)爭(zhēng)可能會(huì)導(dǎo)致性能下降。

實(shí)現(xiàn)原理

ArrayBlockingQueue 內(nèi)部使用數(shù)組作為元素的存儲(chǔ)結(jié)構(gòu)。

執(zhí)行存取操作時(shí)，都必須先獲取鎖，才可以執(zhí)行存取操作，這就保證ArrayBlockingQueue 是線程安全。

ArrayBlockingQueue 通過(guò)兩個(gè) Condition 條件隊(duì)列，一個(gè) notFull 條件，一個(gè) notEmpty 條件。在對(duì)隊(duì)列進(jìn)行插入元素操作時(shí)，判斷當(dāng)前隊(duì)列已經(jīng)滿，則通過(guò) notFull 條件將線程阻塞，直到其他線程通知該線程隊(duì)列可以繼續(xù)插入元素。在對(duì)隊(duì)列進(jìn)行移除元素操作時(shí)，判斷當(dāng)前隊(duì)列已經(jīng)空，則通過(guò) notEmpty 條件阻塞線程，直到其他線程通過(guò)該線程可以繼續(xù)獲取元素。

這樣保證線程的存取操作不會(huì)出現(xiàn)錯(cuò)誤。避免隊(duì)列在滿時(shí)，丟棄插入的元素；也避免在隊(duì)列空時(shí)取到一個(gè) null 值。

構(gòu)造函數(shù)

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

構(gòu)造函數(shù)中，需要指定隊(duì)列的容量和是否使用公平鎖。并且創(chuàng)建了兩個(gè) Condition 條件隊(duì)列，分別命名為 notEmpty 和 notFull，這兩個(gè)條件隊(duì)列是實(shí)現(xiàn)阻塞的關(guān)鍵。

通過(guò)構(gòu)造函數(shù)我們可以知道為什么它叫有界：因?yàn)閯?chuàng)建數(shù)組時(shí)，需要指定數(shù)組的容量，并且數(shù)組容量不能在運(yùn)行中動(dòng)態(tài)擴(kuò)大。所以隊(duì)列的容量是有邊界的，不是無(wú)限擴(kuò)張的。

插入函數(shù)

public void put(E e) throws InterruptedException {
    Objects.requireNonNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

• 獲取鎖
• 判斷當(dāng)前隊(duì)列是否已經(jīng)滿了
• 如果隊(duì)列1已經(jīng)滿了，調(diào)用 notFull 條件隊(duì)列的 await() 方法，將該線程阻塞，暫停該線程的插入操作。避免內(nèi)部溢出的問(wèn)題。
• 如果沒(méi)有滿，則直接調(diào)用入隊(duì)函數(shù) enqueue 插入到隊(duì)列末尾。
• 解鎖

獲取函數(shù)

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

• 獲取鎖
• 判斷當(dāng)前隊(duì)列是否為空
• 如果隊(duì)列沒(méi)有元素，調(diào)用 notEmpty 條件隊(duì)列的 await() 方法，將該線程阻塞，暫停該線程的獲取操作。避免獲取元素出錯(cuò)。
• 如果不為空，則直接調(diào)用出隊(duì)函數(shù) dequeue 移除隊(duì)列第一個(gè)元素，并返回給客戶端。
• 釋放鎖

入隊(duì)函數(shù)

private void enqueue(E e) {
    final Object[] items = this.items;
    items[putIndex] = e;
    if (++putIndex == items.length) putIndex = 0;
    count++;
    notEmpty.signal();
}

將元素插入到隊(duì)列的尾部，在完成插入操作之后會(huì)調(diào)用 notEmpty 對(duì)象的 signal 方法，告訴 notEmpty 阻塞隊(duì)列，現(xiàn)在隊(duì)列中已經(jīng)有元素，之前因?yàn)殛?duì)列沒(méi)有元素而被阻塞的線程，現(xiàn)在可以來(lái)獲取元素了。

內(nèi)部維護(hù)一個(gè) putIndex，用于表示下一個(gè)將要插入元素的坐標(biāo)。當(dāng) putIndex 等于數(shù)組長(zhǎng)度時(shí)，將會(huì)重置為 0。putIndex 是一個(gè)從 0 - length 循環(huán)使用的坐標(biāo)。

維護(hù)一個(gè) count 變量，用于表示隊(duì)列中存在多少元素，在存入的時(shí)候增加，在取出的時(shí)候減少。

出隊(duì)函數(shù)

private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E e = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length) takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();
    return e;
}

將隊(duì)列的第一個(gè)元素移除，并返回給客戶端。在完成移除操作之后會(huì)調(diào)用 notFull 對(duì)象的 signal 方法，告訴 notFull 阻塞隊(duì)列，現(xiàn)在隊(duì)列中已經(jīng)有空位了，之前因?yàn)殛?duì)列沒(méi)有空位而被阻塞的線程，現(xiàn)在可以繼續(xù)插入元素。

內(nèi)部維護(hù)一個(gè) takeIndex，用于表示下一個(gè)可以獲取元素的坐標(biāo)。當(dāng) takeIndex 等于數(shù)組長(zhǎng)度時(shí)，將會(huì)重置為 0。takeIndex 是一個(gè)從 0 至數(shù)組長(zhǎng)度之間循環(huán)使用的坐標(biāo)。

應(yīng)用場(chǎng)景

適用場(chǎng)景

ArrayBlockingQueue 適用于多個(gè)線程之間需要共享數(shù)據(jù)、協(xié)調(diào)任務(wù)執(zhí)行的場(chǎng)景。因此可以總結(jié)出以下幾個(gè)應(yīng)用場(chǎng)景：

• 線程池：線程池是一個(gè)常見(jiàn)的并發(fā)編程模型，它通過(guò)線程池中的線程執(zhí)行任務(wù)。并且可以重復(fù)使用這些線程。在線程池中，可以使用 ArrayBlockingQueue 來(lái)存儲(chǔ)需要執(zhí)行的任務(wù)，以此控制任務(wù)數(shù)量和執(zhí)行順序。當(dāng)線程池中的線程執(zhí)行完任務(wù)之后，可以從 ArrayBlockingQueue 中取出下一個(gè)任務(wù)執(zhí)行。
• 生產(chǎn)者-消費(fèi)者：在生產(chǎn)者-消費(fèi)者模型中，生產(chǎn)者負(fù)責(zé)生產(chǎn)數(shù)據(jù)，消費(fèi)者負(fù)責(zé)對(duì)數(shù)據(jù)進(jìn)行處理。在這種模式下，ArrayBlockingQueue 可以作為生產(chǎn)者與消費(fèi)者之間的數(shù)據(jù)通道，保證線程安全和數(shù)據(jù)正確。

實(shí)際應(yīng)用場(chǎng)景

• Apache Tomcat Apache Tomcat 是一個(gè)流行的 Java Web 應(yīng)用服務(wù)器，它使用 ArrayBlockingQueue 來(lái)實(shí)現(xiàn)內(nèi)部的請(qǐng)求隊(duì)列。當(dāng)請(qǐng)求到達(dá) Tomcat 時(shí)，它們被放入一個(gè) ArrayBlockingQueue 中，并由工作線程從隊(duì)列中取出并處理請(qǐng)求。
• Netty Netty 是一個(gè)高性能的網(wǎng)絡(luò)編程框架，它使用 ArrayBlockingQueue 來(lái)實(shí)現(xiàn)內(nèi)部的事件隊(duì)列。當(dāng)有新的網(wǎng)絡(luò)事件到達(dá)時(shí)，它們被放入一個(gè) ArrayBlockingQueue 中，并由 IO 線程從隊(duì)列中取出并處理事件。

總結(jié)

ArrayBlockingQueue 是一個(gè)固定容量，并且采用阻塞方式的隊(duì)列。內(nèi)部采用鎖和條件隊(duì)列保證了線程安全性。支持公平鎖選項(xiàng)。但是因?yàn)椴捎米枞麢C(jī)制且容量有限，無(wú)法很好滿足高并發(fā)需求。

以上就是一文探究ArrayBlockQueue函數(shù)及應(yīng)用場(chǎng)景的詳細(xì)內(nèi)容，更多關(guān)于ArrayBlockQueue函數(shù)應(yīng)用場(chǎng)景的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論