並發編程系列之如何正確使用執行緒池？

並發編程系列部落格

並發編程系列之如何正確使用執行緒池？在上一章節的學習中，我們掌握了執行緒的基本知識，接著本部落格會繼續學習多執行緒中的執行緒池知識

1、執行緒是不是越多越好？

在學習多執行緒之前，讀者可能會有疑問？如果單執行緒跑得太慢，那麼是否就能多創建多個執行緒來跑任務？並發的情況，執行緒是不是創建越多越好？這是一個很經典的問題，畫圖表示一下創建很多執行緒的情況，然後進行情況分析。
在这里插入图片描述

  • 創建執行緒和銷毀執行緒都是需要時間的，如果創建時間+銷毀時間>執行任務時間就很不划算
  • 創建後的執行緒是需要記憶體去存放的，創建的執行緒對應一個Thread對象，對象是會佔用JVM的堆記憶體的，根據jvm規範，一個執行緒默認最大棧大小為1M，這個棧空間也是需要從系統記憶體中分配的，所以執行緒越多，需要的記憶體就越多
  • 創建執行緒，作業系統是需要頻繁進行執行緒上下文切換的，所以執行緒創建太多，是會影響性能的

上下文切換(context switch)：對於單核CPU來說，在一個時刻只能運行一個執行緒，對於並行來說，單核cpu也是可以支援多執行緒執行程式碼的，CPU是通過給執行緒分配時間片來解決的，所謂時間片是CPU給每個執行緒分配的時間，時間片的時間是非常短的，所以執行完成一個時間片後，進行任務切換，切換之前先保存這個任務的狀態，以便於下次換回來的時候，可以載入這個任務的狀態，所以從保存任務狀態到再載入任務的過程稱為上下文切換，不僅在執行緒間可以上下文切換，進程也同樣可以

2、如何正確使用多執行緒？

  • 如果是計算型任務？
    CPU數量的1~2倍即可
  • 如果是IO密集型任務？
    就需要多一些執行緒，要根據具體的io阻塞時長來進行考量決定

3、Java執行緒池的工作原理

在这里插入图片描述

  • 接收任務，放入執行緒池的任務倉庫
  • 工作執行緒從執行緒池的任務倉庫取，執行
  • 沒有任務時，執行緒阻塞，有任務時喚醒執行緒

4、掌握JUC執行緒池API

  • Executor ： 介面類

  • ExecutorService：加入關閉方法和對Runnable、Callable、Future的支援
    在这里插入图片描述

    • shutdown：已經提交的會執行完成
    • shutdownNow：正在執行的會執行完成，未來執行的返回
    • awaitTermination：阻塞等待任務關閉完成
    • submit類型的：都是提交任務的，支援Runnable和Callable
    • invokeAll類型的：執行集合中所有任務

  • ScheduleExecutorService ：加入對定時任務的支援
    在这里插入图片描述
    其中schedule(Runablle , long, Timeunit)和schedule(Callable<V> , long, TimeUnit)表示的是多久後執行，而scheduleAtFixedRate方法和scheduleWithFixedDelay方法表示的都是周期性重複執行的

再描述scheduleAtFixedRate方法和scheduleWithFixedDelay方法的區別：
scheduleAtFixedRate：以固定的時間頻率重複執行任務，如每10s ，也就是兩個任務直接以固定的時間間隔執行，不管任務執行完成與否
在这里插入图片描述

scheduleWithFixedDelay：以固定的任務時延遲來重複執行任務，這種任務不管任務執行多久都執行完成，然後隔預定的如3s，接著執行下一個任務，每個任務之間的間隔都是一樣的

