隊(duì)列概述以及常用方法

隊(duì)列是一個有序列表，可以用數(shù)組或者鏈表實(shí)現(xiàn)，且遵循先進(jìn)先出的原則，在此基礎(chǔ)上還分為阻塞、非阻塞、優(yōu)先隊(duì)列等。

java中只要是隊(duì)列都實(shí)現(xiàn)了Queue接口，隊(duì)列接口定義的方法如下：

1 非阻塞隊(duì)列

1 ConcurrentLinkedQueue（線程安全）

單向鏈表結(jié)構(gòu)的無界并發(fā)隊(duì)列, 非阻塞隊(duì)列，由CAS實(shí)現(xiàn)線程安全

• 是否有界：無界
• 是否線程安全：是
• 性能：高

2 ConcurrentLinkedDeque（線程安全）

雙向鏈表結(jié)構(gòu)的無界并發(fā)隊(duì)列, 非阻塞隊(duì)列，由CAS實(shí)現(xiàn)線程安全，相較于ConcurrentLinkedQueue，ConcurrentLinkedDeque在某些場景下提供了更加靈活的操作，比如可以從兩端進(jìn)行元素的插入和刪除操作。

• 是否有界：無界
• 是否線程安全：是
• 性能：略低于ConcurrentLinkedQueue

3 PriorityQueue（線程不安全）

PriorityQueue是Java中的一個基于優(yōu)先級堆的隊(duì)列，基于數(shù)組實(shí)現(xiàn)，它可以按照元素的優(yōu)先級進(jìn)行排序，并且在插入和刪除元素時(shí)保持有序狀態(tài)。具體來說，PriorityQueue中的元素必須實(shí)現(xiàn)Comparable接口或者通過構(gòu)造函數(shù)提供一個Comparator對象，以便進(jìn)行元素的比較和排序。

• 是否有界：無界
• 是否線程安全：否（如需線程安全可以使用PriorityBlockingQueue）
• 性能：比普通隊(duì)列略低（插入和獲取操作都需要進(jìn)行堆調(diào)整和排序操作）

PriorityQueue的實(shí)現(xiàn)基于數(shù)組，它可以自動擴(kuò)容，但是它的容量大小是有限制的。在插入和刪除元素時(shí)，PriorityQueue會根據(jù)元素的優(yōu)先級重新調(diào)整堆的結(jié)構(gòu)，保證堆頂元素一定是優(yōu)先級最高的元素。

2 阻塞隊(duì)列（線程安全）

阻塞隊(duì)列BlockingQueue繼承了 Queue 接口，是隊(duì)列的一種，阻塞隊(duì)列是線程安全的，阻塞隊(duì)列在隊(duì)列基礎(chǔ)上加了兩個重要的接口put() take()：

注意：

在ThreadPoolExecutor中就使用的阻塞隊(duì)列來實(shí)現(xiàn)線程的不結(jié)束，以達(dá)到線程復(fù)用。

常見阻塞隊(duì)列，juc中常用的阻塞隊(duì)列如下：

1. ArrayBlockingQueue

基于數(shù)組結(jié)構(gòu)實(shí)現(xiàn)的一個有界阻塞隊(duì)列，在創(chuàng)建ArrayBlockingQueue對象時(shí)必須制定容量大小。并且可以指定公平性與非公平性，默認(rèn)情況下為非公平的，即不保證等待時(shí)間最長的隊(duì)列最優(yōu)先能夠訪問隊(duì)列。

它使用了同步機(jī)制來保證多個線程對隊(duì)列的訪問不會發(fā)生競態(tài)條件。具體來說，ArrayBlockingQueue使用ReentrantLock和Condition來保證線程安全和隊(duì)列的阻塞和喚醒。

• 是否有界：有
• 是否線程安全：是
• 性能：高

2. LinkedBlockingQueue

基于鏈表結(jié)構(gòu)實(shí)現(xiàn)的一個有界阻塞隊(duì)列，在創(chuàng)建LinkedBlockingQueue對象時(shí)如果不指定容量大小，則默認(rèn)大小為Integer.MAX_VALUE

• 是否有界：無
• 是否線程安全：是
• 性能：性能低于ArrayBlockingQueue

