標籤: 網站設計

Java面試中經常會涉及關於volatile的問題。本文梳理下volatile關鍵知識點。

volatile字意為“易失性”，在Java中用做修飾對象變量。它不是Java特有，在C，C++，C#等編程語言也存在，只是在其它編程語言中使用有所差異，但總體語義一致。比如使用volatile 能阻止編譯器對變量的讀寫優化。簡單說，如果一個變量被修飾為volatile，相當於告訴系統說我容易變化，編譯器你不要隨便優化（重排序，緩存）我。

Happens-before

規範上，Java內存模型遵行happens-before。

volatile變量在多線程中，寫線程和讀線程具有happens-before關係。也就是寫值的線程要在讀取線程之前，並且讀線程能完全看見寫線程的相關變量。

happens-before：如果兩個有兩個動作AB，A發生在B之前，那麼A的順序應該在B前面並且A的操作對B完全可見。

happens-before 具有傳遞性，如果A發生在B之前，而B發生在C之前，那麼A發生在C之前。

如何保證可見性

多線程環境下counter變量的更新過程。線程1先從主存拷貝副本到CPU緩存，然後CPU執行counter=7，修改完后寫入CPU緩存，等待時機同步到主存。在線程1同步主存前，線程2讀到counter值依然為0。此時已經發生內存一致性錯誤（對於相同的共享數據，多線程讀到視圖不一致）。因為線程2看不見線程1操作結果，也將這個問題稱為可見性問題。

public class SharedObject {
    public int counter = 0;
}

因為多了緩存優化導致，導致可見性問題。所以volatile通過消除緩存（描述可能不太準確）來避免。例如當使用volatile修飾變量后，操作該變量讀寫直接與主存交互，跳過緩存層，保證其它讀線程每次獲取的都是最新值。

    public volatile int counter = 0;

volatile 不單隻消除修飾的變量的緩存。事實上與之相關的變量在讀寫時也會消除緩存，如同使用了volatile一樣。

如下 years，months，days 三個變量中只有days是volatile，但是對years，months讀寫操作也和days時也會跳過緩存，其它線程每次讀到的都是最新值。

public class MyClass {
    private int years;
    private int months
    private volatile int days;
    public int totalDays() {
        int total = this.days;
        total += months * 30;
        total += years * 365;
        return total;
    }
    public void update(int years, int months, int days){
        this.years  = years;
        this.months = months;
        this.days   = days;
    }
}

這是為什麼？我們分析一下。

一個寫線程調用 update，讀線程調用totalDays。單線程中，對於update方法，wa與wb存在happens-before關係， wa在 wb 之前執行並對wb可見。

多線程中rc與wb存在happens-before關係，wb在rc之前執行並對rc可見。根據 happens-before傳遞性，wa需要在rc前先執行並對rc可見。

因為wb是volatile變量，所以rc獲取的years，months也是最新值。

我們知道出於性能原因，JVM和CPU會對程序中的指令進行重新排序。如果update方法裏面wa和wb順序被重排，那它們的happens-before關係將不在成立。

為了避免這個問題，volatile對重排序做了保證 對於發生在volatile變量操作前的其他變量的操作不能重新排序。

由此我們得到volatile通過消除緩存和防止重排保證線程的可見性。

volatile保證線程安全？

討論線程安全，大家都會提及原子性，順序性，可見性。volatile側重於保證可見性，也就是當寫的線程更新后，讀線程總能獲得最新值。在只有一個線程寫，多個線程讀的場景下，volatile能滿足線程安全。可如果多個線程同時寫入volatile變量時，則需要引入同步語義才能保證線程安全。

模擬10個線程同時寫入volatile變量，一個線程讀counter，執行完后正確結果應該是counter=10。

    public static class WriterTask implements Runnable {
        private final ShareObject share;
        private final CountDownLatch countDownLatch;
        public WriterTask(ShareObject share, CountDownLatch countDownLatch) {
            this.share = share;
            this.countDownLatch = countDownLatch;
        }
        @Override
        public void run() {
            countDownLatch.countDown();
            share.increase();
        }
    }
    
    public class ShareObject {
        private volatile int counter;
        public void increase() {
            this.counter++;
        }
    }

執行結果出現counter=5或6 錯誤結果。

通過 synchronized，Lock或AtomicInteger 原子變量保證了結果的正確。

完整demo https://gist.github.com/onlythinking/ba7ca7aa5faf00a58f4cedae474fa6f6

volatile性能

volatile變量帶來可見性的保證，訪問volatile變量還防止了指令重排序。不過這一切是以犧牲優化（消除緩存，直接操作主存開銷增加）為代價，所以不應該濫用volatile，僅在確實需要增強變量可見性的時候使用。

總結

本文記錄了volatile變量通過消除緩存，防止指令重排序來保證線程可見性，並且在多線程寫入的變量的場景下，不保證線程安全。

歡迎大家留言交流，一起學習分享！！！

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※帶您來了解什麼是 USB CONNECTOR  ?

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站，提高曝光率!

※綠能、環保無空污,成為電動車最新代名詞，目前市場使用率逐漸普及化

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※教你寫出一流的銷售文案?

概述

在程序中，我們會用各種池化技術來緩存創建昂貴的對象，比如線程池、連接池、內存池。一般是預先創建一些對象放入池中，使用的時候直接取出使用，用完歸還以便復用，還會通過一定的策略調整池中緩存對象的數量，實現池的動態伸縮。

由於線程的創建比較昂貴，隨意、沒有控制地創建大量線程會造成性能問題，因此短平快的任務一般考慮使用線程池來處理，而不是直接創建線程。

那麼，如何正確的創建並正確的使用線程池呢，這篇文章就來細看下。

線程池

雖然在 Java 語言中創建線程看上去就像創建一個對象一樣簡單，只需要 new Thread() 就可以了，但實際上創建線程遠不是創建一個對象那麼簡單。

創建對象，僅僅是在 JVM 的堆里分配一塊內存而已；而創建一個線程，卻需要調用操作系統內核的 API，然後操作系統要為線程分配一系列的資源，這個成本就很高了。所以線程是一個重量級的對象，應該避免頻繁創建和銷毀，一般就是採用線程池來避免頻繁的創建和銷毀線程。

 

線程池原理

Java 通過用戶線程與內核線程結合的 1:1 線程模型來實現，Java 將線程的調度和管理設置在了用戶態。在 HotSpot VM 的線程模型中，Java 線程被一對一映射為內核線程。Java 在使用線程執行程序時，需要創建一個內核線程；當該 Java 線程被終止時，這個內核線程也會被回收。因此 Java 線程的創建與銷毀將會消耗一定的計算機資源，從而增加系統的性能開銷。

除此之外，大量創建線程同樣會給系統帶來性能問題，因為內存和 CPU 資源都將被線程搶佔，如果處理不當，就會發生內存溢出、CPU 使用率超負荷等問題。

為了解決上述兩類問題，Java 提供了線程池概念，對於頻繁創建線程的業務場景，線程池可以創建固定的線程數量，並且在操作系統底層，輕量級進程將會把這些線程映射到內核。

線程池可以提高線程復用，又可以固定最大線程使用量，防止無限制地創建線程。當程序提交一個任務需要一個線程時，會去線程池中查找是否有空閑的線程，若有，則直接使用線程池中的線程工作，若沒有，會去判斷當前已創建的線程數量是否超過最大線程數量，如未超過，則創建新線程，如已超過，則進行排隊等待或者直接拋出異常。

 

線程池是一種生產者 – 消費者模式

線程池的設計，普遍採用的都是生產者 – 消費者模式。線程池的使用方是生產者，線程池本身是消費者。

原理實現大致如下：

 1 package com.lyyzoo.test.concurrent.executor;  2 
 3 import java.util.ArrayList;  4 import java.util.List;  5 import java.util.concurrent.BlockingQueue;  6 import java.util.concurrent.LinkedBlockingQueue;  7 
 8 /**
 9  * @author bojiangzhou 2020/02/12 10  */
11 public class CustomThreadPool { 12 
13     public static void main(String[] args) { 14         // 使用有界阻塞隊列 創建線程池
15         CustomThreadPool pool = new CustomThreadPool(2, new LinkedBlockingQueue<>(10)); 16         pool.execute(() -> { 17             System.out.println("提交了一個任務"); 18  }); 19  } 20 
21     // 利用阻塞隊列實現生產者-消費者模式
22     final BlockingQueue<Runnable> workQueue; 23     // 保存內部工作線程
24     final List<Thread> threads = new ArrayList<>(); 25 
26     public CustomThreadPool(int coreSize, BlockingQueue<Runnable> workQueue) { 27         this.workQueue = workQueue; 28         // 創建工作線程
29         for (int i = 0; i < coreSize; i++) { 30             WorkerThread work = new WorkerThread(); 31  work.start(); 32  threads.add(work); 33  } 34  } 35 
36     // 生產者 提交任務
37     public void execute(Runnable command) { 38         try { 39             // 隊列已滿，put 會一直等待
40  workQueue.put(command); 41         } catch (InterruptedException e) { 42  e.printStackTrace(); 43  } 44  } 45 
46     /**
47  * 工作線程負責消費任務，並執行任務 48      */
49     class WorkerThread extends Thread { 50  @Override 51         public void run() { 52             // 循環取任務並執行，take 取不到任務會一直等待
53             while (true) { 54                 try { 55                     Runnable runnable = workQueue.take(); 56  runnable.run(); 57                 } catch (InterruptedException e) { 58  e.printStackTrace(); 59  } 60  } 61  } 62  } 63 }

