Java并發(fā)編程之阻塞隊列(BlockingQueue)詳解

更新時間：2021年09月10日 10:51:18 作者：小黑說Java

這篇文章主要介紹了詳解Java阻塞隊列(BlockingQueue)的實現(xiàn)原理，阻塞隊列是Java util.concurrent包下重要的數(shù)據(jù)結(jié)構(gòu)，有興趣的可以了解一下

大家好，我是小黑，一個在互聯(lián)網(wǎng)茍且偷生的農(nóng)民工。

隊列

學(xué)過數(shù)據(jù)結(jié)構(gòu)的同學(xué)應(yīng)該都知道，隊列是數(shù)據(jù)結(jié)構(gòu)中一種特殊的線性表結(jié)構(gòu)，和平時使用的List，Set這些數(shù)據(jù)結(jié)構(gòu)相比有點特殊，它的特殊之處在于它只允許在隊列的頭部（Head）進(jìn)行刪除操作，在尾部（Tail）進(jìn)行插入操作，這種方式的隊列我們稱之為先進(jìn)先出隊列（FIFO）。

在JDK1.5中推出了隊列這一數(shù)據(jù)結(jié)構(gòu)的具體實現(xiàn)，接口Queue是對于隊列的定義，并有一些列具有特殊功能的隊列實現(xiàn)。

在Queue接口中定義了隊列的如下方法：

其中add(E)并非Queue接口新定義，而是從Collection接口繼承而來的。

阻塞隊列

BlockingQueue接口也是在JDK1.5中推出，存放在java.util.concurrent包中，繼承自Queue，所以在BlockingQueue中有Queue的所有方法。

從名字就可以看出BlockingQueue是一種阻塞隊列，它支持在檢索元素時如果隊列為空可以一直阻塞等待直到有元素可以獲取，同樣在添加元素時如果隊列已滿會阻塞等待隊列中有空閑的存儲空間。

BlockingQueue的方法可以歸納為四類：

在操作時如不能立即滿足，會直接拋出異常
在操作時如不能立即滿足，則返回特殊的值，如插入、移除方法會返回false，檢查方法會返回null
在操作時如不能立即滿足，則會阻塞等待，直到操作成功
在操作時如不能立即滿足，則會阻塞等待給定的時間長度，時間到達(dá)后如果還不能滿足則返回null

這四類方法總結(jié)如下。

因為在BlockingQueue的一些方法中，會通過null表示某種操作的失敗，所以不允許在BlockingQueue中存放null值元素，會在操作時拋出NullPointerExection異常。

BlockingQueue因為是一個容器嘛，所以它也有容量的限制，在具體實現(xiàn)類中有可以設(shè)置容量的實現(xiàn)類，也有不可以設(shè)置容量的實現(xiàn)類，不能設(shè)置容量的實現(xiàn)類容量默認(rèn)為Integer.MAX_VALUE。

BlockingQueue是定義在java.util.concurrent包中，那么它在并發(fā)情況下到底是不是線程安全的呢？

在JDK提供的BlockingQueue的具體實現(xiàn)類中，上面表格中的方法實現(xiàn)都是線程安全的，在內(nèi)部都使用了鎖或者其他形式的并發(fā)控制保證操作的原子性。

但是有一點要注意，就是一些批量處理的方法例如addAll、containsAll、retainAll和removeAll這些方法并不一定是線程安全的，使用時注意。

說完BlockingQueue接口我們接下來看看它都有哪些具體的實現(xiàn)呢？以及在它們內(nèi)部是如何做到線程安全和阻塞的呢？

ArrayBlockingQueue

ArrayBlockingQueue是一個底層由數(shù)組支持額有界阻塞隊列。

重要屬性

先來看看ArrayBlockingQueue中都有哪些屬性。

// 存放元素的數(shù)組
final Object[] items;
// 用來記錄取元素的下標(biāo)，用于下一次在take，poll,remove，peek方法中使用
int takeIndex;
// 用來記錄添加元素的下標(biāo)，用于下一次put,offer,add等方法使用
int putIndex;
// 記錄隊列中元素數(shù)量
int count;
// 用于控制并發(fā)訪問時保證線程安全的鎖
final ReentrantLock lock;
// 用于隊列空時阻塞和喚醒等待線程的條件
private final Condition notEmpty;
// 用于隊列滿時阻塞和喚醒等待線程的條件
private final Condition notFull;

我們通過這些隊列中的屬性基本可以知道ArrayBlockingQueue中都有哪些重要信息，可以看出ArrayBlockingQueue就是使用Object[]來存放元素的。

那么應(yīng)該如何創(chuàng)建一個ArrayBlockingQueue呢？

構(gòu)造方法

public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}

默認(rèn)的構(gòu)造方法需要傳入一個int類型的capacity表示該隊列的容量。在該構(gòu)造方法中會調(diào)用另一個構(gòu)造方法，傳入一個默認(rèn)值false。

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();
}

