一、引言

池的概念大家并不陌生，數(shù)據(jù)庫連接池、線程池等...大體來說，有三個(gè)優(yōu)點(diǎn)：

1. 降低資源消耗。
2. 提高響應(yīng)速度。
3. 便于統(tǒng)一管理。
   

以上是 “池化” 技術(shù)的相同特點(diǎn)，至于他們之間的不同點(diǎn)這里不講，兩者都是為了提高性能和效率，拋開實(shí)際做連連看找不同，沒有意義。

同樣，類比于線程池來說：

• 降低資源消耗：

重復(fù)利用線程池中已經(jīng)創(chuàng)建的線程，相比之下省去了線程創(chuàng)建和銷毀的性能消耗。

• 提高響應(yīng)速度：

當(dāng)有任務(wù)創(chuàng)建時(shí)，不必等待線程創(chuàng)建，可以立即執(zhí)行。

• 便于統(tǒng)一管理：

使用線程池，可以對線程統(tǒng)一管理，對線程的執(zhí)行狀態(tài)做統(tǒng)一監(jiān)控。

二、線程池的使用

public ThreadPoolExecutor(int corePoolSize,
             int maximumPoolSize,
             long keepAliveTime,
             TimeUnit unit,
             BlockingQueue<Runnable> workQueue,
       ThreadFactory threadFactory,
             RejectedExecutionHandler handler);

1、關(guān)鍵參數(shù)

• corePoolSize 核心線程數(shù)

當(dāng)向線程池中提交一個(gè)任務(wù)時(shí)，如果線程池中的線程數(shù)量小于核心線程數(shù)，即使存在空閑線程，也會新建一個(gè)線程來執(zhí)行當(dāng)前任務(wù)，直到線程數(shù)量大于或等于核心線程數(shù)。

• maximunPoolSize 最大線程數(shù)

當(dāng)任務(wù)隊(duì)列滿了，線程池中的線程數(shù)量小于最大線程數(shù)時(shí)，創(chuàng)建新線程執(zhí)行任務(wù)。對于無界隊(duì)列，忽略該參數(shù)。

• keepAliveTime 線程存活時(shí)間

大于核心線程數(shù)的那一部分線程的存活時(shí)間，如果這部分線程空閑超過這段時(shí)間，則進(jìn)行銷毀。

• workqueue 任務(wù)隊(duì)列

線程池中的線程數(shù)大于核心線程數(shù)時(shí)，將任務(wù)放入此隊(duì)列等待執(zhí)行。

• threadFactory 線程工廠

用于創(chuàng)建線程，工廠使用 new Threa() 的方式創(chuàng)建線程，并為每個(gè)線程做統(tǒng)一規(guī)則的命名：pool-m-thread-n（m為線程池的編號，n為線程池內(nèi)的線程編號）。

• handler 飽和策略

當(dāng)線程池和隊(duì)列都滿了，則根據(jù)此策略處理任務(wù)。

2、任務(wù)隊(duì)列類型

3、飽和策略類型

同時(shí)，還可以自行實(shí)現(xiàn) RejectedExecutionHandler 接口來自定義飽和策略，比如記錄日志、持久化等等。

void execute(Runnable command)

ThreadFactory namedThreadFactory =
new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build();
ExecutorService executor =
new ThreadPoolExecutor(
10,  
1000,
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(10),
namedThreadFactory,
new ThreadPoolExecutor.AbortPolicy());
executor.execute(
() -> {
System.out.println(1111);
});

注意使用 execute 方法提交任務(wù)時(shí)，沒有返回值。

Future<?> submit(Runnable task)

Future<Integer> future = executor.submit(() -> {
   return 1 + 1;
  });
Integer result = future.get();

還可以使用 submit 方法提交任務(wù)，該方法返回一個(gè) Future 對象，通過 Future#get( ) 方法可以獲得任務(wù)的返回值，該方法會一直阻塞知道任務(wù)執(zhí)行完畢。還可以使用 Future#get(long timeout, TimeUnit unit) 方法，該方法會阻塞一段時(shí)間后立即返回，而這時(shí)任務(wù)可能沒有執(zhí)行完畢。

5、關(guān)閉線程池

ThreadPoolExecutor 提供了 shutdown( ) 和 shutdownNow( ) 兩個(gè)方法關(guān)閉線程池。原理是首先遍歷線程池的工作線程，依次調(diào)用 interrupt( ) 方法中斷線程，這樣看來如果無法響應(yīng)中斷的任務(wù)就不能終止。

兩者區(qū)別是：

shutdownNow( )
shutdown( )

如果調(diào)用了其中一種方法，isShutdown 方法就會返回 true。當(dāng)所有的任務(wù)都已關(guān)閉后, 才表示線程池關(guān)閉成功，這時(shí)調(diào)用 isTerminaed 方法會返回 true。實(shí)際應(yīng)用中可以根據(jù)任務(wù)是否 一定要執(zhí)行完畢 的特性，決定使用哪種方法關(guān)閉線程池。

6、合理的配置線程池

通常我們可以 根據(jù) CPU 核心數(shù)量來設(shè)計(jì)線程池?cái)?shù)量 。

可以通過 Runtime.getRuntime().availableProcessors() 方法獲得當(dāng)前設(shè)備的物理核心數(shù)量。值得注意的是，如果應(yīng)用運(yùn)行在一些 docker 或虛擬機(jī)容器上時(shí)，該方法取得的是當(dāng)前物理機(jī)的 CPU 核心數(shù)。

• IO 密集型 2nCPU
• 計(jì)算密集型 nCPU+1

其中 n 為 CPU 核心數(shù)量。