ThreadPoolExecutor

線程池參數說明

Java 提供的線程池相關的工具類中，最核心的是 ThreadPoolExecutor，通過名字也能看出來，它強調的是 Executor，而不是一般意義上的池化資源。

ThreadPoolExecutor 的構造函數非常複雜，這個最完備的構造函數有 7 個參數：

 

各個參數的含義如下：

• corePoolSize：表示線程池保有的最小線程數。
• maximumPoolSize：表示線程池創建的最大線程數。
• keepAliveTime & unit：如果一個線程空閑了 keepAliveTime & unit 這麼久，而且線程池的線程數大於 corePoolSize ，那麼這個空閑的線程就要被回收了。
• workQueue：工作隊列，一般定義有界阻塞隊列。
• threadFactory：通過這個參數你可以自定義如何創建線程，例如你可以給線程指定一個有意義的名字。
• handler：通過這個參數可以自定義任務的拒絕策略。如果線程池中所有的線程都在忙碌，並且工作隊列也滿了（前提是工作隊列是有界隊列），那麼此時提交任務，線程池就會拒絕接收。ThreadPoolExecutor 已經提供了以下 4 種拒絕策略。
  • 　　CallerRunsPolicy：提交任務的線程自己去執行該任務。
  • 　　AbortPolicy：默認的拒絕策略，會 throws RejectedExecutionException。
  • 　　DiscardPolicy：直接丟棄任務，沒有任何異常拋出。
  • 　　DiscardOldestPolicy：丟棄最老的任務，其實就是把最早進入工作隊列的任務丟棄，然後把新任務加入到工作隊列。

 

ThreadPoolExecutor 構造完成后，還可以通過如下方法定製默認行為：

• executor.allowCoreThreadTimeOut(true)：將包括“核心線程”在內的，沒有任務分配的所有線程，在等待 keepAliveTime 時間后回收掉。
• executor.prestartAllCoreThreads()：創建線程池后，立即創建核心數個工作線程；線程池默認是在任務來時才創建工作線程。

 

創建線程池示例：

 1 public void test() throws InterruptedException {  2     ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(  3             // 核心線程數
 4             2,  5             // 最大線程數
 6             16,  7             // 線程空閑時間
 8             60, TimeUnit.SECONDS,  9             // 使用有界阻塞隊列
10             new LinkedBlockingQueue<>(1024), 11             // 定義線程創建方式，可自定線程名稱
12             new ThreadFactoryBuilder().setNameFormat("executor-%d").build(), 13             // 自定義拒絕策略，一般和降級策略配合使用
14             (r, executor) -> { 15                 // 隊列已滿，拒絕執行
16                 throw new RejectedExecutionException("Task " + r.toString() +
17                         " rejected from " + executor.toString()); 18  } 19  ); 20 
21     poolExecutor.submit(() -> { 22         LOGGER.info("submit task"); 23  }); 24 }

 

線程池的線程分配流程

任務提交后的大致流程如下圖所示。提交任務后，如果線程數小於 corePoolSize，則創建新線程執行任務，無論當前線程池的線程是否空閑都會創建新的線程。

當創建的線程數等於 corePoolSize 時，提交的任務會被加入到設置的阻塞隊列中。

當隊列滿了，則會創建非核心線程執行任務，直到線程池中的線程數量等於 maximumPoolSize。

當線程數量已經等於 maximumPoolSize 時， 新提交的任務無法加入到等待隊列，也無法創建非核心線程直接執行，如果沒有為線程池設置拒絕策略，這時線程池就會拋出 RejectedExecutionException 異常，即默認拒絕接受任務。

 

線程池默認的拒絕策略就是丟棄任務，所以我們在設置有界隊列時，需要考慮設置合理的拒絕策略，要考慮到高峰時期任務的數量，避免任務被丟棄而影響業務流程。

 

強烈建議使用有界隊列

創建 ThreadPoolExecutor 時強烈建議使用有界隊列。如果設置為無界隊列，那麼一般最大線程數的設置是不起作用的，而且遇到任務高峰時，如果一直往隊列添加任務，容易出現OOM，拋出如下異常。

Exception in thread "http-nio-45678-ClientPoller" java.lang.OutOfMemoryError: GC overhead limit exceeded

 

使用有界隊列時，需要注意，當任務過多時，線程池會觸發執行拒絕策略，線程池默認的拒絕策略會拋出 RejectedExecutionException，這是個運行時異常，對於運行時異常編譯器並不強制 catch 它，所以開發人員很容易忽略，因此默認拒絕策略要慎重使用。如果線程池處理的任務非常重要，建議自定義自己的拒絕策略；並且在實際工作中，自定義的拒絕策略往往和降級策略配合使用。

 

監控線程池的狀態

建議用一些監控手段來觀察線程池的狀態。線程池這個組件往往會表現得任勞任怨、默默無聞，除非是出現了拒絕策略，否則壓力再大都不會拋出一個異常。如果我們能提前觀察到線程池隊列的積壓，或者線程數量的快速膨脹，往往可以提早發現並解決問題。

 1 public static void displayThreadPoolStatus(ThreadPoolExecutor threadPool, String threadPoolName, long period, TimeUnit unit) {
 2     Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
 3         LOGGER.info("[>>ExecutorStatus<<] ThreadPool Name: [{}], Pool Status: [shutdown={}, Terminated={}], Pool Thread Size: {}, Active Thread Count: {}, Task Count: {}, Tasks Completed: {}, Tasks in Queue: {}",
 4                 threadPoolName,
 5                 threadPool.isShutdown(), threadPool.isTerminated(), // 線程是否被終止
 6                 threadPool.getPoolSize(), // 線程池線程數量
 7                 threadPool.getActiveCount(), // 工作線程數
 8                 threadPool.getTaskCount(), // 總任務數
 9                 threadPool.getCompletedTaskCount(), // 已完成的任務數
10                 threadPool.getQueue().size()); // 線程池中線程的數量
11     }, 0, period, unit);
12 }

線程池任務提交方式

提交任務可以通過 execute 和 submit 方法提交任務，下面就來看下它們的區別。

submit 方法簽名：

execute 方法簽名：

 

使用 execute 提交任務

使用 execute 提交任務，線程池內拋出異常會導致線程退出，線程池只能重新創建一個線程。如果每個異步任務都以異常結束，那麼線程池可能完全起不到線程重用的作用。

而且主線程無法捕獲（catch）到線程池內拋出的異常。因為沒有手動捕獲異常進行處理，ThreadGroup 幫我們進行了未捕獲異常的默認處理，向標準錯誤輸出打印了出現異常的線程名稱和異常信息。顯然，這種沒有以統一的錯誤日誌格式記錄錯誤信息打印出來的形式，對生產級代碼是不合適的。

 

如下，execute 提交任務，拋出異常后，從線程名稱可以看出，老線程退出，創建了新的線程。

ThreadGroup 處理未捕獲異常：直接輸出到 System.err

 

解決方式：

• 以 execute 方法提交到線程池的異步任務，最好在任務內部做好異常處理；
• 設置自定義的異常處理程序作為保底，比如在聲明線程池時自定義線程池的未捕獲異常處理程序。或者設置全局的默認未捕獲異常處理程序。

 1 // 自定義線程池的未捕獲異常處理程序
 2 ThreadPoolExecutor executor = new ThreadPoolExecutor(8, 8,  3         30, TimeUnit.MINUTES,  4         new LinkedBlockingQueue<>(),  5         new ThreadFactoryBuilder()  6                 .setNameFormat("pool-%d")  7                 .setUncaughtExceptionHandler((Thread t, Throwable e) -> {  8                     log.error("pool happen exception, thread is {}", t, e);  9  }) 10  .build()); 11                 
12 // 設置全局的默認未捕獲異常處理程序
13 static { 14     Thread.setDefaultUncaughtExceptionHandler((thread, throwable)-> { 15         log.error("Thread {} got exception", thread, throwable) 16  }); 17 }  

定義的異常處理程序將未捕獲的異常信息打印到標準日誌中了，老線程同樣會退出。如果要避免這個問題，就需要使用 submit 方法提交任務。

 

使用 submit 提交任務

使用 submit，線程不會退出，但是異常不會記錄，會被生吞掉。查看 FutureTask 源碼可以發現，在執行任務出現異常之後，異常存到了一個 outcome 字段中，只有在調用 get 方法獲取 FutureTask 結果的時候，才會以 ExecutionException 的形式重新拋出異常。所以我們可以通過捕獲 get 方法拋出的異常來判斷線程的任務是否拋出了異常。

 

submit 提交任務，可以通過 Future 獲取返回結果，如果拋出異常，可以捕獲 ExecutionException 得到異常棧信息。通過線程名稱可以看出，老線程也沒有退出。

需要注意的是，使用 submit 時，setUncaughtExceptionHandler 設置的異常處理器不會生效。

 

submit 與 execute 的區別

execute提交的是Runnable類型的任務，而submit提交的是Callable或者Runnable類型的任務；

execute的提交沒有返回值，而submit的提交會返回一個Future類型的對象；

execute提交的時候，如果有異常，就會直接拋出異常，而submit在遇到異常的時候，通常不會立馬拋出異常，而是會將異常暫時存儲起來，等待你調用Future.get()方法的時候，才會拋出異常；

execute 提交的任務拋出異常，老線程會退出，線程池會立即創建一個新的線程。submit 提交的任務拋出異常，老線程不會退出；

線程池設置的 UncaughtExceptionHandler 對 execute 提交的任務生效，對 submit 提交的任務不生效。

線程數設置多少合適