在这里插入图片描述

  • Executors：快速得到執行緒池的工具類，創建執行緒池的工廠類

    • newFixedThreadPool(int nThreads)：創建一個固定大小、任務隊列 無界的執行緒池。執行緒池的核心執行緒數=最大執行緒池=nThreads
    • newCachedThreadPool()：創建的是一個大小無界的緩衝執行緒池。它的任務隊列是一個同步隊列。如果隊列中有空閑的執行緒，則用空閑執行緒執行，如果沒有就創建新執行緒執行。池中執行緒空閑超過60s，就會被釋放。緩衝執行緒池使用於執行耗時比較小的非同步任務。執行緒池的核心執行緒數=0，最大執行緒池=Integer.MAX_VALUE
    • newSingleThreadExecutor()：創建的是只有一個執行緒來執行無界任務隊列的單一執行緒池。該執行緒池按順序執行一個一個加入的任務，任何時刻都只有一個執行緒在執行。單一執行緒池和newFixedThreadPool(1)的區別在於，單一執行緒池的池大小是不能再改變的
    • newScheduleThreadPool(int corePoolSize)： 能定時執行任務的執行緒池，該池的核心執行緒數由參數corePoolSize指定，最大執行緒數=Integer.MAX_VALUE
    • newWorkStealingPool()：以當前系統可用的處理器數作為並行級別創建的work-stealing thread pool(ForkJoinPool)
    • newWorkStealingPool(int parallelism)：以指定的parallelism並行級別創建的work-stealing thread pool(ForkJoinPool)

  • ThreadPoolExecutor：執行緒池的標準實現
    在这里插入图片描述
    下面列舉出ThreadPoolExecutor的主要參數：

參數 描述
corePoolSize 核心執行緒數量
maxPoolSize 最大執行緒數量
keepAliveTime+時間單位 空閑執行緒的存活時間
ThreadFactory 執行緒工廠，用於創建執行緒
workQueue 用於存放任務的隊列，可以稱之為工作隊列
Handler 用於處理被拒絕的任務

雖然Executors使用起來很方便，不過在阿里編程規範里是強調了慎用Executors創建執行緒池，下面摘錄自阿里編程規範手冊：

【強制】執行緒池不允許使用Executors去創建，而是通過ThreadPoolExecutor的方式，
這樣的處理方式讓寫的同學更加明確執行緒池的運行規則，規避資源耗盡的風險。
說明：Executors各個方法的弊端：
1）newFixedThreadPool和newSingleThreadExecutor:
  主要問題是堆積的請求處理隊列可能會耗費非常大的記憶體，甚至OOM。
2）newCachedThreadPool和newScheduledThreadPool:
  主要問題是執行緒數最大數是Integer.MAX_VALUE，可能會創建數量非常多的執行緒，甚至OOM。

ThreadPoolExecutor的基本參數：

new ThreadPoolExecutor(
				2, // 核心執行緒數 
				5, // 最大執行緒數
                60L, // keepAliveTime,執行緒空閑超過這個數，就會被銷毀釋放
                TimeUnit.SECONDS, // keepAliveTime的時間單位
                new ArrayBlockingQueue(5)); // 傳入邊界為5的工作隊列

畫流程圖表示，執行緒池的核心參數是corePoolSize、maxPoolSize、workQueue（工作隊列）
在这里插入图片描述
執行緒池工作原理示意圖：任務可以一直放，直到執行緒池滿了的情況，才會拒絕，然後除了核心執行緒，其它的執行緒會被合理回收。所以正常情況下，執行緒池中的執行緒數量會處在 corePoolSize 與 maximumPoolSize 的閉區間內
在这里插入图片描述
ThreadPoolExecutor基本實例：

ExecutorService service = new ThreadPoolExecutor(2, 5,
                60L, TimeUnit.SECONDS,
                new ArrayBlockingQueue(5));
 service.execute(() ->{
     System.out.println(String.format("thread name:%s",Thread.currentThread().getName()));
 });
  // 避免記憶體泄露，記得關閉執行緒池
 service.shutdown();

ThreadPoolExecutor加上Callable、Future使用的例子：

public static void  main(String[] args) {
	ExecutorService service = new ThreadPoolExecutor(2, 5,
	                60L, TimeUnit.SECONDS,
	                new ArrayBlockingQueue(5));
	
	Future<Integer> future = service.submit(new CallableTask());
	Thread.sleep(3000);
	System.out.println("future is done?" + future.isDone());
	if (future.isDone()) {
	    System.out.println("callableTask返回參數:"+future.get());
	}
	service.shutdown();
}

static class CallableTask implements Callable<Integer>{
     @Override
     public Integer call() {
         return ThreadLocalRandom.current().ints(0, (99 + 1)).limit(1).findFirst().getAsInt();
     }
 }
Tags:

VirMach 便宜 VPS