LinkedBlockingQueue的實(shí)現(xiàn)使用了兩個鎖，一個用于生產(chǎn)者線程的訪問，另一個用于消費(fèi)者線程的訪問。這樣可以避免在高并發(fā)場景下產(chǎn)生競態(tài)條件，保證了線程安全性。

LinkedBlockingQueue的性能比ArrayBlockingQueue略低，因?yàn)樗褂面湵矶皇菙?shù)組來存儲元素。

但是，由于LinkedBlockingQueue是無界隊(duì)列，因此它在需要動態(tài)調(diào)整容量時(shí)更加靈活，可以根據(jù)實(shí)際情況自動擴(kuò)容或縮容。

總的來說，LinkedBlockingQueue是一個非常實(shí)用的數(shù)據(jù)結(jié)構(gòu)，在生產(chǎn)者-消費(fèi)者模型、線程池等多線程場景中被廣泛使用。

3. SynchronousQueue

SynchronousQueue內(nèi)部并不存儲任何元素，它只是在等待其他線程將元素插入或刪除隊(duì)列時(shí)才會阻塞，每一個put操作必須等待一個take操作，否則不能繼續(xù)添加元素

• 是否有界：有
• 是否線程安全：是
• 性能：-

具體來說，當(dāng)一個線程調(diào)用SynchronousQueue的put()方法時(shí)，如果沒有其他線程在等待從隊(duì)列中取出元素，那么當(dāng)前線程將會阻塞，直到有其他線程調(diào)用take()方法來取出這個元素。

同樣地，當(dāng)一個線程調(diào)用take()方法時(shí)，如果沒有其他線程在等待向隊(duì)列中插入元素，那么當(dāng)前線程也會阻塞，直到有其他線程調(diào)用put()方法來插入元素。

SynchronousQueue可以用于一些特殊的場景，例如在生產(chǎn)者和消費(fèi)者之間進(jìn)行高效的交互。在這種情況下，生產(chǎn)者線程將元素直接交給消費(fèi)者線程處理，而不需要通過中間緩存。另外，SynchronousQueue還可以用于實(shí)現(xiàn)一些并發(fā)算法和數(shù)據(jù)結(jié)構(gòu)，例如公平的交換器（Fair Exchanger）。

需要注意的是，SynchronousQueue不允許null元素，如果試圖插入null元素，將會拋出NullPointerException。

4. LinkedTransferQueue

基于鏈表結(jié)構(gòu)實(shí)現(xiàn)的一個無界阻塞隊(duì)列，是 SynchronousQueue 和 LinkedBlockingQueue 的合體，它使用了CAS（Compare and Swap）操作來保證并發(fā)安全性，與其他阻塞隊(duì)列不同，LinkedTransferQueue內(nèi)部使用了多個節(jié)點(diǎn)來實(shí)現(xiàn)隊(duì)列，每個節(jié)點(diǎn)包含一個元素以及指向下一個節(jié)點(diǎn)的指針，這種方式可以避免在并發(fā)場景下使用鎖導(dǎo)致的性能瓶頸，性能比 LinkedBlockingQueue 更高（沒有鎖操作），比 SynchronousQueue能存儲更多的元素

• 是否有界：無
• 是否線程安全：是
• 性能：高

TransferQueue接口相較普通的阻塞隊(duì)列，增加了這么幾個方法：

public interface TransferQueue<E> extends BlockingQueue<E> {
    // 如果可能，立即將元素轉(zhuǎn)移給等待的消費(fèi)者。 
    // 更確切地說，如果存在消費(fèi)者已經(jīng)等待接收它（在 take 或 timed poll（long，TimeUnit）poll）中，則立即傳送指定的元素，否則返回 false。
    boolean tryTransfer(E e);

    // 將元素轉(zhuǎn)移給消費(fèi)者，如果需要的話等待。 
    // 更準(zhǔn)確地說，如果存在一個消費(fèi)者已經(jīng)等待接收它（在 take 或timed poll（long，TimeUnit）poll）中，則立即傳送指定的元素，否則等待直到元素由消費(fèi)者接收。
    void transfer(E e) throws InterruptedException;