創建多少線程合適，要看多線程具體的應用場景。我們的程序一般都是 CPU 計算和 I/O 操作交叉執行的，由於 I/O 設備的速度相對於 CPU 來說都很慢，所以大部分情況下，I/O 操作執行的時間相對於 CPU 計算來說都非常長，這種場景我們一般都稱為 I/O 密集型計算；和 I/O 密集型計算相對的就是 CPU 密集型計算了，CPU 密集型計算大部分場景下都是純 CPU 計算。I/O 密集型程序和 CPU 密集型程序，計算最佳線程數的方法是不同的。

 

CPU 密集型計算

多線程本質上是提升多核 CPU 的利用率，所以對於一個 4 核的 CPU，每個核一個線程，理論上創建 4 個線程就可以了，再多創建線程也只是增加線程切換的成本。所以，對於 CPU 密集型的計算場景，理論上“線程的數量 = CPU 核數”就是最合適的。不過在工程上，線程的數量一般會設置為“CPU 核數 +1”，這樣的話，當線程因為偶爾的內存頁失效或其他原因導致阻塞時，這個額外的線程可以頂上，從而保證 CPU 的利用率。

 

I/O 密集型的計算場景

如果 CPU 計算和 I/O 操作的耗時是 1:1，那麼 2 個線程是最合適的。如果 CPU 計算和 I/O 操作的耗時是 1:2，那設置 3 個線程是合適的，如下圖所示：CPU 在 A、B、C 三個線程之間切換，對於線程 A，當 CPU 從 B、C 切換回來時，線程 A 正好執行完 I/O 操作。這樣 CPU 和 I/O 設備的利用率都達到了 100%。

會發現，對於 I/O 密集型計算場景，最佳的線程數是與程序中 CPU 計算和 I/O 操作的耗時比相關的，可以總結出這樣一個公式：最佳線程數 =1 +（I/O 耗時 / CPU 耗時）

對於多核 CPU，需要等比擴大，計算公式如下：最佳線程數 =CPU 核數 * [ 1 +（I/O 耗時 / CPU 耗時）]

 

線程池線程數設置 

可通過如下方式獲取CPU核數：

1 /**
2  * 獲取返回CPU核數 3  * 4  * @return 返回CPU核數，默認為8 5  */
6 public static int getCpuProcessors() { 7     return Runtime.getRuntime() != null && Runtime.getRuntime().availableProcessors() > 0 ?
8             Runtime.getRuntime().availableProcessors() : 8; 9 }

 

在一些非核心業務中，我們可以將核心線程數設置小一些，最大線程數量設置為CPU核心數量，阻塞隊列大小根據具體場景設置；不要過大，防止大量任務進入等待隊列而超時，應儘快創建非核心線程執行任務；也不要過小，避免隊列滿了任務被拒絕丟棄。

 1 public ThreadPoolExecutor executor() {  2     int coreSize = getCpuProcessors();  3     ThreadPoolExecutor executor = new ThreadPoolExecutor(  4             2, coreSize,  5             10, TimeUnit.MINUTES,  6             new LinkedBlockingQueue<>(512),  7             new ThreadFactoryBuilder().setNameFormat("executor-%d").build(), 10             new ThreadPoolExecutor.AbortPolicy() 11  );14 
15     return executor; 16 }

 

在一些核心業務中，核心線程數設置為CPU核心數，最大線程數可根據公式 最佳線程數 =CPU 核數 * [ 1 +（I/O 耗時 / CPU 耗時）] 來計算。阻塞隊列可以根據具體業務場景設置，如果線程處理業務非常迅速，我們可以考慮將阻塞隊列設置大一些，處理的請求吞吐量會大些；如果線程處理業務非常耗時，阻塞隊列設置小些，防止請求在阻塞隊列中等待過長時間而導致請求已超時。

public ThreadPoolExecutor executor() { int coreSize = getCpuProcessors(); ThreadPoolExecutor executor = new ThreadPoolExecutor( coreSize, coreSize * 8, 30, TimeUnit.MINUTES, new LinkedBlockingQueue<>(1024), new ThreadFactoryBuilder().setNameFormat("executor-%d").build(), new ThreadPoolExecutor.AbortPolicy() );return executor; }

 

注意：一般不要將 corePoolSize 設置為 0，例如下面的線程池，使用了無界隊列，雖 maximumPoolSize > 0，但實際上只會有一個工作線程，因為其它任務都加入等待隊列了。

1 ThreadPoolExecutor executor = new ThreadPoolExecutor(0, 5, 30, TimeUnit.SECONDS, 3         new LinkedBlockingQueue<>(), 4         new ThreadFactoryBuilder().setNameFormat("test-%d").build() 5 );

 

線程池如何優先啟用非核心線程

如果想讓線程池激進一點，優先開啟更多的線程，而把隊列當成一個後備方案，可以自定義隊列，重寫 offer 方法，因為線程池是通過 offer 方法將任務放入隊列。

 

通過重寫隊列的 offer 方法，直接返回 false，造成這個隊列已滿的假象，線程池在工作隊列滿了無法入隊的情況下會擴容線程池。直到線程數達到最大線程數，就會觸發拒絕策略，此時再通過自定義的拒絕策略將任務通過隊列的 put 方法放入隊列中。這樣就可以優先開啟更多線程，而不是進入隊列了。

 1 public static void main(String[] args) {  2     // ThreadPoolExecutor 通過 offer 將元素放入隊列，重載隊列的 offer 方法，直接返回 false，造成隊列已滿的假象  3     // 隊列滿時，會創建新的線程直到達到 maximumPoolSize，之後會觸發執行拒絕策略
 4     LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {  5         private static final long serialVersionUID = 8303142475890427046L;  6 
 7  @Override  8         public boolean offer(Runnable e) {  9             return false; 10  } 11  }; 12 
13     // 當線程達到 maximumPoolSize 時會觸發拒絕策略，此時將任務 put 到隊列中
14     RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() { 15  @Override 16         public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { 17             try { 18                 // 任務拒絕時，通過 put 放入隊列
19  queue.put(r); 20             } catch (InterruptedException e) { 21  Thread.currentThread().interrupt(); 22  } 23  } 24  }; 25 
26     // 構造線程池
27     ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 28             600, TimeUnit.SECONDS, 29  queue, 30             new ThreadFactoryBuilder().setNameFormat("demo-%d").build(), 31  rejectedExecutionHandler); 32 
33     IntStream.rangeClosed(1, 50).forEach(i -> { 34         executor.submit(() -> { 35             log.info("start..."); 36             sleep(9000); 37  }); 38  }); 39 }

優雅的終止線程和線程池

優雅地終止線程

在程序中，我們不能隨便中斷一個線程，因為這是極其不安全的操作，我們無法知道這個線程正運行在什麼狀態，它可能持有某把鎖，強行中斷可能導致鎖不能釋放的問題；或者線程可能在操作數據庫，強行中斷導致數據不一致混亂的問題。正因此，JAVA里將Thread的stop方法設置為過時，以禁止大家使用。

優雅地終止線程，不是自己終止自己，而是在一個線程 T1 中，終止線程 T2；這裏所謂的“優雅”，指的是給 T2 一個機會料理後事，而不是被一劍封喉。兩階段終止模式，就是將終止過程分成兩個階段，其中第一個階段主要是線程 T1 向線程 T2發送終止指令，而第二階段則是線程 T2響應終止指令。

Java 線程進入終止狀態的前提是線程進入 RUNNABLE 狀態，而實際上線程也可能處在休眠狀態，也就是說，我們要想終止一個線程，首先要把線程的狀態從休眠狀態轉換到 RUNNABLE 狀態。如何做到呢？這個要靠 Java Thread 類提供的 interrupt() 方法，它可以將休眠狀態的線程轉換到 RUNNABLE 狀態。

線程轉換到 RUNNABLE 狀態之後，我們如何再將其終止呢？RUNNABLE 狀態轉換到終止狀態，優雅的方式是讓 Java 線程自己執行完 run() 方法，所以一般我們採用的方法是設置一個標誌位，然後線程會在合適的時機檢查這個標誌位，如果發現符合終止條件，則自動退出 run() 方法。這個過程其實就是第二階段：響應終止指令。終止指令，其實包括兩方面內容：interrupt() 方法和線程終止的標誌位。

如果我們在線程內捕獲中斷異常(如Thread.sleep()拋出了中斷一次)之後，需通過 Thread.currentThread().interrupt() 重新設置線程的中斷狀態，因為 JVM 的異常處理會清除線程的中斷狀態。

 

建議自己設置線程終止標誌位，避免線程內調用第三方類庫的方法未處理線程中斷狀態，如下所示。

 1 public class InterruptDemo {  2 
 3     /**
 4  * 輸出：調用 interrupt() 時，只是設置了線程中斷標識，線程依舊會繼續執行當前方法，執行完之後再退出線程。  5  * do something...  6  * continue do something...  7  * do something...  8  * continue do something...  9  * do something... 10  * 線程被中斷... 11  * continue do something... 12      */
13     public static void main(String[] args) throws InterruptedException { 14         Proxy proxy = new Proxy(); 15  proxy.start(); 16 
17         Thread.sleep(6000); 18  proxy.stop(); 19  } 20 
21     static class Proxy { 22         // 自定義線程終止標誌位
23         private volatile boolean terminated = false; 24 
25         private boolean started = false; 26 
27  Thread t; 28 
29         public synchronized void start() { 30             if (started) { 31                 return; 32  } 33             started = true; 34             terminated = false; 35 
36             t = new Thread(() -> { 37                 while (!terminated) { // 取代 while (true)
38                     System.out.println("do something..."); 39                     try { 40                         Thread.sleep(2000); 41                     } catch (InterruptedException e) { 42                         // 如果其它線程中斷此線程，拋出異常時，需重新設置線程中斷狀態，因為 JVM 的異常處理會清除線程的中斷狀態。
43                         System.out.println("線程被中斷..."); 44  Thread.currentThread().interrupt(); 45  } 46                     System.out.println("continue do something..."); 47  } 48                 started = false; 49  }); 50  t.start(); 51  } 52 
53         public synchronized void stop() { 54             // 設置中斷標誌
55             terminated = true; 56  t.interrupt(); 57  } 58  } 59 
60 }

 

