併發隊列之LinkedBlockingQueue

時間  2020-02-09
標籤 併發 隊列 linkedblockingqueue 欄目 Java 简体版
原文   原文鏈接

　　上一篇咱們看了一下這個隊列ConcurrentLinkedQueue，那就是一個無界非阻塞鏈表，咱們此次來看看LinkedBlockingQueue，這個隊列看名字就知道是一個阻塞式隊列(也就是一個單向鏈表)，基於獨佔鎖實現的，比較簡單；node

 

一.LinkedBlockingQueue基本結構併發

　　內部也是有一個Node類，下圖所示，item存 實際數據，next指向下一個節點，一個有參構造器，沒啥好說的；this

 

　　咱們能夠看看這個隊列有的一些屬性，其實大概能猜出來就是生產者消費者模型：spa

//隊列實際容量
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();
    }
}

 

　　咱們簡單的介紹了這個隊列的基本結構，如今咱們能夠看看一些重要的方法；　code

 

二.offer方法blog

　　這個方法向隊列最後添加一個元素，插入成功返回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方法隊列

　　這個方法在隊列尾部插入一個元素，若是隊列有空閒則插入後直接返回，不然就阻塞當前線程直到隊列空閒再插入；並且當前線程在阻塞的時候被其餘線程調用了中斷方法，就會拋異常；ci

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隊列，一個是條件隊列，一個是存放數據的隊列，不要弄混淆了啊！

　　用下面這個圖強化記憶：

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程