    // 上面方法的基礎(chǔ)上設(shè)置超時(shí)時(shí)間
    boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException;

    // 如果至少有一位消費(fèi)者在等待，則返回 true
    boolean hasWaitingConsumer();

    // 返回等待消費(fèi)者人數(shù)的估計(jì)值
    int getWaitingConsumerCount();
}

但是在使用LinkedTransferQueue時(shí)，需要注意一些細(xì)節(jié)：

• LinkedTransferQueue不支持null元素，如果插入null元素將會拋出NullPointerException異常。
• 在使用transfer()方法時(shí)，如果隊(duì)列已滿或?yàn)榭?，調(diào)用transfer()方法的線程將會被阻塞，直到另一個線程將元素插入或取走。因此，在使用transfer()方法時(shí)需要注意線程的阻塞問題，避免出現(xiàn)死鎖或線程饑餓的情況。
• LinkedTransferQueue的迭代器只能用于遍歷當(dāng)前隊(duì)列中的元素，無法保證在迭代期間隊(duì)列中的元素不被修改或刪除。

5. LinkedBlockingDeque

基于雙向鏈表結(jié)構(gòu)實(shí)現(xiàn)的一個雙向阻塞隊(duì)列，它可以同時(shí)從隊(duì)列的兩端插入和刪除元素，因此支持隊(duì)列和棧的操作。

• 是否有界：無界，但是可以設(shè)置隊(duì)列上限
• 是否線程安全：是
• 性能：高

使用LinkedBlockingDeque時(shí)需要注意以下幾點(diǎn)：

• LinkedBlockingDeque不支持null元素，如果插入null元素將會拋出NullPointerException異常。
• LinkedBlockingDeque的阻塞特性可能會導(dǎo)致線程饑餓，即某些線程一直無法獲得訪問隊(duì)列的機(jī)會。為避免這種情況，可以在創(chuàng)建LinkedBlockingDeque時(shí)指定容量，或者使用putFirst()、putLast()、takeFirst()、takeLast()等非阻塞操作。

6. PriorityBlockingQueue

基于優(yōu)先級的阻塞隊(duì)列，它使用數(shù)組實(shí)現(xiàn)，并且具有無界限的容量，它會按照元素的優(yōu)先級對元素進(jìn)行排序，按照優(yōu)先級順序出隊(duì)，每次出隊(duì)的元素都是優(yōu)先級最高的元素。

PriorityBlockingQueue的注意事項(xiàng)如下：

• PriorityBlockingQueue允許插入null元素，但是不允許插入不可比較的元素，如果插入不可比較的元素，將會拋出ClassCastException異常。
• PriorityBlockingQueue的迭代順序并不是按照元素的優(yōu)先級順序排列的。雖然PriorityBlockingQueue能夠保證每次取出的元素都是優(yōu)先級最高的元素，但是在迭代PriorityBlockingQueue時(shí)，它的順序是無法保證的。
• PriorityBlockingQueue在插入元素時(shí)，會根據(jù)元素的優(yōu)先級對元素進(jìn)行排序。因此，如果隊(duì)列中的元素的優(yōu)先級發(fā)生了變化，需要調(diào)用隊(duì)列中的元素的offer、add、put方法重新排序。
• PriorityBlockingQueue不支持remove(Object)操作，因?yàn)樗鼰o法快速地找到并刪除指定元素。如果需要刪除指定元素，可以先將PriorityBlockingQueue中的元素復(fù)制到另一個集合中，然后再從新集合中刪除指定元素，最后將新集合中的元素重新添加到PriorityBlockingQueue中。
• PriorityBlockingQueue在迭代、插入和刪除元素時(shí)，都使用了ReentrantLock來保證線程安全。因此，在使用PriorityBlockingQueue時(shí)需要注意避免死鎖和饑餓等問題。

簡單示例：

import java.util.concurrent.PriorityBlockingQueue;

public class Task implements Comparable<Task> {
    private String name;
    private int priority;

    public Task(String name, int priority) {
        this.name = name;
        this.priority = priority;
    }

    public String getName() {
        return name;
    }

    public int getPriority() {
        return priority;
    }

    @Override
    public int compareTo(Task o) {
        return Integer.compare(o.priority, this.priority);
    }
}

