1. 程式人生 > >併發佇列之LinkedBlockingQueue

併發佇列之LinkedBlockingQueue

  上一篇我們看了一下這個佇列ConcurrentLinkedQueue,那就是一個無界非阻塞連結串列,我們這次來看看LinkedBlockingQueue,這個佇列看名字就知道是一個阻塞式佇列(也就是一個單向連結串列),基於獨佔鎖實現的,比較簡單;

 

一.LinkedBlockingQueue基本結構

  內部也是有一個Node類,下圖所示,item存 實際資料,next指向下一個節點,一個有參構造器,沒啥好說的;

 

  我們可以看看這個佇列有的一些屬性,其實大概能猜出來就是生產者消費者模型:

//佇列實際容量
private final int capacity;
//這個原子變數記錄節點數量
private final AtomicInteger count = new AtomicInteger();
//頭節點
transient Node<E> head;
//尾節點
private transient Node<E> last;
//這個鎖用於控制多個執行緒從佇列頭部獲取元素
private final ReentrantLock takeLock = new ReentrantLock();
//當佇列為空,執行出隊操作的執行緒就放到這條件變數裡來
private final Condition notEmpty = takeLock.newCondition();
//用於控制多個執行緒往佇列尾部新增元素
private final ReentrantLock putLock = new ReentrantLock();
//如果佇列滿了,執行入隊操作的執行緒就丟到這裡面來
private final Condition notFull = putLock.newCondition();

 

  構造器中可以看到,預設最大數量就是65536個,雖然說也可以指定大小,我們一定程度上可以說這是一個有界阻塞佇列;

//預設佇列最大的數量就是65536個
public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }
//也可以指定佇列大小,預設頭節點和尾節點都是指向哨兵節點
public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    last = head = new Node<E>(null);
}
//也可以傳一個實現Collection介面的類,比如List,然後遍歷將其中的元素封裝成節點丟到佇列中去
//注意這裡獲取鎖和釋放鎖
public LinkedBlockingQueue(Collection<? extends E> c) {
    this(Integer.MAX_VALUE);
    final ReentrantLock putLock = this.putLock;
    putLock.lock(); // Never contended, but necessary for visibility
    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方法

  這個方法向佇列最後新增一個元素,插入成功返回true,如果佇列滿了,就拋棄當前元素返回false;

public boolean offer(E e) {
    //如果為null,就直接拋錯
    if (e == null) throw new NullPointerException();
    //count表示佇列中實際數量
    final AtomicInteger count = this.count;
    //如果實際數量和佇列最大容量相同,那麼就不能再添加了,返回false
    if (count.get() == capacity)
        return false;
    int c = -1;
    //將元素封裝成Node節點
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    //獲取鎖
    putLock.lock();
    try {
        //佇列沒有滿,就把新的節點丟進去,遞增計數器,為什麼這裡要進行判斷呢?上面不是進行判斷了麼?
        //還是由於併發,如果在上面那裡判斷容量之後,當前執行緒還沒有獲取鎖,此時一個其他執行緒先獲取鎖然後執行offer方法然後釋放鎖,那麼
        //這裡需要再判斷一次
        if (count.get() < capacity) {
            enqueue(node);
        //注意getAndIncrement和incrementAndGet方法的區別,前者是返回自增之前的值,後者是返回自增之後的值 c = count.getAndIncrement(); //這裡判斷如果佇列還沒有滿,就喚醒之前notFully條件佇列中的執行緒,前面說過了notfull中存放的是什麼執行緒 if (c + 1 < capacity) notFull.signal(); } } finally { //釋放鎖 putLock.unlock(); } //如果c==0,表示佇列中在新增節點之前就已經有一個節點了,就喚醒條件變數notEmpty中的執行緒,這些執行緒就會從佇列中去取資料 if (c == 0) signalNotEmpty(); return c >= 0; } private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); } }

 

三.put方法

  這個方法在佇列尾部插入一個元素,如果佇列有空閒則插入後直接返回,否則就阻塞當前執行緒直到佇列空閒再插入;而且當前執行緒在阻塞的時候被其他執行緒呼叫了中斷方法,就會拋異常;

public void put(E e) throws InterruptedException {
    //如果插入的元素為null,直接拋異常
    if (e == null) throw new NullPointerException();
    int c = -1;
    //封裝成一個節點
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    //原子變數記錄佇列中節點數量
    final AtomicInteger count = this.count;
    //這個方法後面帶有Interruptibly,說明當前執行緒獲取鎖後是可中斷的
    putLock.lockInterruptibly();
    try {
        //節點數量到達了最大容量,就將當前執行緒放到條件變數notFull的佇列中
        while (count.get() == capacity) {
            notFull.await();
        }
        //節點數量沒有到最大,就在連結串列最後新增節點
        enqueue(node);
        //計數器加一,注意如果count等於4,那麼c還是等於4,這個方法是原子自增,返回原來的值,注意和incrementAndGet方法的區別
        c = count.getAndIncrement();
        //這裡如果c+1<capacity的話,說明佇列還沒有滿,就喚醒notFull中的執行緒可以往佇列中新增元素
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        //釋放鎖
        putLock.unlock();
    }
    //如果c==0表示佇列中新增節點之前就已經有了一個節點,喚醒notEmpty中的執行緒去佇列中取資料
    if (c == 0)
        signalNotEmpty();
}

private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
}

 

四.poll方法

  這個方法是從頭部移除一個元素,如果佇列為空就返回null,這個方法不阻塞;

public E poll() {
    //記錄佇列中節點數量
    final AtomicInteger count = this.count;
    //如果佇列為空就返回null
    if (count.get() == 0)
        return null;
    E x = null;
    int c = -1;
    final ReentrantLock takeLock = this.takeLock;
    //獲取鎖
    takeLock.lock();
    try {
        //這裡再判斷一次也是為了防止在獲取鎖之前其他執行緒呼叫了poll方法取了節點
        //如果佇列不為空,計數器減一
        if (count.get() > 0) {
            //刪除第一個有資料的節點(由於第一個節點是哨兵節點,所以相當於刪除的是第二個節點),方法實現在下面
            x = dequeue();
            c = count.getAndDecrement();
            //如果c>1說明移除頭節點之後,佇列不為空,就喚醒notEmpty中條件佇列中的執行緒去佇列中取資料
            if (c > 1)
                notEmpty.signal();
        }
    } finally {
        //釋放鎖
        takeLock.unlock();
    }
    //這裡這個判斷,注意一下在原子類AtomicInteger中兩個方法,比如初始值為5,那麼呼叫decrementAndGet方法返回的事4,
    //而呼叫getAndDecrement方法返回的是5,我們這裡額c呼叫的是後者,所以表示刪除節點之前佇列的數量
    //所以這裡的意思就是:如果刪除佇列之前的數量等於佇列最大容量,那麼刪除之後佇列肯定有空位置,於是就喚醒notFull條件佇列中的執行緒
    //往佇列中新增新的節點
    if (c == capacity)
        signalNotFull();
    return x;
}
//這個方法很簡單稍微提一下,就是將第一個哨兵節點自己引用自己,更好的被gc收集
//將head指向第二個節點,取出該節點的值,然後將該節點內的item置為null,此節點就變成了一個哨兵節點
private E dequeue() {
    //刪除頭節點
    Node<E> h = head;
    Node<E> first = h.next;
    h.next = h; // help GC
    head = first;
    E x = first.item;
    first.item = null;
    return x;
}

private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        notFull.signal();
    } finally {
        putLock.unlock();
    }
}

 

 

 五.peek方法

  這個方法獲取佇列頭部元素但是不移除該節點;

public E peek() {
    //佇列為空,返回null
    if (count.get() == 0)
        return null;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        //如果頭節點的下一個節點是空,就返回null,因為此時頭節點指的是哨兵節點啊
        Node<E> first = head.next;
        if (first == null)
            return null;
        else
        //頭節點的下一個節點不為空,那麼下一個節點肯定有資料,拿過來就行了
            return first.item;
    } finally {
        takeLock.unlock();
    }
}

 

 