優雅的終止線程池

線程池提供了兩個方法來中斷線程池：shutdown() 和 shutdownNow()。

shutdown()：是一種很保守的關閉線程池的方法。線程池執行 shutdown() 后，就會拒絕接收新的任務，但是會等待線程池中正在執行的任務和已經進入阻塞隊列的任務都執行完之後才最終關閉線程池。

shutdownNow()：相對激進一些，線程池執行 shutdownNow() 后，會拒絕接收新的任務，同時還會中斷線程池中正在執行的任務，已經進入阻塞隊列的任務也被剝奪了執行的機會，不過這些被剝奪執行機會的任務會作為 shutdownNow() 方法的返回值返回。因為 shutdownNow() 方法會中斷正在執行的線程，所以提交到線程池的任務，如果需要優雅地結束，就需要正確地處理線程中斷。如果提交到線程池的任務不允許取消，那就不能使用 shutdownNow() 方法終止線程池。

 

如果想在jvm關閉的時候進行內存清理、對象銷毀等操作，或者僅僅想起個線程然後這個線程不會退出，可以使用Runtime.addShutdownHook。

這個方法的作用就是在JVM中增加一個關閉的鈎子。當程序正常退出、系統調用 System.exit 方法或者虛擬機被關閉時才會執行系統中已經設置的所有鈎子，當系統執行完這些鈎子后，JVM才會關閉。

利用這個性質，就可以在這個最後執行的線程中把線程池優雅的關閉掉。雖然jvm關閉了，但優雅關閉線程池總是好的，特別是涉及到服務端的 tcp 連接。

 1 /**
 2  * 添加Hook在Jvm關閉時優雅的關閉線程池  3  *  4  * @param threadPool 線程池  5  * @param threadPoolName 線程池名稱  6  */
 7 public static void hookShutdownThreadPool(ExecutorService threadPool, String threadPoolName) {  8     Runtime.getRuntime().addShutdownHook(new Thread(() -> {  9         LOGGER.info("[>>ExecutorShutdown<<] Start to shutdown the thead pool: [{}]", threadPoolName); 10         // 使新任務無法提交
11  threadPool.shutdown(); 12         try { 13             // 等待未完成任務結束
14             if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) { 15                 threadPool.shutdownNow(); // 取消當前執行的任務
16                 LOGGER.warn("[>>ExecutorShutdown<<] Interrupt the worker, which may cause some task inconsistent. Please check the biz logs."); 17 
18                 // 等待任務取消的響應
19                 if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) { 20                     LOGGER.error("[>>ExecutorShutdown<<] Thread pool can't be shutdown even with interrupting worker threads, which may cause some task inconsistent. Please check the biz logs."); 21  } 22  } 23         } catch (InterruptedException ie) { 24             // 重新取消當前線程進行中斷
25  threadPool.shutdownNow(); 26             LOGGER.error("[>>ExecutorShutdown<<] The current server thread is interrupted when it is trying to stop the worker threads. This may leave an inconsistent state. Please check the biz logs."); 27 
28             // 保留中斷狀態
29  Thread.currentThread().interrupt(); 30  } 31 
32         LOGGER.info("[>>ExecutorShutdown<<] Finally shutdown the thead pool: [{}]", threadPoolName); 33  })); 34 }

Executors

考慮到 ThreadPoolExecutor 的構造函數實在是有些複雜，所以 Java 併發包里提供了一個線程池的靜態工廠類 Executors，利用 Executors 你可以快速創建線程池。

但《阿里巴巴 Java 開發手冊》中提到，禁止使用這些方法來創建線程池，而應該手動 new ThreadPoolExecutor 來創建線程池。最重要的原因是：Executors 提供的很多方法默認使用的都是無界的 LinkedBlockingQueue，高負載情境下，無界隊列很容易導致 OOM，而 OOM 會導致所有請求都無法處理，這是致命問題。最典型的就是 newFixedThreadPool 和 newCachedThreadPool，可能因為資源耗盡導致 OOM 問題。

 

newCachedThreadPool

具有緩存性質的線程池，線程最大空閑時間60s，線程可重複利用，沒有最大線程數限制。使用的是 SynchronousQueue 無容量阻塞隊列，沒有最大線程數限制。這意味着，只要有請求到來，就必須找到一條工作線程來處理，如果當前沒有空閑的線程就再創建一條新的。

高併發情況下，大量的任務進來後會創建大量的線程，導致OOM（無法創建本地線程）：