public class PriorityBlockingQueueExample {
    public static void main(String[] args) {
        // 創(chuàng)建一個PriorityBlockingQueue對象
        PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<Task>();
        // 添加任務(wù)到隊(duì)列中
        queue.offer(new Task("Task1", 3));
        queue.offer(new Task("Task2", 2));
        queue.offer(new Task("Task3", 1));
        // 取出隊(duì)列中的任務(wù)
        while (!queue.isEmpty()) {
            Task task = queue.poll();
            System.out.println("Execute task " + task.getName() + " with priority " + task.getPriority());
        }
    }
}

輸出結(jié)果:
Execute task Task1 with priority 3
Execute task Task2 with priority 2
Execute task Task3 with priority 1

7. DelayQueue

是一種延時(shí)阻塞隊(duì)列，它可以存儲實(shí)現(xiàn)了Delayed接口的元素，其中每個元素都有一個過期時(shí)間，即當(dāng)元素的過期時(shí)間到達(dá)時(shí)，元素會被取出

• 是否有界：無界的，但是在實(shí)際應(yīng)用中，我們可以通過指定初始容量來控制隊(duì)列的大小，以避免內(nèi)存溢出等問題。
• 是否線程安全：是
• 性能：性能要比普通的阻塞隊(duì)列略低（插入和獲取操作都需要進(jìn)行堆調(diào)整和排序操作）

DelayQueue內(nèi)部使用PriorityQueue實(shí)現(xiàn)，元素會按照過期時(shí)間排序，即過期時(shí)間最短的元素排在隊(duì)列的頭部。DelayQueue的插入操作是非阻塞的，但是取出操作是阻塞的，即當(dāng)隊(duì)列中沒有過期元素時(shí)，取出操作會一直被阻塞，直到隊(duì)列中有過期元素被插入。

使用DelayQueue時(shí)需要注意以下幾點(diǎn)：

• DelayQueue的元素必須實(shí)現(xiàn)Delayed接口，并重寫getDelay()和compareTo()方法，其中g(shù)etDelay()方法返回元素的過期時(shí)間距當(dāng)前時(shí)間的剩余時(shí)間（單位為毫秒），compareTo()方法用于比較元素的過期時(shí)間。
• DelayQueue內(nèi)部使用ReentrantLock和Condition實(shí)現(xiàn)線程同步和阻塞操作，因此是線程安全的。
• DelayQueue的取出操作可能會被阻塞，如果需要在隊(duì)列為空時(shí)立即返回，可以使用poll()方法而不是take()方法。
• DelayQueue在理論上是無界的，因?yàn)樗褂肞riorityQueue來存儲元素，PriorityQueue的長度是沒有限制的。但是，如果你想控制DelayQueue的大小，你可以在創(chuàng)建DelayQueue實(shí)例時(shí)指定一個初始容量。

DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>(new ArrayList<DelayedTask>(capacity));

下面是一個簡單的使用DelayQueue的示例代碼，在取出元素時(shí)，隊(duì)列會自動按照元素的過期時(shí)間進(jìn)行排序，并等待元素過期后再取出。

注意，在實(shí)際使用中，需要根據(jù)具體的業(yè)務(wù)場景來設(shè)置元素的過期時(shí)間和比較方式：

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class DelayQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<DelayElement> queue = new DelayQueue<>();

        // 添加元素到隊(duì)列
        queue.add(new DelayElement("A", 3000));
        queue.add(new DelayElement("B", 2000));
        queue.add(new DelayElement("C", 1000));

        // 取出元素
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
    }
}

class DelayElement implements Delayed {
    private String name;
    private long expireTime;

    public DelayElement(String name, long delayTime) {
        this.name = name;
        this.expireTime = System.currentTimeMillis() + delayTime;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return expireTime - System.currentTimeMillis();
    }

    @Override
    public int compareTo(Delayed o) {
        return Long.compare(this.expireTime, ((DelayElement) o).expireTime);
    }

    @Override
    public String toString() {
        return "DelayElement{" +
                "name='" + name + '\'' +
                '}';
    }
}

總結(jié)

以上為個人經(jīng)驗(yàn)，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論