併發隊列之LinkedBlockingQueue

来源:https://www.cnblogs.com/wyq1995/archive/2020/02/08/12285317.html
-Advertisement-
Play Games

上一篇我們看了一下這個隊列ConcurrentLinkedQueue，那就是一個無界非阻塞鏈表，我們這次來看看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隊列，一個是條件隊列，一個是存放數據的隊列，不要弄混淆了啊！

　　用下麵這個圖強化記憶：

 


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • vuex源碼簡析
    前言 基於 vuex 3.1.2 按如下流程進行分析： Vue.use(Vuex) Vue.use() 會執行插件的 install 方法，並把插件放入緩存數組中。 而 Vuex 的 install 方法很簡單，保證只執行一次，以及使用 applyMixin 初始化。 applyMixin 方法定義 ...
  • 使用placeholder屬性設置input文本框的提示信息
    input文本框中設置提示信息,可以使用placeholder屬性,結果就是這樣,有當文本框內容為空時就顯示提示信息,當輸入內容時提示內容就消失了 ...
  • 什麼是數據結構？
    本篇文章主要來介紹什麼是數據結構。 首先讓我們來看一張圖片： 數據存儲於電腦的記憶體中。記憶體如上圖所示，形似排成 1 列的箱子，1 個箱子里存儲 1 個數據。 數據存儲於記憶體時， 決定了數據順序和位置關係的便是數據結構 。 其實在我們生活中用到很多數據結構的知識，那麼舉一個我們生活中的慄子： 首先舉 ...
  • 什麼是鏈表？
    在瞭解完 "什麼是數據結構" 之後，讓我們一起來探索下數據結構中常見的一種— 鏈表 。 鏈表 鏈表是數據結構之一， 其中的數據呈線性排列。在鏈表中，數據的添加和刪除都較為方便，就是訪問比較耗費時間。 如上圖所示就是鏈表的概念圖，Blue、Yellow、Red 這 3 個字元串作為數據被存儲於鏈表中， ...
  • Java連載80-數字類格式、隨機數、BigDecimal
    一、數字類 1.關於數字格式化：java.text.DecimalFormat; 2.數字格式元素： # 任意數字 ， 千分位 . 小數點 0 不夠補零 package com.bjpowernode.java_learning; ​ import java.text.DecimalFormat; ...
  • JavaSE學習筆記（6）---異常
    JavaSE學習筆記（6） 異常 ​ 軟體程式在運行過程中，非常可能遇到問題，我們稱之為異常，英文是：Exception，意思是例外。遇到這些例外情況，或者叫異常，我們怎麼讓寫的程式做出合理的處理，安全的退出，而不至於程式崩潰呢? 1.異常產生的原因及使用原則 在 Java 中一個異常的產生，主要有 ...
  • linux安裝jdk並設置環境變數（看這一篇文章即可）
    1.查看linux位數 查看linux是32位還是64位，影響需要下載JDK的版本 系統位數 jdk位數 x86（32位） 32位 x86_64（64位） 32位 64位 在linux命令輸入： uname -a 如果是64位機器，會輸出x86_64 2.下載JDK 下載地址：https://www ...
  • 併發編程之線程第二篇
    併發編程之線程第二篇 3.12 五種狀態 3.13 六種狀態 4.1 共用帶來的問題 4.2 synchronized解決方案 4.4 變數的線程安全分析 4.6 Monitor概念 1. 輕量級鎖 2. 鎖膨脹 3.12 五種狀態 這是從操作系統層面來描述的 【初始狀態】僅是在語音層面創建了線程對 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類