1 [11:30:30.487] [http-nio-45678-exec-1] [ERROR] [.a.c.c.C.[.[.[/].[dispatcherServlet]:175 ] - Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Handler dispatch failed; 2     nested exception is java.lang.OutOfMemoryError: unable to create new native thread] with root cause 3 java.lang.OutOfMemoryError: unable to create new native thread 

 

newFixedThreadPool

具有固定數量的線程池，核心線程數等於最大線程數，超出最大線程數進行等待。使用的是 LinkedBlockingQueue 無界阻塞隊列。雖然使用 newFixedThreadPool 可以把工作線程控制在固定的數量上，但任務隊列是無界的。如果任務較多並且執行較慢的話，隊列可能會快速積壓，撐爆內存導致 OOM。

如果一直往這個無界隊列中添加任務，不久就會出現OOM異常（內存佔滿）：

1 Exception in thread "http-nio-45678-ClientPoller" 
2     java.lang.OutOfMemoryError: GC overhead limit exceeded

 

newSingleThreadExecutor

核心線程數與最大線程數均為1，可用於當鎖控制同步。使用的是 LinkedBlockingQueue 無界阻塞隊列。

 

newScheduledThreadPool

具有時間調度性的線程池，必須初始化核心線程數。

沒有最大線程數限制，線程最大空閑時間為0，空閑線程執行完即銷毀。底層使用 DelayedWorkQueue 實現延遲特性。

線程池創建正確姿勢

最後，總結一下，從如下的一些方面考慮如何正確地創建線程池。

線程池配置

我們需要根據自己的場景、併發情況來評估線程池的幾個核心參數，包括核心線程數、最大線程數、線程回收策略、工作隊列的類型，以及拒絕策略，確保線程池的工作行為符合需求，一般都需要設置有界的工作隊列和可控的線程數。

要根據任務的“輕重緩急”來指定線程池的核心參數，包括線程數、回收策略和任務隊列：

• 對於執行比較慢、數量不大的 IO 任務，要考慮更多的線程數，而不需要太大的隊列。
• 對於吞吐量較大的計算型任務，線程數量不宜過多，可以是 CPU 核數或核數 *2（理由是，線程一定調度到某個 CPU 進行執行，如果任務本身是 CPU 綁定的任務，那麼過多的線程只會增加線程切換的開銷，並不能提升吞吐量），但可能需要較長的隊列來做緩衝。

 

任何時候，都應該為自定義線程池指定有意義的名稱，以方便排查問題。當出現線程數量暴增、線程死鎖、線程佔用大量 CPU、線程執行出現異常等問題時，我們往往會抓取線程棧。此時，有意義的線程名稱，就可以方便我們定位問題。

除了建議手動聲明線程池以外，還建議用一些監控手段來觀察線程池的狀態。如果我們能提前觀察到線程池隊列的積壓，或者線程數量的快速膨脹，往往可以提早發現並解決問題。

 

確認線程池本身是不是復用的

既然使用了線程池就需要確保線程池是在復用的，每次 new 一個線程池出來可能比不用線程池還糟糕。如果你沒有直接聲明線程池而是使用其他同學提供的類庫來獲得一個線程池，請務必查看源碼，以確認線程池的實例化方式和配置是符合預期的。

 

斟酌線程池的混用策略

不要盲目復用線程池，別人定義的線程池屬性不一定適合你的任務，而且混用會相互干擾。

另外，Java 8 的 parallel stream 背後是共享同一個 ForkJoinPool，默認并行度是 CPU 核數 -1。對於 CPU 綁定的任務來說，使用這樣的配置比較合適，但如果集合操作涉及同步 IO 操作的話（比如數據庫操作、外部服務調用等），建議自定義一個 ForkJoinPool（或普通線程池）。因此在使用 Java8 的并行流時，建議只用在計算密集型的任務，IO密集型的任務建議自定義線程池來提交任務，避免影響其它業務。

 

CommonExecutor

如下是我自己封裝的一個線程池工具類，還提供了執行批量任務的方法，關於批量任務後面再單獨寫篇文章來介紹。

 1 package org.hzero.core.util;  2 
 3 import java.util.ArrayList;  4 import java.util.Collections;  5 import java.util.List;  6 import java.util.concurrent.*;  7 import java.util.stream.Collectors;  8 import javax.annotation.Nonnull;  9 
 10 import com.google.common.util.concurrent.ThreadFactoryBuilder;  11 import org.apache.commons.collections4.CollectionUtils;  12 import org.apache.commons.lang3.RandomUtils;  13 import org.slf4j.Logger;  14 import org.slf4j.LoggerFactory;  15 import org.springframework.dao.DuplicateKeyException;  16 
 17 import io.choerodon.core.exception.CommonException;  18 
 19 import org.hzero.core.base.BaseConstants;  20 
 21 /**
 22  * @author bojiangzhou 2020/02/24  23  */
 24 public class CommonExecutor {  25 
 26     private static final Logger LOGGER = LoggerFactory.getLogger(CommonExecutor.class);  27 
 28     private static final ThreadPoolExecutor BASE_EXECUTOR;  29 
 30     static {  31         BASE_EXECUTOR = buildThreadFirstExecutor("BaseExecutor");  32  }  33 
 34     /**
 35  * 構建線程優先的線程池  36  * <p>  37  * 線程池默認是當核心線程數滿了后，將任務添加到工作隊列中，當工作隊列滿了之後，再創建線程直到達到最大線程數。  38  *  39  * <p>  40  * 線程優先的線程池，就是在核心線程滿了之後，繼續創建線程，直到達到最大線程數之後，再把任務添加到工作隊列中。  41  *  42  * <p>  43  * 此方法默認設置核心線程數為 CPU 核數，最大線程數為 8倍 CPU 核數，空閑線程超過 5 分鐘銷毀，工作隊列大小為 65536。  44  *  45  * @param poolName 線程池名稱  46  * @return ThreadPoolExecutor  47      */
 48     public static ThreadPoolExecutor buildThreadFirstExecutor(String poolName) {  49         int coreSize = CommonExecutor.getCpuProcessors();  50         int maxSize = coreSize * 8;  51         return buildThreadFirstExecutor(coreSize, maxSize, 5, TimeUnit.MINUTES, 1 << 16, poolName);  52  }  53 
 54     /**
 55  * 構建線程優先的線程池  56  * <p>  57  * 線程池默認是當核心線程數滿了后，將任務添加到工作隊列中，當工作隊列滿了之後，再創建線程直到達到最大線程數。  58  *  59  * <p>  60  * 線程優先的線程池，就是在核心線程滿了之後，繼續創建線程，直到達到最大線程數之後，再把任務添加到工作隊列中。  61  *  62  * @param corePoolSize 核心線程數  63  * @param maximumPoolSize 最大線程數  64  * @param keepAliveTime 空閑線程的空閑時間  65  * @param unit 時間單位  66  * @param workQueueSize 工作隊列容量大小  67  * @param poolName 線程池名稱  68  * @return ThreadPoolExecutor  69      */
 70     public static ThreadPoolExecutor buildThreadFirstExecutor(int corePoolSize,  71                                                               int maximumPoolSize,  72                                                               long keepAliveTime,  73  TimeUnit unit,  74                                                               int workQueueSize,  75  String poolName) {  76         // 自定義隊列，優先開啟更多線程，而不是放入隊列
 77         LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(workQueueSize) {  78             private static final long serialVersionUID = 5075561696269543041L;  79 
 80  @Override  81             public boolean offer(@Nonnull Runnable o) {  82                 return false; // 造成隊列已滿的假象
 83  }  84  };  85 
 86         // 當線程達到 maximumPoolSize 時會觸發拒絕策略，此時將任務 put 到隊列中
 87         RejectedExecutionHandler rejectedExecutionHandler = (runnable, executor) -> {  88             try {  89                 // 任務拒絕時，通過 offer 放入隊列
 90  queue.put(runnable);  91             } catch (InterruptedException e) {  92                 LOGGER.warn("{} Queue offer interrupted. ", poolName, e);  93  Thread.currentThread().interrupt();  94  }  95  };  96 
 97         ThreadPoolExecutor executor = new ThreadPoolExecutor(  98  corePoolSize, maximumPoolSize,  99  keepAliveTime, unit, 100  queue, 101                 new ThreadFactoryBuilder() 102                         .setNameFormat(poolName + "-%d") 103                         .setUncaughtExceptionHandler((Thread thread, Throwable throwable) -> { 104                             LOGGER.error("{} catching the uncaught exception, ThreadName: [{}]", poolName, thread.toString(), throwable); 105  }) 106  .build(), 107  rejectedExecutionHandler 108  ); 109 
110  CommonExecutor.displayThreadPoolStatus(executor, poolName); 111  CommonExecutor.hookShutdownThreadPool(executor, poolName); 112         return executor; 113  } 114 
115     /**
116  * 批量提交異步任務，使用默認的線程池
117  * 118  * @param tasks 將任務轉化為 AsyncTask 批量提交 119      */
120     public static <T> List<T> batchExecuteAsync(List<AsyncTask<T>> tasks, @Nonnull String taskName) { 121         return batchExecuteAsync(tasks, BASE_EXECUTOR, taskName); 122  } 123 
124     /**
125  * 批量提交異步任務，執行失敗可拋出異常或返回異常編碼即可 <br> 126  * <p> 127  * 需注意提交的異步任務無法控制事務，一般需容忍產生一些垃圾數據的情況下才能使用異步任務，異步任務執行失敗將拋出異常，主線程可回滾事務. 128  * <p> 129  * 異步任務失敗后，將取消剩餘的任務執行. 130  * 131  * @param tasks 將任務轉化為 AsyncTask 批量提交 132  * @param executor 線程池，需自行根據業務場景創建相應的線程池 133  * @return 返回執行結果 134      */
135     public static <T> List<T> batchExecuteAsync(@Nonnull List<AsyncTask<T>> tasks, @Nonnull ThreadPoolExecutor executor, @Nonnull String taskName) { 136         if (CollectionUtils.isEmpty(tasks)) { 137             return Collections.emptyList(); 138  } 139 
140         int size = tasks.size(); 141 
142         List<Callable<T>> callables = tasks.stream().map(t -> (Callable<T>) () -> { 143             try { 144                 T r = t.doExecute(); 145 
146                 LOGGER.debug("[>>Executor<<] Async task execute success. ThreadName: [{}], BatchTaskName: [{}], SubTaskName: [{}]", 147  Thread.currentThread().getName(), taskName, t.taskName()); 148                 return r; 149             } catch (Throwable e) { 150                 LOGGER.warn("[>>Executor<<] Async task execute error. ThreadName: [{}], BatchTaskName: [{}], SubTaskName: [{}], exception: {}", 151  Thread.currentThread().getName(), taskName, t.taskName(), e.getMessage()); 152                 throw e; 153  } 154  }).collect(Collectors.toList()); 155 
156         CompletionService<T> cs = new ExecutorCompletionService<>(executor, new LinkedBlockingQueue<>(size)); 157         List<Future<T>> futures = new ArrayList<>(size); 158         LOGGER.info("[>>Executor<<] Start async tasks, BatchTaskName: [{}], TaskSize: [{}]", taskName, size); 159 
160         for (Callable<T> task : callables) { 161  futures.add(cs.submit(task)); 162  } 163 
164         List<T> resultList = new ArrayList<>(size); 165         for (int i = 0; i < size; i++) { 166             try { 167                 Future<T> future = cs.poll(6, TimeUnit.MINUTES); 168                 if (future != null) { 169                     T result = future.get(); 170  resultList.add(result); 171                     LOGGER.debug("[>>Executor<<] Async task [{}] - [{}] execute success, result: {}", taskName, i, result); 172                 } else { 173  cancelTask(futures); 174                     LOGGER.error("[>>Executor<<] Async task [{}] - [{}] execute timeout, then cancel other tasks.", taskName, i); 175                     throw new CommonException(BaseConstants.ErrorCode.TIMEOUT); 176  } 177             } catch (ExecutionException e) { 178                 LOGGER.warn("[>>Executor<<] Async task [{}] - [{}] execute error, then cancel other tasks.", taskName, i, e); 179  cancelTask(futures); 180                 Throwable throwable = e.getCause(); 181                 if (throwable instanceof CommonException) { 182                     throw (CommonException) throwable; 183                 } else if (throwable instanceof DuplicateKeyException) { 184                     throw (DuplicateKeyException) throwable; 185                 } else { 186                     throw new CommonException("error.executorError", e.getCause().getMessage()); 187  } 188             } catch (InterruptedException e) { 189  cancelTask(futures); 190                 Thread.currentThread().interrupt(); // 重置中斷標識
191                 LOGGER.error("[>>Executor<<] Async task [{}] - [{}] were interrupted.", taskName, i); 192                 throw new CommonException(BaseConstants.ErrorCode.ERROR); 193  } 194  } 195         LOGGER.info("[>>Executor<<] Finish async tasks , BatchTaskName: [{}], TaskSize: [{}]", taskName, size); 196         return resultList; 197  } 198 
199     /**
200  * 根據一定周期輸出線程池的狀態 201  * 202  * @param threadPool 線程池 203  * @param threadPoolName 線程池名稱 204      */
205     public static void displayThreadPoolStatus(ThreadPoolExecutor threadPool, String threadPoolName) { 206         displayThreadPoolStatus(threadPool, threadPoolName, RandomUtils.nextInt(60, 600), TimeUnit.SECONDS); 207  } 208 
209     /**
210  * 根據一定周期輸出線程池的狀態 211  * 212  * @param threadPool 線程池 213  * @param threadPoolName 線程池名稱 214  * @param period 周期 215  * @param unit 時間單位 216      */
217     public static void displayThreadPoolStatus(ThreadPoolExecutor threadPool, String threadPoolName, long period, TimeUnit unit) { 218         Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> { 219             LOGGER.info("[>>ExecutorStatus<<] ThreadPool Name: [{}], Pool Status: [shutdown={}, Terminated={}], Pool Thread Size: {}, Active Thread Count: {}, Task Count: {}, Tasks Completed: {}, Tasks in Queue: {}", 220  threadPoolName, 221                     threadPool.isShutdown(), threadPool.isTerminated(), // 線程是否被終止
222                     threadPool.getPoolSize(), // 線程池線程數量
223                     threadPool.getActiveCount(), // 工作線程數
224                     threadPool.getTaskCount(), // 總任務數
225                     threadPool.getCompletedTaskCount(), // 已完成的任務數
226                     threadPool.getQueue().size()); // 線程池中線程的數量
227         }, 0, period, unit); 228  } 229 
230     /**
231  * 添加Hook在Jvm關閉時優雅的關閉線程池 232  * 233  * @param threadPool 線程池 234  * @param threadPoolName 線程池名稱 235      */
236     public static void hookShutdownThreadPool(ExecutorService threadPool, String threadPoolName) { 237         Runtime.getRuntime().addShutdownHook(new Thread(() -> { 238             LOGGER.info("[>>ExecutorShutdown<<] Start to shutdown the thead pool: [{}]", threadPoolName); 239             // 使新任務無法提交
240  threadPool.shutdown(); 241             try { 242                 // 等待未完成任務結束
243                 if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) { 244                     threadPool.shutdownNow(); // 取消當前執行的任務
245                     LOGGER.warn("[>>ExecutorShutdown<<] Interrupt the worker, which may cause some task inconsistent. Please check the biz logs."); 246 
247                     // 等待任務取消的響應
248                     if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) { 249                         LOGGER.error("[>>ExecutorShutdown<<] Thread pool can't be shutdown even with interrupting worker threads, which may cause some task inconsistent. Please check the biz logs."); 250  } 251  } 252             } catch (InterruptedException ie) { 253                 // 重新取消當前線程進行中斷
254  threadPool.shutdownNow(); 255                 LOGGER.error("[>>ExecutorShutdown<<] The current server thread is interrupted when it is trying to stop the worker threads. This may leave an inconsistent state. Please check the biz logs."); 256 
257                 // 保留中斷狀態
258  Thread.currentThread().interrupt(); 259  } 260 
261             LOGGER.info("[>>ExecutorShutdown<<] Finally shutdown the thead pool: [{}]", threadPoolName); 262  })); 263  } 264 
265     /**
266  * 獲取返回CPU核數 267  * 268  * @return 返回CPU核數，默認為8 269      */
270     public static int getCpuProcessors() { 271         return Runtime.getRuntime() != null && Runtime.getRuntime().availableProcessors() > 0 ?
272                 Runtime.getRuntime().availableProcessors() : 8; 273  } 274 
275     private static <T> void cancelTask(List<Future<T>> futures) { 276         for (Future<T> future : futures) { 277             if (!future.isDone()) { 278                 future.cancel(true); 279  } 280  } 281  } 282 
283 }

