public class ThreadPoolDemo {
public static void main(String[] args) {
// 1. 創建固定大小線程池
ExecutorService pool = Executors.newFixedThreadPool(2);
// 2. 提交任務
for (int i = 0; i < 5; i++) {
pool.execute(new MyTask());
}
// 3. 關閉線程池
pool.shutdown();
}
}
class MyTask implements Runnable {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 執行任務");
}
}
輸出： 兩個線程交替執行 5 個任務，驗證線程複用機制。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
關鍵點：corePoolSize == maximumPoolSize，使用無界 LinkedBlockingQueue，任務積壓不會創建額外線程。
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
// 參數校驗
if (corePoolSize < 0 || maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize || keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
if (t.isDaemon()) t.setDaemon(false); // 非守護線程
if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
統一命名格式：pool-N-thread-M，便於日誌追蹤。
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
public static class AbortPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString());
}
}
public void execute(Runnable command) {
if (command == null) throw new NullPointerException();
int c = ctl.get(); // 獲取控制狀態：高3位=運行狀態，低29位=線程數
// 情況1：線程數 < corePoolSize → 創建核心線程執行
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true)) return;
c = ctl.get();
}
// 情況2：線程數 >= corePoolSize 且線程池運行中 → 入隊列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 二次檢查：線程池意外關閉 → 移除任務並拒絕
if (!isRunning(recheck) && remove(command))
reject(command);
// 隊列有任務但無線程 → 創建非核心線程保活
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 情況3：隊列滿 → 嘗試創建非核心線程，失敗則拒絕
else if (!addWorker(command, false))
reject(command);
}
三大決策路徑圖解：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 狀態檢查：SHUTDOWN 且有首任務/隊列不空 → 拒絕
if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
// 容量檢查
if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// CAS 增加線程計數
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
if (runStateOf(c) != rs) continue retry; // 狀態變化重試
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int c = ctl.get();
int rs = runStateOf(c);
if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) throw new IllegalThreadStateException();
workers.add(w); // 加入工作集合
int s = workers.size();
if (s > largestPoolSize) largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start(); // 啟動線程 → 觸發 Worker.run()
workerStarted = true;
}
}
} finally {
if (!workerStarted) addWorkerFailed(w);
}
return workerStarted;
}
關鍵點：
mainLock 同步加入集合core=true 以 corePoolSize 為界，false 以 maximumPoolSize 為界Worker 實現 Runnable，run() 中循環調用 getTask() 從隊列獲取任務public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess(); // 權限檢查
advanceRunState(SHUTDOWN); // 狀態轉為 SHUTDOWN
interruptIdleWorkers(); // 中斷空閒線程
onShutdown(); // 鉤子方法（定時任務擴展用）
} finally {
mainLock.unlock();
}
tryTerminate(); // 嘗試終止
}
shutdown() 不接受新任務，但會處理隊列中已有任務。shutdownNow() 則會中斷正在執行的任務並返回未處理任務列表。
