當前位置：首頁 > 编程语言 > java >内容正文

java

Java 线程池艺术探索

發布時間：2025/3/21 java 22 豆豆

生活随笔收集整理的這篇文章主要介紹了 Java 线程池艺术探索小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

線程池

Wiki 上是這樣解釋的：Thread Pool

作用：利用線程池可以大大減少在創建和銷毀線程上所花的時間以及系統資源的開銷！

下面主要講下線程池中最重要的一個類 ThreadPoolExecutor 。

ThreadPoolExecutor

ThreadPoolExecutor 構造器：

有四個構造器的，挑了參數最長的一個進行講解。

七個參數：

corePoolSize：核心池的大小，在創建了線程池后，默認情況下，線程池中并沒有任何線程，而是等待有任務到來才創建線程去執行任務，當有任務來之后，就會創建一個線程去執行任務，當線程池中的線程數目達到corePoolSize后，就會把到達的任務放到緩存隊列當中；
maximumPoolSize：線程池最大線程數；
keepAliveTime：表示線程沒有任務執行時最多保持多久時間會終止；
unit：參數keepAliveTime的時間單位（DAYS、HOURS、MINUTES、SECONDS 等）；
workQueue：阻塞隊列，用來存儲等待執行的任務；
- ArrayBlockingQueue （有界隊列）
- LinkedBlockingQueue （無界隊列）
- SynchronousQueue
threadFactory：線程工廠，主要用來創建線程
handler：拒絕處理任務的策略
- AbortPolicy：丟棄任務并拋出 RejectedExecutionException 異常。（默認這種）
- DiscardPolicy：也是丟棄任務，但是不拋出異常
- DiscardOldestPolicy：丟棄隊列最前面的任務，然后重新嘗試執行任務（重復此過程）
- CallerRunsPolicy：由調用線程處理該任務

重要方法：

execute()：通過這個方法可以向線程池提交一個任務，交由線程池去執行；
shutdown()：關閉線程池；

execute() 方法：

注：JDK 1.7 和 1.8 這個方法有點區別，下面代碼是 1.8 中的。

public void execute(Runnable command) {

if (command == null)

throw new NullPointerException();

int c = ctl.get();

//1、如果當前的線程數小于核心線程池的大小，根據現有的線程作為第一個 Worker 運行的線程，新建一個 Worker，addWorker 自動的檢查當前線程池的狀態和 Worker 的數量，防止線程池在不能添加線程的狀態下添加線程

if (workerCountOf(c) < corePoolSize) {

if (addWorker(command, true))

return;

c = ctl.get();

}

//2、如果線程入隊成功，然后還是要進行 double-check 的，因為線程在入隊之后狀態是可能會發生變化的

if (isRunning(c) && workQueue.offer(command)) {

int recheck = ctl.get();

// recheck 防止線程池狀態的突變，如果突變，那么將 reject 線程，防止 workQueue 中增加新線程

if (! isRunning(recheck) && remove(command))

reject(command);

else if (workerCountOf(recheck) == 0)//上下兩個操作都有 addWorker 的操作，但是如果在workQueue.offer 的時候 Worker 變為 0，那么將沒有 Worker 執行新的 task，所以增加一個 Worker.

addWorker(null, false);

}

//3、如果 task 不能入隊(隊列滿了)，這時候嘗試增加一個新線程，如果增加失敗那么當前的線程池狀態變化了或者線程池已經滿了然后拒絕task

else if (!addWorker(command, false))

reject(command);

}

其中調用了 addWorker() 方法：

