java多線程系列(九)---ArrayBlockingQueue源碼分析

時間 2019-11-07

標籤 java 多線程系列 arrayblockingqueue 源碼分析欄目 Java 简体版

原文原文鏈接

java多線程系列(九)---ArrayBlockingQueue源碼分析

成員變量

數組大小

final Object[] items;

下一個進隊元素的下標

/** items index for next take, poll, peek or remove */
    int takeIndex;

下一個出隊元素的下標

/** items index for next put, offer, or add */
    int putIndex;

隊列中元素的數目

/** Number of elements in the queue */
    int count;

出隊和入隊須要的鎖

/*
     * Concurrency control uses the classic two-condition algorithm
     * found in any textbook.
     */

    /** Main lock guarding all access */
    final ReentrantLock lock;

出隊條件

/** Condition for waiting takes */
    private final Condition notEmpty;

入隊條件

/** Condition for waiting puts */
    private final Condition notFull;

構造方法

配置容量，先建立是否公平公平訪問

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

從源碼能夠看到，建立一個object數組，而後建立一個公平或非公平鎖，而後建立出隊條件和入隊條件

offer方法

public boolean offer(E e) {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

首先檢查是否爲null
而後lock鎖住
若是當前數目count已經爲初始的時候容量，這時候本身返回false
不然的話執行enqueue方法

private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        notEmpty.signal();
    }

將新的元素添加到數組的下一個進隊的位置
而後notEmpty出隊條件喚醒，這個時候能夠進行出隊
執行enqueue後而後釋放鎖

指定時間的offer方法

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) {
                if (nanos <= 0)
                    return false;
                nanos = notFull.awaitNanos(nanos);
            }
            enqueue(e);
            return true;
        } finally {
            lock.unlock();
        }
    }

方法大體和前面的一致，不一樣的時候是當隊列滿的時候，會等待一段時間，此時入隊條件等待一段時間，一段時間後繼續進入循環進行判斷隊列還滿
當隊列不滿的時候執行enqueue

add方法

調用父類的add方法

public boolean add(E e) {
        return super.add(e);
    }

父類AbstractQueue的add方法

public boolean add(E e) {
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }

執行offer方法，這個時候能夠對比上面直接調用offer，offer方法若是入隊失敗會直接返回false，而add方法會拋出異常

put方法

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

和限定時間的offer方法不一樣，當隊列滿的時候，會一直等待，直到有人喚醒

poll方法

public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return (count == 0) ? null : dequeue();
        } finally {
            lock.unlock();
        }
    }

首先執行lock方法鎖定
若是當前隊中無元素，那麼返回null，不然執行dequeue方法

private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        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;
    }

根據出隊下標取出元素，而後將該位置置爲null
將出隊下標加一，若是出隊下標等於了數組的大小，出隊下標置爲0
隊中元素數量減一
notFull喚醒，此時能夠喚醒入隊阻塞的線程

指定時間的poll方法

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0) {
                if (nanos <= 0)
                    return null;
                nanos = notEmpty.awaitNanos(nanos);
            }
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

與offer的指定時間和沒有指定時間相似，poll指定時間的方法和沒有指定時間的poll思路大體是同樣的
當此時隊列爲空的，爲等待一段時間，而後自動喚醒，繼續進入循環，直到隊列中有元素，而後執行dequeue方法

take方法

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

和前面指定時間的poll方法不一樣，當隊中爲空的時候，會一直等待，直到被喚醒

總結

入隊

方法	特色
offer(E e)	隊列滿的時候，返回false
offer(E e, long timeout, TimeUnit unit)	隊列滿的時候，等待一段時間，釋放鎖，一段時間後，進入就緒狀態
add(E e)	隊列滿的時候，直接拋出異常
put(E e)	隊列慢的時候，線程阻塞，直到被喚醒

出隊

方法	特色
poll()	隊列爲空的時候，直接返回null
poll(long timeout, TimeUnit unit)	隊列爲空的時候，等待一段時間，釋放鎖，一段時候後，進入就緒狀態
take()	隊列爲空的時候，一直等待，釋放鎖，直到被喚醒

整體的設計思路，經過一個數組來模擬一個數組，出隊和入隊都是同步的，也就是同一時間只能有一個入隊或者出隊操做，而後在入隊的時候，若是隊列已滿的話，根據方法的不一樣有不一樣的策略，能夠直接返回或者拋出異常，也能夠阻塞一段時間，等會在嘗試入隊，或者直接阻塞，直到有人喚醒。而出隊的時候，若是爲空能夠直接返回，也能夠等待一段時間而後再次嘗試，也能夠阻塞，直到有人喚醒

我以爲分享是一種精神，分享是個人樂趣所在，不是說我以爲我講得必定是對的，我講得可能不少是不對的，可是我但願我講的東西是我人生的體驗和思考，是給不少人反思，也許給你一秒鐘、半秒鐘，哪怕說一句話有點道理，引起本身心裏的感觸，這就是我最大的價值。（這是我喜歡的一句話，也是我寫博客的初衷）

做者：jiajun 出處： http://www.cnblogs.com/-new/
本文版權歸做者和博客園共有，歡迎轉載，但未經做者贊成必須保留此段聲明，且在文章頁面明顯位置給出原文鏈接，不然保留追究法律責任的權利。若是以爲還有幫助的話，能夠點一下右下角的【推薦】，但願可以持續的爲你們帶來好的技術文章！想跟我一塊兒進步麼？那就【關注】我吧。html