六.take方法

  當前方法跟peek方法基本一樣,只不過這個方法是阻塞的:從佇列頭刪除一個節點,如果佇列為空則阻塞當前執行緒直到佇列不為空再執行操作,如果在阻塞的時候其他執行緒修改了中斷標誌,那麼該執行緒就拋錯;

//這個方法其實和poll方法基本一樣,沒什麼好說的,注意可以拋異常
public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    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)
        signalNotFull();
    return x;
}

 

 

七.remove方法

  刪除佇列中某個指定的元素,刪除成功返回true,失敗返回false;

public boolean remove(Object o) {
    //傳入的是null,直接返回false
    if (o == null) return false;
    //獲取兩個鎖,方法實現如下,在刪除節點的時候,不能進行入隊和出隊操作,那些執行緒會被阻塞丟到AQS佇列裡
    fullyLock();
    try {
        //遍歷找到對應的節點,刪除掉,沒找到返回false
        for (Node<E> trail = head, p = trail.next; p != null; trail = p, p = p.next) {
            if (o.equals(p.item)) {
                //刪除節點的具體方法,實現在下面
                unlink(p, trail);
                return true;
            }
        }
        return false;
    } finally {
        //釋放兩個鎖,注意釋放鎖的順序要和獲取鎖的順序相反喲
        fullyUnlock();
    }
}

void fullyLock() {
    putLock.lock();
    takeLock.lock();
}
void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

//刪除其實很容易,就是跟普通連結串列的刪除一樣,就是把當前要刪除的節點前面的節點指向後面的節點
void unlink(Node<E> p, Node<E> trail) {
    p.item = null;
    trail.next = p.next;
    if (last == p)
        last = trail;
    //還是注意getAndDecrement方法返回的是減一之前的值,如果減一之後佇列不是滿的,就喚醒notFull中條件佇列中的執行緒往佇列中新增節點
    if (count.getAndDecrement() == capacity)
        notFull.signal();
}

 

  

八.總結

  我們簡單的看了看LinkedBlockingQueue中一些比較重要的方法,比上一篇的ConcurrentLinkedQueue容易好多呀!

  其中ConcurrentLinkedQueue是無界非阻塞佇列,底層是用單向連結串列實現,入隊和出隊使用CAS實現;而LinkedBlockingQueue是有界阻塞佇列,底層是用單向連結串列實現,入隊和出隊分別用獨佔鎖的方式去處理,所以入隊和出隊是可以同時進行的,而且還為兩個獨佔鎖配置了兩個條件佇列,用於存放被阻塞的線層,注意,這裡涉及到好幾個佇列,一個是獨佔鎖的AQS佇列,一個是條件佇列,一個是存放資料的佇列,不要弄混淆了啊!

  用下面這個圖強化記憶:

&n