private boolean addWorker(Runnable firstTask, boolean core) {// firstTask: 新增一個線程并執行這個任務，可空，增加的線程從隊列獲取任務；core：是否使用 corePoolSize 作為上限，否則使用 maxmunPoolSize

retry:

for (;;) {

int c = ctl.get();

int rs = runStateOf(c);

// Check if queue empty only if necessary.

/**

* rs!=Shutdown || fistTask！=null || workQueue.isEmpty

* 如果當前的線程池的狀態 > SHUTDOWN 那么拒絕 Worker 的 add 如果 =SHUTDOWN

* 那么此時不能新加入不為 null 的 Task，如果在 workQueue 為 empty 的時候不能加入任何類型的 Worker，

* 如果不為 empty 可以加入 task 為 null 的 Worker, 增加消費的 Worker

if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null &&! workQueue.isEmpty()))

return false;

for (;;) {

int wc = workerCountOf(c);

//如果當前的數量超過了 CAPACITY，或者超過了 corePoolSize 和 maximumPoolSize（試 core 而定）

if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))

return false;

//CAS 嘗試增加線程數，如果失敗，證明有競爭，那么重新到 retry。

if (compareAndIncrementWorkerCount(c))// AtomicInteger 的 CAS 操作;

break retry;

c = ctl.get(); // Re-read ctl

//判斷當前線程池的運行狀態,狀態發生改變，重試 retry;

if (runStateOf(c) != rs)

continue retry;

// else CAS failed due to workerCount change; retry inner loop

}

boolean workerStarted = false;

boolean workerAdded = false;

Worker w = null;

try {

w = new Worker(firstTask);// Worker 為內部類，封裝了線程和任務，通過 ThreadFactory 創建線程，可能失敗拋異常或者返回 null

final Thread t = w.thread;

if (t != null) {

final ReentrantLock mainLock = this.mainLock;

mainLock.lock();

try {

// Recheck while holding lock.

// Back out on ThreadFactory failure or if

// shut down before lock acquired.

int rs = runStateOf(ctl.get());

if (rs < SHUTDOWN ||

(rs == SHUTDOWN && firstTask == null)) {

if (t.isAlive()) // precheck that t is startable

// SHUTDOWN 以后的狀態和 SHUTDOWN 狀態下 firstTask 為 null，不可新增線程

throw new IllegalThreadStateException();

workers.add(w);

int s = workers.size();

if (s > largestPoolSize)

largestPoolSize = s;//記錄最大線程數

workerAdded = true;

}

} finally {

mainLock.unlock();

}

if (workerAdded) {

t.start();

workerStarted = true;

}

} finally {

if (! workerStarted)

addWorkerFailed(w);//失敗回退,從 wokers 移除 w, 線程數減一，嘗試結束線程池(調用tryTerminate 方法)

}

return workerStarted;

}

示意圖：

執行流程：

1、當有任務進入時，線程池創建線程去執行任務，直到核心線程數滿為止

2、核心線程數量滿了之后，任務就會進入一個緩沖的任務隊列中

當任務隊列為無界隊列時，任務就會一直放入緩沖的任務隊列中，不會和最大線程數量進行比較
當任務隊列為有界隊列時，任務先放入緩沖的任務隊列中，當任務隊列滿了之后，才會將任務放入線程池，此時會與線程池中最大的線程數量進行比較，如果超出了，則默認會拋出異常。然后線程池才會執行任務，當任務執行完，又會將緩沖隊列中的任務放入線程池中，然后重復此操作。

shutdown() 方法：

public void shutdown() {

final ReentrantLock mainLock = this.mainLock;

mainLock.lock();

try {

//判斷是否可以操作目標線程

checkShutdownAccess();

//設置線程池狀態為 SHUTDOWN, 此處之后，線程池中不會增加新 Task

advanceRunState(SHUTDOWN);

//中斷所有的空閑線程

interruptIdleWorkers();

onShutdown(); // hook for ScheduledThreadPoolExecutor

} finally {

mainLock.unlock();

}

//轉到 Terminate

tryTerminate();

}

參考資料：深入理解java線程池—ThreadPoolExecutor

JDK 自帶四種線程池分析與比較

1、newFixedThreadPool

創建一個定長線程池，可控制線程最大并發數，超出的線程會在隊列中等待。

2、newSingleThreadExecutor

創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。

3、newCachedThreadPool

創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程。

4、newScheduledThreadPool

創建一個定長線程池，支持定時及周期性任務執行。

四種線程池其實內部方法都是調用的 ThreadPoolExecutor 類，只不過利用了其不同的構造器方法而已（傳入自己需要傳入的參數），那么利用這個特性，我們自己也是可以實現自己定義的線程池的。

自定義線程池

1、創建任務類

package com.zhisheng.thread.threadpool.demo;

/**

* Created by 10412 on 2017/7/24.

* 任務

public class MyTask implements Runnable

{

private int taskId; //任務 id

private String taskName; //任務名字

public int getTaskId() {

return taskId;

}

public void setTaskId(int taskId) {

this.taskId = taskId;

}

public String getTaskName() {

return taskName;

}

public void setTaskName(String taskName) {

this.taskName = taskName;

}

public MyTask(int taskId, String taskName) {

this.taskId = taskId;

this.taskName = taskName;

}

@Override

public void run() {

System.out.println("當前正在執行 ****** 線程Id-->" + taskId + ",任務名稱-->" + taskName);

try {

Thread.currentThread().sleep(5 * 1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("線程Id-->" + taskId + ",任務名稱-->" + taskName + " ----------- 執行完畢！");

}

2、自定義拒絕策略，實現 RejectedExecutionHandler 接口，重寫 rejectedExecution 方法

package com.zhisheng.thread.threadpool.demo;

import java.util.concurrent.RejectedExecutionHandler;

import java.util.concurrent.ThreadPoolExecutor;

/**

* Created by 10412 on 2017/7/24.

* 自定義拒絕策略，實現 RejectedExecutionHandler 接口

public class RejectedThreadPoolHandler implements RejectedExecutionHandler

{

public RejectedThreadPoolHandler() {

}

@Override

public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {

System.out.println("WARNING 自定義拒絕策略: Task " + r.toString() + " rejected from " + executor.toString());

}

3、創建線程池

package com.zhisheng.thread.threadpool.demo;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

/**

* Created by 10412 on 2017/7/24.

public class ThreadPool

{

public static void main(String[] args) {

//核心線程數量為 2，最大線程數量 4，空閑線程存活的時間 60s，有界隊列長度為 3,

//ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 4, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(3));

//核心線程數量為 2，最大線程數量 4，空閑線程存活的時間 60s，無界隊列,

//ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 4, 60L, TimeUnit.SECONDS, new LinkedBlockingDeque<>());

//核心線程數量為 2，最大線程數量 4，空閑線程存活的時間 60s，有界隊列長度為 3, 使用自定義拒絕策略

ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 4, 60, TimeUnit.SECONDS,

new ArrayBlockingQueue<Runnable>(3), new RejectedThreadPoolHandler());

for (int i = 1; i <= 10; i++) {

//創建 10 個任務

MyTask task = new MyTask(i, "任務" + i);

//運行

pool.execute(task);

System.out.println("活躍的線程數："+pool.getActiveCount() + ",核心線程數：" + pool.getCorePoolSize() + ",線程池大小：" + pool.getPoolSize() + ",隊列的大小" + pool.getQueue().size());

}

//關閉線程池

pool.shutdown();

}

這里運行結果就不截圖了，我在本地測試了代碼是沒問題的，感興趣的建議還是自己跑一下，然后分析下結果是不是和前面分析的一樣，如有問題，請在我博客下面評論！

總結

本文一開始講了線程池的介紹和好處，然后分析了線程池中最核心的 ThreadPoolExecutor 類中構造器的七個參數的作用、類中兩個重要的方法，然后在對比研究了下 JDK 中自帶的四種線程池的用法和內部代碼細節，最后寫了一個自定義的線程池。

總結

以上是生活随笔為你收集整理的Java 线程池艺术探索的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： Hystrix文档-实现原理
下一篇： Java 多线程：Inheritable