前面一篇文章我們介紹了使用CAS算法實(shí)現(xiàn)的非阻塞隊(duì)列ConcurrentLinedQueue, 下面我們來介紹使用獨(dú)占鎖實(shí)現(xiàn)的阻塞隊(duì)列LinkedBlockingQueue。

LinkedBlockingQueue也是使用單向鏈表實(shí)現(xiàn)的，其也有兩個(gè)Node，分別用來存放首、尾節(jié)點(diǎn)，并且還有一個(gè)初始值為0的原子變量count，用來記錄隊(duì)列元素個(gè)數(shù)。另外還有兩個(gè)ReentrantLock的實(shí)例，分別用來控制元素入隊(duì)和出隊(duì)的原子性，其中takeLock用來控制同時(shí)只有一個(gè)線程可以從隊(duì)列頭獲取元素，其他線程必須等待，putLock控制同時(shí)只能有一個(gè)線程可以獲取鎖，在隊(duì)列尾部添加元素，其他線程必須等待。另外，notEmpty 和 notFull 是條件變量，它們內(nèi)部都有一個(gè)條件隊(duì)列用來存放進(jìn)隊(duì)和出隊(duì)時(shí)被阻塞的線程，其實(shí)這是生產(chǎn)者-消費(fèi)者模型。如下是獨(dú)占鎖的創(chuàng)建代碼。

private final AtomicInteger count = new AtomicInteger();

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();

/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

• 當(dāng)調(diào)用線程在LinkedBlockingQueue 實(shí)例上執(zhí)行take、poll 等操作時(shí)需要獲取到 takeLock 鎖，從而保證同時(shí)只有一個(gè)線程可以操作鏈表頭節(jié)點(diǎn)。另外由于條件變量 notEmpty 內(nèi)部的條件隊(duì)列的維護(hù)使用的是takeLock的鎖狀態(tài)管理機(jī)制，所以在調(diào)用notEmpty的await 和signal方法前調(diào)用線程必須先獲取到 takeLock鎖，否則會(huì)拋出IllegalMonitorStateException 異常。notEmpty內(nèi)部則維護(hù)著一個(gè)條件隊(duì)列，當(dāng)線程獲取到takeLock 鎖后調(diào)用 notEmpty的await 方法時(shí)，調(diào)用線程會(huì)被阻塞，然后該線程會(huì)被放到notEmpty內(nèi)部的條件隊(duì)列進(jìn)行等待，直到有線程調(diào)用了notEmpty的 signal 方法。

• 在LinkedBlockingQueue實(shí)例上執(zhí)行put、offer等操作時(shí)需要獲取到putLock鎖，從而保證同時(shí)只有一個(gè)線程可以操作鏈表尾節(jié)點(diǎn)。同樣由于條件變量 notFull 內(nèi)部的條件隊(duì)列的維護(hù)使用的是putLock的鎖狀態(tài)管理機(jī)制，所以在調(diào)用 notFull 的 await 和 signal 方法前調(diào)用線程必須先獲取到putLock鎖，否則會(huì)拋出 IllegalMonitorStateException 異常。notFull 內(nèi)部則維護(hù)著一個(gè)條件隊(duì)列，當(dāng)線程獲取到 putLock 鎖后調(diào)用notFull的await 方法時(shí)，調(diào)用線程會(huì)被阻塞，然后該線程會(huì)被放到notFull 內(nèi)部的條件隊(duì)列進(jìn)行等待，直到有線程調(diào)用了 notFull 的 signal 方法。如下是LinkedBlockingQueue 的無參構(gòu)造函數(shù)的代碼。

如下是LinkedBlockingQueue的無參構(gòu)造代碼

public static final int MAX_VALUE = 0x7fffffff;
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}


public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalAgrumentException();
    this.capacity = capacity;
    last = head = new Node<E>(null);
}

由該代碼可知，默認(rèn)隊(duì)列容量為0x7fffffff，用戶也可以自己指定容量，所以從一定程度上可以說LinkedBlockingQueue是有界阻塞隊(duì)列。

offer操作

public boolean offer(E e) {
//(1)
    if (e == null) throw new NullPointerException();
    //(2)
    final AtomicInteger count = this.count;
    if (count.get() == capacity)
        return false;
        //(3)
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
    //(4)
        if (count.get() < capacity) {
            enqueue(node);
            c = count.getAndIncrement();
            //(5)
            if (c + 1 < capacity)
                notFull.signal();
        }
    } finally {
    //(6)
        putLock.unlock();
    }
    //(7)
    if (c == 0)
        signalNotEmpty();
        //(8)
    return c >= 0;
}

代碼(2)判斷如果當(dāng)前隊(duì)列已滿則丟棄當(dāng)前元素并返回false

代碼(3)獲取到 putLock 鎖，當(dāng)前線程獲取到該鎖后，則其他調(diào)用put和 offer操的線程將會(huì)被阻塞(阻塞的線程被放到putLock鎖的AQS阻塞隊(duì)列)。

代碼(4)這里重新判斷當(dāng)前隊(duì)列是否滿，這是因?yàn)樵趫?zhí)行代碼(2)和獲取到 putLock 鎖期間可能其他線程通過 put 或者offer 操作向隊(duì)列里面添加了新元素。重新判斯隊(duì)列確實(shí)不滿則新元素入隊(duì)，并遞增計(jì)數(shù)器。

代碼(5)判斷如果新元素入隊(duì)后隊(duì)列還有空閑空間，則喚醒notFull的條件隊(duì)列里面因?yàn)檎{(diào)用了notFull的await操作(比如執(zhí)行put方法而隊(duì)列滿了的時(shí)候)而被阻塞的一個(gè)線程，因?yàn)殛?duì)列現(xiàn)在有空閑所以這里可以提前喚醒一個(gè)入隊(duì)線程。

代碼(6)則釋放獲取的putLock 鎖，這里要注意，鎖的釋放一定要在finally里面做因?yàn)榧词箃ry塊拋出異常了，finally也是會(huì)被執(zhí)行到。另外釋放鎖后其他因?yàn)檎{(diào)用put 操作而被阻塞的線程將會(huì)有一個(gè)獲取到該鎖。

代碼(7)中的c0說明在執(zhí)行代碼(6)釋放鎖時(shí)隊(duì)列里面至少有一個(gè)元素，隊(duì)列里面有元素則執(zhí)行signalNotEmpty操作.

到此這篇關(guān)于深入理解Java并發(fā)編程之LinkedBlockingQueue隊(duì)列的文章就介紹到這了,更多相關(guān)Java LinkedBlockingQueue隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論