從這個方法我們看出傳入的false表示會在內(nèi)部用于創(chuàng)建一個ReentrantLock對象，我們都知道ReentrantLock支持公平和非公平的實現(xiàn)，我們猜想一下，這里的這個fair值是不是表示該阻塞隊列對于阻塞排隊的線程支持公平和非公平的策略呢？這里先賣個關(guān)子，在后面的方法中我們具體說。

除了這兩種創(chuàng)建的方式，ArrayBlockingQueue還支持傳入一個Collection集合。

public ArrayBlockingQueue(int capacity, boolean fair,
                              Collection<? extends E> c) {
    // 先創(chuàng)建一個ArrayBlockingQueue實例
    this(capacity, fair);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        int i = 0;
        try {
            // 循環(huán)將collection中的元素放入queue中
            for (E e : c) {
                checkNotNull(e);
                items[i++] = e;
            }
        } catch (ArrayIndexOutOfBoundsException ex) {
            // 如果collection的元素個數(shù)超出queue的容量大小，會拋出異常
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;
    } finally {
        lock.unlock();
    }
}

添加元素

先來看看添加一個新元素到ArrayBlockingQueue是如何實現(xiàn)的，怎樣保證線程安全的。

add(e)

public boolean add(E e) {
    // 調(diào)用父類中的add(e)方法
    return super.add(e);
}
public boolean add(E e) {
    // 這里會直接調(diào)用offer(e)方法，如果offer方法返回false,則直接拋出異常
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}

add方法的實現(xiàn)邏輯本質(zhì)上是對offer方法套了一層殼，如果offer方法返回false時，拋出異常。所以我們直接看offer方法的實現(xiàn)就好。

offer(e)

public boolean offer(E e) {
    // 這里先判斷空，如果e為空會拋出空指針異常
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
	// 加鎖，保證入隊操作的原子性
    lock.lock();
    try {
        // 隊列滿時直接返回false
        if (count == items.length)
            return false;
        else {
            // 元素入隊
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

可以看到offer方法的邏輯還是比較簡單的，先檢查入?yún)⒉荒転榭?，然后加鎖保證入隊操作的原子性，在獲取鎖成功后入隊，如果隊列已滿則直接返回false，所以offer方法并不會阻塞。

put(e)

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
	// 可被中斷方式獲取鎖
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            // 隊列滿時會阻塞
            notFull.await();
        // 入隊
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

put方法和offer方法唯一的區(qū)別，就是會在隊列滿的時候使用Condition條件對象notFull阻塞等待。

private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    // 入隊成功，喚醒等待的移除元素操作線程
    notEmpty.signal();
}

在enqueue方法中才會完成對隊列中的數(shù)組元素的賦值動作，完成之后喚醒阻塞等待的移除元素操作線程。

offer(e,time,unit)

public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
    checkNotNull(e);
    // 加鎖之前先獲取需要等待的時間值
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length) {
            // 時間小于等于0時，返回false
            if (nanos <= 0)
                return false;
            // 阻塞等待指定時間
            nanos = notFull.awaitNanos(nanos);
        }
        enqueue(e);
        return true;
    } finally {
        lock.unlock();
    }
}

offer(e,time,unit)方法與offer(e)方法相比，主要時多了一個等待時間，會在時間到達(dá)時如果沒有空間添加元素返回false。

移除元素

ArrayBlockingQueue中移除元素的方法主要有remove()，poll()，take()，poll(time,unit)四個。這幾個方法的實現(xiàn)邏輯都比較簡單，這里不在單獨貼代碼。我們來看一下阻塞方法take()的實現(xiàn)即可。

take()

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
	// 加鎖
    lock.lockInterruptibly();
    try {
        while (count == 0)
            // 如果元素數(shù)量==0，表示隊列中為空，則阻塞等待
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

dequeue()

private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    // 取出元素之后，喚醒其他等待線程。
    notFull.signal();
    return x;
}

LinkedBlockingQueue

LinkedBlockingQueue是一個基于鏈表結(jié)構(gòu)的阻塞隊列，可以在創(chuàng)建時指定邊界大小，也可以不指定，在不指定邊界時容量為Integer.MAX_VALUE。

重要屬性

我們先來看看在LinkedBlockingQueue中都有哪些重要的屬性。

// 內(nèi)部類Node節(jié)點，用來存放鏈表中的元素
static class Node<E> {
    // 節(jié)點元素
	E item;
	// 當(dāng)前節(jié)點的下一個節(jié)點，如果為空表示沒有下一個節(jié)點
	Node<E> next;
	Node(E x) { item = x; }
}
// 隊列的容量
private final int capacity;
// 隊列中元素的數(shù)量
private final AtomicInteger count = new AtomicInteger();
// 頭節(jié)點
transient Node<E> head;
// 最后一個節(jié)點
private transient Node<E> last;
// 獲取元素時控制線程安全的鎖
private final ReentrantLock takeLock = new ReentrantLock();
// 添加元素時控制線程安全的鎖
private final ReentrantLock putLock = new ReentrantLock();
// 控制消費者的條件
private final Condition notEmpty = takeLock.newCondition();
// 控制生產(chǎn)者的條件
private final Condition notFull = putLock.newCondition();