AsyncTask：

 1 package org.hzero.core.util;  2 
 3 import java.util.UUID;  4 
 5 public interface AsyncTask<T> {  6 
 7     default String taskName() {  8         return UUID.randomUUID().toString();  9  } 10 
11  T doExecute(); 12 }

 

————————————————————————————————————–

 

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理
【其他文章推薦】

※USB CONNECTOR掌控什麼技術要點? 帶您認識其相關發展及效能

※台北網頁設計公司這麼多該如何選擇?

※智慧手機時代的來臨，RWD網頁設計為架站首選

※評比南投搬家公司費用收費行情懶人包大公開

※幫你省時又省力,新北清潔一流服務好口碑

※回頭車貨運收費標準

| 好看請贊，養成習慣

• 你有一個思想，我有一個思想，我們交換后，一個人就有兩個思想

• If you can NOT explain it simply, you do NOT understand it well enough

現陸續將Demo代碼和技術文章整理在一起 Github實踐精選 ，方便大家閱讀查看，本文同樣收錄在此，覺得不錯，還請Star

前言

• 文章 Java AQS隊列同步器以及ReentrantLock的應用 介紹了AQS獨佔式獲取同步狀態的實現，並以 ReentrantLock 為例說明其是如何自定義同步器實現互斥鎖的
• 文章 Java AQS共享式獲取同步狀態及Semaphore的應用分析 介紹 AQS 共享式獲取同步狀態的實現，並說明了 Semaphore 是如何自定義同步器實現簡單限流作用的

有了以上兩篇文章的鋪墊，來理解本文要介紹的既有獨佔式，又有共享式獲取同步狀態的 ReadWriteLock，就非常輕鬆了

ReadWriteLock

ReadWriteLock 直譯過來為【讀寫鎖】。現實中，讀多寫少的業務場景是非常普遍的，比如應用緩存

一個線程將數據寫入緩存，其他線程可以直接讀取緩存中的數據，提高數據查詢效率

之前提到的互斥鎖都是排他鎖，也就是說同一時刻只允許一個線程進行訪問，當面對可共享讀的業務場景，互斥鎖顯然是比較低效的一種處理方式。為了提高效率，讀寫鎖模型就誕生了

效率提升是一方面，但併發編程更重要的是在保證準確性的前提下提高效率

一個寫線程改變了緩存中的值，其他讀線程一定是可以 “感知” 到的，否則可能導致查詢到的值不準確

所以關於讀寫鎖模型就了下面這 3 條規定：

1. 允許多個線程同時讀共享變量
2. 只允許一個線程寫共享變量
3. 如果寫線程正在執行寫操作，此時則禁止其他讀線程讀共享變量

ReadWriteLock 是一個接口，其內部只有兩個方法：

public interface ReadWriteLock {
    // 返回用於讀的鎖
    Lock readLock();

    // 返回用於寫的鎖
    Lock writeLock();
}

所以要了解整個讀/寫鎖的整個應用過程，需要從它的實現類 ReentrantReadWriteLock 說起

ReentrantReadWriteLock 類結構

直接對比ReentrantReadWriteLock 與 ReentrantLock的類結構

他們又很相似吧，根據類名稱以及類結構，按照咱們前序文章的分析，你也就能看出 ReentrantReadWriteLock 的基本特性：

其中黃顏色標記的的 鎖降級 是看不出來的， 這裏先有個印象，下面會單獨說明

另外，不知道你是否還記得，Java AQS隊列同步器以及ReentrantLock的應用 說過，Lock 和 AQS 同步器是一種組合形式的存在，既然這裡是讀/寫兩種鎖，他們的組合模式也就分成了兩種：

1. 讀鎖與自定義同步器的聚合
2. 寫鎖與自定義同步器的聚合

    public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }

這裏只是提醒大家，模式沒有變，不要被讀/寫兩種鎖迷惑

基本示例

說了這麼多，如果你忘了前序知識，整體理解感覺應該是有斷檔的，所以先來看個示例（模擬使用緩存）讓大家對 ReentrantReadWriteLock 有個直觀的使用印象

public class ReentrantReadWriteLockCache {

	// 定義一個非線程安全的 HashMap 用於緩存對象
	static Map<String, Object> map = new HashMap<String, Object>();
	// 創建讀寫鎖對象
	static ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
	// 構建讀鎖
	static Lock rl = readWriteLock.readLock();
	// 構建寫鎖
	static Lock wl = readWriteLock.writeLock();

	public static final Object get(String key) {
		rl.lock();
		try{
			return map.get(key);
		}finally {
			rl.unlock();
		}
	}

	public static final Object put(String key, Object value){
		wl.lock();
		try{
			return map.put(key, value);
		}finally {
			wl.unlock();
		}
	}
}

你瞧，使用就是這麼簡單。但是你知道的，AQS 的核心是鎖的實現，即控制同步狀態 state 的值，ReentrantReadWriteLock 也是應用AQS的 state 來控制同步狀態的，那麼問題來了：

一個 int 類型的 state 怎麼既控制讀的同步狀態，又可以控制寫的同步狀態呢？

顯然需要一點設計了

讀寫狀態設計

如果要在一個 int 類型變量上維護多個狀態，那肯定就需要拆分了。我們知道 int 類型數據佔32位，所以我們就有機會按位切割使用state了。我們將其切割成兩部分：

1. 高16位表示讀
2. 低16位表示寫

所以，要想準確的計算讀/寫各自的狀態值，肯定就要應用位運算了，下面代碼是 JDK1.8，ReentrantReadWriteLock 自定義同步器 Sync 的位操作

abstract static class Sync extends AbstractQueuedSynchronizer {
       

        static final int SHARED_SHIFT   = 16;
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;


        static int sharedCount(int c) { 
          return c >>> SHARED_SHIFT; 
        }

        static int exclusiveCount(int c) { 
          return c & EXCLUSIVE_MASK; 
        }
}

乍一看真是有些複雜的可怕，別慌，咱們通過幾道小小數學題就可以搞定整個位運算過程

整個 ReentrantReadWriteLock 中 讀/寫狀態的計算就是反覆應用這幾道數學題，所以，在閱讀下面內容之前，希望你搞懂這簡單的運算

基礎鋪墊足夠了，我們進入源碼分析吧

源碼分析

寫鎖分析

由於寫鎖是排他的，所以肯定是要重寫 AQS 中 tryAcquire 方法

        protected final boolean tryAcquire(int acquires) {        
            Thread current = Thread.currentThread();
          	// 獲取 state 整體的值
            int c = getState();
            // 獲取寫狀態的值
            int w = exclusiveCount(c);
            if (c != 0) {
                // w=0: 根據推理二，整體狀態不等於零，寫狀態等於零，所以，讀狀態大於0，即存在讀鎖
              	// 或者當前線程不是已獲取寫鎖的線程
              	// 二者之一條件成真，則獲取寫狀態失敗
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                // 根據推理一第 1 條，更新寫狀態值
                setState(c + acquires);
                return true;
            }
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
            setExclusiveOwnerThread(current);
            return true;
        }