為什么加 1：即使當(dāng)計(jì)算密集型的線程偶爾由于缺失故障或者其他原因而暫停時(shí)，這個(gè)額外的線程也能確保 CPU 的時(shí)鐘周期不會被浪費(fèi)。

三、線程池的運(yùn)行過程

當(dāng)提交一個(gè)新任務(wù)時(shí)，線程池的處理步驟：

1. 判斷當(dāng)前線程池內(nèi)的線程數(shù)量是否小于核心線程數(shù)，如果小于則新建線程執(zhí)行任務(wù)。否則，進(jìn)入下個(gè)階段。
2. 判斷隊(duì)列是否已滿，如果沒滿，則將任務(wù)加入等待隊(duì)列。否則，進(jìn)入下個(gè)階段。
3. 在上面基礎(chǔ)上判斷是否大于最大線程數(shù)，如果是根據(jù)響應(yīng)的策略處理。否則，新建線程執(zhí)行當(dāng)前任務(wù)。
   

線程池的源碼比較簡單易懂，感興趣的小伙伴可以自行查看 java.util.concurrent.ThreadPoolExecutor ，在線程池中每個(gè)任務(wù)都被包裝為一個(gè)一個(gè)的 Worker ，下面簡單看下 Worker 的 run( ) 方法：

try {
      while (task != null || (task = getTask()) != null) {
        w.lock();
        // If pool is stopping, ensure thread is interrupted;
        // if not, ensure thread is not interrupted. This
        // requires a recheck in second case to deal with
        // shutdownNow race while clearing interrupt
        if ((runStateAtLeast(ctl.get(), STOP) ||
           (Thread.interrupted() &&
           runStateAtLeast(ctl.get(), STOP))) &&
          !wt.isInterrupted())
          wt.interrupt();
        try {
          beforeExecute(wt, task);
          Throwable thrown = null;
          try {
            task.run();
          } catch (RuntimeException x) {
            thrown = x; throw x;
          } catch (Error x) {
            thrown = x; throw x;
          } catch (Throwable x) {
            thrown = x; throw new Error(x);
          } finally {
            afterExecute(task, thrown);
          }
        } finally {
          task = null;
          w.completedTasks++;
          w.unlock();
        }
      }
      completedAbruptly = false;
    } finally {
      processWorkerExit(w, completedAbruptly);
    }

可以看到不斷的循環(huán)取出 Task 并執(zhí)行，而在任務(wù)的執(zhí)行前后，有 beforeExecute 和 afterExecute 方法，我們可以實(shí)現(xiàn)兩個(gè)方法實(shí)現(xiàn)一些監(jiān)控邏輯。除此之外還可以集合線程池的一些屬性或者重寫 terminated() 方法在線程池關(guān)閉時(shí)進(jìn)行監(jiān)控。

四、常見的幾種線程池實(shí)現(xiàn)

在 Executors 中提供了集中常見的線程池，分別應(yīng)用在不同的場景。

• FixThreadPool 固定數(shù)量的線程池，適用于對線程管理，高負(fù)載的系統(tǒng)
• SingleThreadPool 只有一個(gè)線程的線程池，適用于保證任務(wù)順序執(zhí)行
• CacheThreadPool 創(chuàng)建一個(gè)不限制線程數(shù)量的線程池，適用于執(zhí)行短期異步任務(wù)的小程序，低負(fù)載系統(tǒng)
• ScheduledThreadPool 定時(shí)任務(wù)使用的線程池，適用于定時(shí)任務(wù)
  

上面幾種線程池的特性主要依賴于 ThreadPoolExecutor 的幾個(gè)參數(shù)來實(shí)現(xiàn)，不同的核心線程數(shù)量，以及不同類型的阻塞隊(duì)列，同時(shí)我們還可以自行實(shí)現(xiàn)自己的線程池滿足業(yè)務(wù)需求。

值得注意的是，并不推薦使用 Executors 創(chuàng)建線程池，詳見下：

Executors.newFixedThreadPool(int nThread)

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                   0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
 }

繼續(xù)來看 LinkedBlockingQueue ：

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);
}

可以看到使用 LinkedBlockingQueue 創(chuàng)建的是 Integer.MAX_VALUE 大小的隊(duì)列，會堆積大量的請求，從而造成 OOM

Executors.newSingleThreadExexutor( )

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
      (new ThreadPoolExecutor(1, 1,
                  0L, TimeUnit.MILLISECONDS,
                  new LinkedBlockingQueue<Runnable>()));
}

同樣，使用的 LinkedBlockingQueue ，一樣的情況

Executors.newCachedThreadPool( )

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                   60L, TimeUnit.SECONDS,
                   new SynchronousQueue<Runnable>());
}

代碼課件線程池使用的最大線程數(shù)是 Integer.MAX_VALUE ，可能會創(chuàng)建大量線程，導(dǎo)致 OOM

Executors.newScheduleThreadPool()

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
       new DelayedWorkQueue());
}

和上面是一樣的問題，最大線程數(shù)是 Integer.MAX_VALUE
所以原則上來說禁止使用 Executors 創(chuàng)建線程池， 而使用 ThreadPoolExecutor 的構(gòu)造函數(shù)來創(chuàng)建線程池。

五、結(jié)語

線程池在開發(fā)中還是比較常見的，結(jié)合不同的業(yè)務(wù)場景，結(jié)合最佳實(shí)踐配置正確的參數(shù)，可以幫助我們的應(yīng)用性能得到提升。

以上就是談?wù)凧ava 線程池的詳細(xì)內(nèi)容，更多關(guān)于Java 線程池的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論