在LinkedBlockingQueue中使用Node來存放元素，和指向下一個節(jié)點的鏈表指針。

構(gòu)造方法

在LinkedBlockingQueue的構(gòu)造方法中，會創(chuàng)建一個創(chuàng)建一個不存放元素的Node對象賦值給head和last。

public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    // 創(chuàng)建一個不存放元素的Node對象賦值給head和last
    last = head = new Node<E>(null);
}
public LinkedBlockingQueue(Collection<? extends E> c) {
        this(Integer.MAX_VALUE);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            int n = 0;
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (n == capacity)
                    throw new IllegalStateException("Queue full");
                // 入隊
                enqueue(new Node<E>(e));
                ++n;
            }
            count.set(n);
        } finally {
            putLock.unlock();
        }
    }

添加元素

offer(e)

public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final AtomicInteger count = this.count;
    if (count.get() == capacity)
        return false;
    int c = -1;
    Node<E> node = new Node<E>(e);
    // 使用putLock加鎖
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        if (count.get() < capacity) {
            // 入隊
            enqueue(node);
            // 數(shù)量+1
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                // 喚醒一個生產(chǎn)者線程
                notFull.signal();
        }
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        // 喚醒消費者線程
        signalNotEmpty();
    // 入隊失敗情況會返回false
    return c >= 0;
}

對于鏈表結(jié)構(gòu)的LinkedBlockingQueue來說，入隊操作要簡單很多，只需要將node節(jié)點掛在最后一個節(jié)點last的next，然后將自己賦值給last。

private void enqueue(Node<E> node) {
    last = last.next = node;
}

put(e)

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -1;
    Node<E> node = new Node<E>(e);
    // 使用putLock加鎖
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        while (count.get() == capacity) {
            // 如果隊列容量已使用完則阻塞
            notFull.await();
        }
        enqueue(node);
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
}

對比結(jié)果也和我們最開始的方法匯總表格一樣，offer(e)方法會在入隊時如果隊列已滿直接返回false，而put(e)會一直阻塞等待，知道入隊成功。

add(e)方法和offer(e,time,unit)方法實現(xiàn)邏輯上沒有特殊之處，這里不再放源碼。

移除元素

poll()

public E poll() {
        final AtomicInteger count = this.count;
        if (count.get() == 0)
            return null;
        E x = null;
        int c = -1;
    // 使用takeLock加鎖
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            if (count.get() > 0) {
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    // 還有元素時喚醒一個生產(chǎn)者線程
                    notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            // 喚醒生產(chǎn)者線程
            signalNotFull();
        return x;
    }

poll()方法會在元素出隊時如果沒有元素則直接返回null。

// 出隊方法
private E dequeue() {
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; 
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

take()

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    // 使用takeLock加鎖
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
        while (count.get() == 0) {
            //阻塞等待
            notEmpty.await();
        }
        x = dequeue();
        c = count.getAndDecrement();
        if (c > 1)
            // 還有元素時喚醒一個消費者線程
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        // 喚醒生產(chǎn)者線程
        signalNotFull();
    return x;
}

同樣，take方法會在沒有元素時一直等待。

對比

我們來對比一下ArrayBlockingQueue和LinkedBlockingQueue都有哪些區(qū)別。

ArrayBlockingQueue基于數(shù)組實現(xiàn)，LinkedBlockingQueue基于鏈表實現(xiàn)
ArrayBlockingQueue在添加和移除元素的操作中共用一把鎖，LinkedBlockingQueue使用takeLock和putLock兩把鎖
ArrayBlockingQueue在添加和移除元素時直接使用元素的類型處理，LinkedBlockingQueue需要轉(zhuǎn)成Node對象
ArrayBlockingQueue創(chuàng)建時必須指定容量，LinkedBlockingQueue可以不指定，默認(rèn)容量為Integer.MAX_VALUE

由于LinkedBlockingQueue使用兩把鎖將入隊操作和出隊操作分離，這會大大提高隊列的吞吐量，在高并發(fā)情況下生產(chǎn)者和消費者可以并行處理，提高并發(fā)性能。

但是LinkedBlockingQueue默認(rèn)是無界隊列，要小心內(nèi)存溢出風(fēng)險，所以最好在創(chuàng)建時指定容量大小。

BlockingQueue接口的實現(xiàn)類除了本期介紹的這兩種，還有PriorityBlockingQueue，SynchronousQueue，LinkedBlockingDeque等，每一個都有它獨特的特性和使用場景，后面我們再單獨深入解析。