上述代碼 第 19 行 writerShouldBlock 也並沒有什麼神秘的，只不過是公平/非公平獲取鎖方式的判斷（是否有前驅節點來判斷）

你瞧，寫鎖獲取方式就是這麼簡單

讀鎖分析

由於讀鎖是共享式的，所以肯定是要重寫 AQS 中 tryAcquireShared 方法

        protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();
            int c = getState();
          	// 寫狀態不等於0，並且鎖的持有者不是當前線程，根據約定 3，則獲取讀鎖失敗
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
          	// 獲取讀狀態值
            int r = sharedCount(c);
          	// 這個地方有點不一樣，我們單獨說明
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                if (r == 0) {
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    firstReaderHoldCount++;
                } else {
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;
            }
          	// 如果獲取讀鎖失敗則進入自旋獲取
            return fullTryAcquireShared(current);
        }

readerShouldBlock 和 writerShouldBlock 在公平鎖的實現上都是判斷是否有前驅節點，但是在非公平鎖的實現上，前者是這樣的：

final boolean readerShouldBlock() {
	return apparentlyFirstQueuedIsExclusive();
}

final boolean apparentlyFirstQueuedIsExclusive() {
  Node h, s;
  return (h = head) != null &&
    // 等待隊列頭節點的下一個節點
    (s = h.next)  != null &&
    // 如果是排他式的節點
    !s.isShared()         &&
    s.thread != null;
}

簡單來說，如果請求讀鎖的當前線程發現同步隊列的 head 節點的下一個節點為排他式節點，那麼就說明有一個線程在等待獲取寫鎖（爭搶寫鎖失敗，被放入到同步隊列中），那麼請求讀鎖的線程就要阻塞，畢竟讀多寫少，如果還沒有這點判斷機制，寫鎖可能會發生【飢餓】

上述條件都滿足了，也就會進入 tryAcquireShared 代碼的第 14 行到第 25 行，這段代碼主要是為了記錄線程持有鎖的次數。讀鎖是共享式的，還想記錄每個線程持有讀鎖的次數，就要用到 ThreadLocal 了，因為這不影響同步狀態 state 的值，所以就不分析了, 只把關係放在這吧

到這裏讀鎖的獲取也就結束了，比寫鎖稍稍複雜那麼一丟丟，接下來就說明一下那個可能讓你迷惑的鎖升級/降級問題吧

讀寫鎖的升級與降級

個人理解：讀鎖是可以被多線程共享的，寫鎖是單線程獨佔的，也就是說寫鎖的併發限制比讀鎖高，所以

在真正了解讀寫鎖的升級與降級之前，我們需要完善一下本文開頭 ReentrantReadWriteLock 的例子

	public static final Object get(String key) {
		Object obj = null;
		rl.lock();
		try{
      // 獲取緩存中的值
			obj = map.get(key);
		}finally {
			rl.unlock();
		}
		// 緩存中值不為空，直接返回
		if (obj!= null) {
			return obj;
		}
		
    // 緩存中值為空，則通過寫鎖查詢DB，並將其寫入到緩存中
		wl.lock();
		try{
      // 再次嘗試獲取緩存中的值
			obj = map.get(key);
      // 再次獲取緩存中值還是為空
			if (obj == null) {
        // 查詢DB
				obj = getDataFromDB(key); // 偽代碼：getDataFromDB
        // 將其放入到緩存中
				map.put(key, obj);
			}
		}finally {
			wl.unlock();
		}
		return obj;
	}

有童鞋可能會有疑問

在寫鎖裏面，為什麼代碼第19行還要再次獲取緩存中的值呢？不是多此一舉嗎？

其實這裏再次嘗試獲取緩存中的值是很有必要的，因為可能存在多個線程同時執行 get 方法，並且參數 key 也是相同的，執行到代碼第 16 行 wl.lock() ,比如這樣：

線程 A，B，C 同時執行到臨界區 wl.lock()， 只有線程 A 獲取寫鎖成功，線程B，C只能阻塞，直到線程A 釋放寫鎖。這時，當線程B 或者 C 再次進入臨界區時，線程 A 已經將值更新到緩存中了，所以線程B，C沒必要再查詢一次DB，而是再次嘗試查詢緩存中的值

既然再次獲取緩存很有必要，我能否在讀鎖里直接判斷，如果緩存中沒有值，那就再次獲取寫鎖來查詢DB不就可以了嘛，就像這樣：

	public static final Object getLockUpgrade(String key) {
		Object obj = null;
		rl.lock();
		try{
			obj = map.get(key);
			if (obj == null){
				wl.lock();
				try{
					obj = map.get(key);
					if (obj == null) {
						obj = getDataFromDB(key); // 偽代碼：getDataFromDB
						map.put(key, obj);
					}
				}finally {
					wl.unlock();
				}
			}
		}finally {
			rl.unlock();
		}

		return obj;
	}

這還真是不可以的，因為獲取一個寫入鎖需要先釋放所有的讀取鎖，如果有兩個讀取鎖試圖獲取寫入鎖，且都不釋放讀取鎖時，就會發生死鎖，所以在這裏，鎖的升級是不被允許的

讀寫鎖的升級是不可以的，那麼鎖的降級是可以的嘛？這個是 Oracle 官網關於鎖降級的示例 ，我將代碼粘貼在此處，大家有興趣可以點進去連接看更多內容

 class CachedData {
   Object data;
   volatile boolean cacheValid;
   final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

   void processCachedData() {
     rwl.readLock().lock();
     if (!cacheValid) {
        // 必須在獲取寫鎖之前釋放讀鎖，因為鎖的升級是不被允許的
        rwl.readLock().unlock();
        rwl.writeLock().lock();
        try {
          // 再次檢查，原因可能是其他線程已經更新過緩存
          if (!cacheValid) {
            data = ...
            cacheValid = true;
          }
					//在釋放寫鎖前，降級為讀鎖
          rwl.readLock().lock();
        } finally {
          //釋放寫鎖，此時持有讀鎖
          rwl.writeLock().unlock(); 
        }
     }

     try {
       use(data);
     } finally {
       rwl.readLock().unlock();
     }
   }
 }

代碼中聲明了一個 volatile 類型的 cacheValid 變量，保證其可見性。

1. 首先獲取讀鎖，如果cache不可用，則釋放讀鎖
2. 然後獲取寫鎖
3. 在更改數據之前，再檢查一次cacheValid的值，然後修改數據，將cacheValid置為true
4. 然後在釋放寫鎖前獲取讀鎖 此時
5. cache中數據可用，處理cache中數據，最後釋放讀鎖

這個過程就是一個完整的鎖降級的過程，目的是保證數據可見性，聽起來很有道理的樣子，那麼問題來了：

上述代碼為什麼在釋放寫鎖之前要獲取讀鎖呢？

如果當前的線程A在修改完cache中的數據后，沒有獲取讀鎖而是直接釋放了寫鎖；假設此時另一個線程B 獲取了寫鎖並修改了數據，那麼線程A無法感知到數據已被修改，但線程A還應用了緩存數據，所以就可能出現數據錯誤

如果遵循鎖降級的步驟，線程A 在釋放寫鎖之前獲取讀鎖，那麼線程B在獲取寫鎖時將被阻塞，直到線程A完成數據處理過程，釋放讀鎖，從而保證數據的可見性

那問題又來了：

使用寫鎖一定要降級嗎？

如果你理解了上面的問題，相信這個問題已經有了答案。假如線程A修改完數據之後， 經過耗時操作后想要再使用數據時，希望使用的是自己修改后的數據，而不是其他線程修改后的數據，這樣的話確實是需要鎖降級；如果只是希望最後使用數據的時候，拿到的是最新的數據，而不一定是自己剛修改過的數據，那麼先釋放寫鎖，再獲取讀鎖，然後使用數據也無妨

在這裏我要額外說明一下你可能存在的誤解：

• 如果已經釋放了讀鎖再獲取寫鎖不叫鎖的升級

• 如果已經釋放了寫鎖在獲取讀鎖也不叫鎖的降級

相信你到這裏也理解了鎖的升級與降級過程，以及他們被允許或被禁止的原因了

總結

本文主要說明了 ReentrantReadWriteLock 是如何應用 state 做位拆分實現讀/寫兩種同步狀態的，另外也通過源碼分析了讀/寫鎖獲取同步狀態的過程，最後又了解了讀寫鎖的升級/降級機制，相信到這裏你對讀寫鎖已經有了一定的理解。如果你對文中的哪些地方覺得理解有些困難，強烈建議你回看本文開頭的兩篇文章，那裡鋪墊了非常多的內容。接下來我們就看看在應用AQS的最後一個併發工具類 CountDownLatch 吧

靈魂追問

1. 讀鎖也沒修改數據，還允許共享式獲取，那還有必要設置讀鎖嗎？
2. 在分佈式環境中，你是如何保證緩存數據一致性的呢？
3. 當你打開看ReentrantReadWriteLock源碼時，你會發現，WriteLock 中可以使用 Condition，但是ReadLock 使用Condition卻會拋出UnsupportedOperationException，這是為什麼呢？

// WriteLock
public Condition newCondition() {
	return sync.newCondition();
}

// ReadLock
public Condition newCondition() {
	throw new UnsupportedOperationException();
}

個人博客：https://dayarch.top
加我微信好友, 進群娛樂學習交流，備註「進群」

歡迎持續關注公眾號：「日拱一兵」

• 前沿 Java 技術乾貨分享
• 高效工具匯總 | 回復「工具」
• 面試問題分析與解答
• 技術資料領取 | 回復「資料」

以讀偵探小說思維輕鬆趣味學習 Java 技術棧相關知識，本着將複雜問題簡單化，抽象問題具體化和圖形化原則逐步分解技術問題，技術持續更新，請持續關注……

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※想知道最厲害的網頁設計公司"嚨底家"!

※幫你省時又省力,新北清潔一流服務好口碑

※別再煩惱如何寫文案,掌握八大原則!

1. 簡介

三月份已經介紹過R2DBC,它是一種異步的、非阻塞的關係式數據庫連接規範。儘管一些NoSQL數據庫供應商為其數據庫提供了反應式數據庫客戶端，但對於大多數項目而言，遷移到NoSQL並不是一個理想的選擇。這促使了一個通用的響應式關係數據庫連接規範的誕生。 作為擁有龐大用戶群的關係式數據庫MySQL也有了反應式驅動，不過並不是官方的。但是Spring官方將其納入了依賴池，說明該類庫的質量並不低。所以今天就嘗嘗鮮，試一下使用R2DBC連接MySQL。

2. 環境依賴

基於Spring Boot 2.3.1和Spring Data R2DBC，還有反應式Web框架Webflux，同時也要依賴r2dbc-mysql庫，所有的Maven依賴為：

       <!--r2dbc mysql 庫-->
        <dependency>
            <groupId>dev.miku</groupId>
            <artifactId>r2dbc-mysql</artifactId>
        </dependency>
        <!--Spring r2dbc 抽象層-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
        </dependency>
        <!--自動配置需要引入的一個嵌入式數據庫類型對象-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jdbc</artifactId>
        </dependency>
       <!--反應式web框架-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

MySQL版本為5.7，沒有測試其它版本。

3. R2DBC配置

所有的R2DBC自動配置都在org.springframework.boot.autoconfigure.data.r2dbc包下，如果要配置MySQL必須針對性的配置對應的連接工廠接口ConnectionFactory，當然也可以通過application.yml配置。個人比較喜歡JavaConfig。

@Bean
ConnectionFactory connectionFactory() {
    return MySqlConnectionFactory.from(MySqlConnectionConfiguration.builder()
            .host("127.0.0.1")
            .port(3306)
            .username("root")
            .password("123456")
            .database("database_name")
             // 額外的其它非必選參數省略                          
            .build());
}

詳細配置可參考r2dbc-mysql的官方說明：https://github.com/mirromutth/r2dbc-mysql

當ConnectionFactory配置好后，就會被注入DatabaseClient 對象。該對象是非阻塞的，用於執行數據庫反應性客戶端調用與反應流背壓請求。我們可以通過該接口反應式地操作數據庫。

4. 編寫反應式接口

我們先創建一張表並寫入一些數據：

create table client_user
(
    user_id         varchar(64)                              not null comment '用戶唯一標示' primary key,
    username        varchar(64)                              null comment '名稱',
    phone_number    varchar(64)                              null comment '手機號',
    gender          tinyint(1) default 0                     null comment '0 未知 1 男 2 女  '
)

對應的實體為：

package cn.felord.r2dbc.config;

import lombok.Data;

/**
 * @author felord.cn
 */
@Data
public class ClientUser {

    private String userId;
    private String username;
    private String phoneNumber;
    private Integer gender;
}

然後我們編寫一個Webflux的反應式接口：

package cn.felord.r2dbc.config;

import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;

/**
 * The type User controller.
 *
 * @author felord.cn
 * @since 17 :07
 */
@RestController
@RequestMapping("/user")
public class UserController {
    @Resource
    private DatabaseClient databaseClient;

    /**
     * 查詢
     *
     * @return 返回Flux序列 包含所有的ClientUser
     */
    @GetMapping("/get")
    public Flux<ClientUser> clientUserFlux() {
        return databaseClient.execute("select * from client_user").as(ClientUser.class)
                .fetch()
                .all();
    }

    /**
     * 響應式寫入.
     *
     * @return Mono對象包含更新成功的條數
     */
    @GetMapping("/add")
    public Mono<Integer> insert() {
        ClientUser clientUser = new ClientUser();
        clientUser.setUserId("34345514644");
        clientUser.setUsername("felord.cn");
        clientUser.setPhoneNumber("3456121");
        clientUser.setGender(1);

        return databaseClient.insert().into(ClientUser.class)
                .using(clientUser)
                .fetch().rowsUpdated();
    }

}

調用接口就能獲取到期望的數據結果。

5. 總結

乍一看R2DBC並沒有想象中的那麼難，但是間接的需要了解Flux、Mono等抽象概念。同時目前來說如果不和Webflux框架配合也沒有使用場景。就本文的MySQL而言，R2DBC驅動還是社區維護（不得不說PgSQL就做的很好）。

然而需要你看清的是反應式才是未來。如果你要抓住未來就需要現在就了解一些相關的知識。這讓我想起五年前剛剛接觸Spring Boot的感覺。另外這裡有一份Spring官方關於R2DBC的PPT，也是讓你更好了解R2DBC的權威資料。可以關注：碼農小胖哥 回復r2dbc獲取。

關注公眾號：Felordcn 獲取更多資訊

個人博客：https://felord.cn

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※新北清潔公司,居家、辦公、裝潢細清專業服務

※別再煩惱如何寫文案,掌握八大原則!

※網頁設計一頭霧水該從何著手呢? 台北網頁設計公司幫您輕鬆架站!

※超省錢租車方案

※教你寫出一流的銷售文案?

如果你想從頭學習Jmeter，可以看看這個系列的文章哦

https://www.cnblogs.com/poloyy/category/1746599.html

 

前言

• JDBC Request 主要是向數據庫發送一個 JDBC 請求（sql 語句），並獲取返回的數據集
• 它需要和數據庫連接池配置（JDBC Connection Configuration）一起使用，可參考此篇博文：https://www.cnblogs.com/poloyy/p/13182706.html

 

JDBC Request

 

JDBC Request 界面介紹

 

字段含義

字段	含義
Variable Name Bound to Pool	數據庫連接池配置的名稱
Query Type	sql 語句的類型
SQL Query	sql 語句 語句結尾不需要添加 ;  變量用 ? 佔位
Parameter values	需要傳遞的變量值，多個變量用 , 分隔
Parameter types	變量類型
Variable Names	保存sql語句返回的數據和返回數據的總行數 用 , 分隔 跳過列用空
Result Variable Name	一個 Object 變量存儲所有返回值
Query timeout(s)	超時時間；默認0，代表無限時間
Limit ResultSet	和 limit 類似作用，限制 sql 語句返回結果集的行數
Handle ResultSet	如何定義 callable statements 返回的結果集；默認是存儲為字符串

後續通過各種栗子來深入理解常用字段的含義

 

舉栗子的前提

需要自己找一個有數據庫的數據來練手哦！這裏拿的表數據如下哈

 

只有 sql 語句的栗子

JDBC Request

沒啥特別的，平時 sql 怎麼寫，這裏就怎麼寫

 

運行結果

 

參數化的栗子

JDBC Request

 

運行結果

 

知識點

• 有幾個問號，Parameter value、Parameter type 填寫值的數量要保持一致，用,分隔
• 問號其實是佔位符，如果學過編程的童鞋應該也知道這種寫法，可以避免 SQL 注入的問題
• sql 中使用佔位符時，Query Type 必須選擇 Prepared Select Statement 或者 Prepared Update Statement 
• 我們可以用 Jmeter 變量去賦值，看下面栗子

 

參數化+變量的栗子

JDBC Request

 

運行結果

 

知識點

• 如果在 sql 語句中使用變量，且是字符串類型，需要加上引號（前提是變量值沒有加引號），如 ‘${name}‘ 
• 如果在 Parameter values 中使用變量，且是字符串類型，不需要加上引號，只需要在 Parameter types 里寫明為 varchar 即可

 

使用 Variable Names 的栗子

結構樹

 

JDBC Request

添加一個 Debug Sampler 就知道這個字段有什麼作用了

 

JDBC Request 運行結果

 

調試取樣器運行結果

 

知識點

• mysql：數據庫連接池對象
• a_#、b_#、c_#、d_#：代錶行數
• a_1：第 1 行、第 1 列
• b_2：第 2 行、第 2 列
• c_3：第 3 行、第 3 列
• d_3：第 3 行、第 4 列
• 以此類推….
• 一般如果 HTTP 請求需要用到 sql 查出來的數據的話，就會用到 Variable names 這個字段

 

使用 Result variable name 的栗子

JDBC Request

 

Debug Sampler  運行結果

 

知識點

該變量是個數組，每一個元素代表一條記錄

 

重點

關於通過 Variable names、Result variable name 獲取到的值如何提取，我們將在下一篇文章中詳細講解

 

使用 Limit ResultSet 的栗子

JDBC Request

 

 

運行結果

 

知識點

• Limit ResultSet 是對 sql 語句返回的結果集限制行數
•  limit 10 限制只返回了 10 條數據，然後 Limit ResultSet = 6 限制結果集最終只返回 6 條數據

 

  本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※帶您來了解什麼是 USB CONNECTOR  ?

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站，提高曝光率!

※綠能、環保無空污,成為電動車最新代名詞，目前市場使用率逐漸普及化

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※教你寫出一流的銷售文案?