public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    //線程池狀態(tài) 高3位表示線程狀態(tài) 低29位代表線程數(shù)量
    int c = ctl.get();
    //判斷當前線程池線程數(shù)量是否小于核心線程數(shù)
    if (workerCountOf(c) < corePoolSize) {
        //作為核心線程數(shù)進行線程的創(chuàng)建，并且創(chuàng)建成功線程會將command的任務執(zhí)行--》對應圖上的直接執(zhí)行
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //創(chuàng)建核心線程失敗或者當前線程數(shù)量超過核心線程數(shù)
    //當前線程池是否還在運行狀態(tài)，嘗試將任務添加到阻塞隊列 --》對應圖上的緩沖執(zhí)行
    //BlockingQueue隊列的頂級抽象定義了offer不是進行阻塞添加而是立即返回，添加失敗直接返回false，區(qū)別于put
    if (isRunning(c) && workQueue.offer(command)) {
        //重新獲取線程池標志位
        int recheck = ctl.get();
        //如果線程此時不在運行狀態(tài)中，那么將任務刪除
        if (! isRunning(recheck) && remove(command))
            //刪除任務成功，走拒絕策略拒絕掉當前任務
            reject(command);
        else if (workerCountOf(recheck) == 0)
            //如果線程池中的工作線程都沒有的時候，這里需要創(chuàng)建一個線程去執(zhí)行添加到隊列中的任務
            //防止因為并發(fā)的原因工作線程都被終止掉了，此時任務在阻塞隊列里等著，缺沒有工作線程
            addWorker(null, false);
    }
    //到這里那就是添加隊列失敗，或者線程池狀態(tài)異常，但是這里仍然嘗試進行創(chuàng)建一個worker
    //如果創(chuàng)建失敗，也是走拒絕策略拒絕當前任務
    else if (!addWorker(command, false))
        reject(command);
}

//核心邏輯其實就是在無限循環(huán)創(chuàng)建一個worker，創(chuàng)建失敗直接返回，創(chuàng)建成功，則將worker執(zhí)行
// 因為worker有thread的成員變量，最終添加worker成功，會啟動線程的start方法
//start方法最終會回調(diào)到外層的runWorker方法，改方法會不停的從阻塞隊列里以阻塞的take方式
//獲取任務，除非達到能被終止的條件，此時當前線程會終止
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        
        // Check if queue empty only if necessary.
        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;
            //不停的重試添加worker的計數(shù)，只有添加成功的才會進行后續(xù)的worker啟動
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            //重試期間，如果其他線程導致線程池狀態(tài)不一致了。重新回到第一個循環(huán)進行check判斷
            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);
        final Thread t = w.thread;
        if (t != null) {
            //這里加鎖一個是workers.add時需要加鎖，另外是防止其他線程已經(jīng)在嘗試修改線程池狀態(tài)了
            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
                        throw new IllegalThreadStateException();
                    //將worker的引用添加到workers的hashSet中
                    workers.add(w);
                    int s = workers.size();
                    //更新線程池此時最大的線程數(shù)
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            //如果添加成功，就啟動worker中的線程
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        //這里添加失敗的話，需要把線程池的count數(shù)進行--，并且要把worker引用從hashSer中移除
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

//線程池的狀態(tài)和計數(shù)采用一個Integer變量設置的
//這里之所以用一個變量來儲存狀態(tài)和數(shù)量，其實很有講究的，因為我們在上面的運行原理上可以看到
//源碼中有大量的進行狀態(tài)以及數(shù)量的判斷，如果分開采用變量的記錄的話，在維護二者一致性方面
//可能就需要加鎖的維護成本了，而且計算中都是位移運算也是非常高效的
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//線程池的大小由ctl低29位表示，現(xiàn)成狀態(tài)由ctl高3位表示
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;


// 線程池的狀態(tài)通過簡單的位移就能計算出來，狀態(tài)只能從低到高流轉(zhuǎn)，不能逆向
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;


// 這里是獲取線程狀態(tài)以及獲取線程數(shù)量的簡單高效的位移方法
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

先說下結(jié)論：


RUNNING     這個就是線程池運行中狀態(tài)，我們可以添加任務也可以處理阻塞隊列任務


SHUTDOWN   不能添加新的任務，但是會將阻塞隊列中任務執(zhí)行完畢


STOP       不能添加新的任務，執(zhí)行中的線程也會被打斷，也不會處理阻塞隊列的任務


TIDYING    所有線程都被終止，并且workCount=0時會被置為的狀態(tài)


TERMINATED   調(diào)用完鉤子方法terminated()被置為的狀態(tài)

//線程池關(guān)閉
 public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        //循環(huán)cas設置線程池狀態(tài)，直到成功或狀態(tài)已經(jīng)state>=SHUTDOWN
        advanceRunState(SHUTDOWN);
        //這個是真正得出結(jié)論的地方
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}


private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}


//打斷空閑的線程，如何判斷線程是否空閑還是運行？
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            //worker的線程沒有被打斷過，并且能獲取到worker的aqs獨占鎖
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    //打斷當前線程，如果線程在阻塞隊列中阻塞，此時會被中斷
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

//循環(huán)cas修改線程池狀態(tài)為stop。打斷所有線程，取出阻塞隊列的所有任務
public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //檢查線程的權(quán)限
        checkShutdownAccess();
        //將狀態(tài)case為stop
        advanceRunState(STOP);
        //打斷所有worker不管是不是正在執(zhí)行任務
        interruptWorkers();
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}


//這里獲取鎖之后。打斷了所有的線程
private void interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}

//這個方法在每個線程退出時都會進行調(diào)用，如果是運行中、或者狀態(tài)大于等于TIDYING或者shutdown但是隊列不為空都
//直接返回，如果不滿足以上條件，并且線程數(shù)不為0的話，打斷一個空閑線程
final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }


        //此時到這里，狀態(tài)要么為STOP。要么是shutdown并且隊列為空了
        // 獲取一個鎖，嘗試cas修改狀態(tài)為TIDYING
        //調(diào)用terminated()的鉤子方法，
        //修改線程池為終態(tài)TERMINATED，并且喚醒阻塞在termination隊列上的線程
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

Callable<String> task = () -> {
    while (true) {
        //缺少此檢查將導致超時失效
        if (Thread.interrupted()) break; 
        // 耗時計算...
    }
    return "done";
};

Callable<Integer> task = () -> {
    Files.copy(in, path); // 某些NIO操作不響應中斷
    return 1;
};

Callable<Result> task = () -> {
    //未設查詢超時時間
    return jdbcTemplate.query("SELECT * FROM large_table"); 
};

new ThreadPoolExecutor(
    100, 100, // 核心線程數(shù)過大
    0L, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<>() // 無界隊列
);

import java.util.*;
import java.util.concurrent.*;


public class InvokeAllTimeoutDemo {
    
    // 模擬耗時任務（可配置是否響應中斷）
    static class Task implements Callable<String> {
        private final int id;
        private final long durationMs;
        private final boolean respectInterrupt;
        
        Task(int id, long durationMs, boolean respectInterrupt) {
            this.id = id;
            this.durationMs = durationMs;
            this.respectInterrupt = respectInterrupt;
        }
        
        @Override
        public String call() throws Exception {
            System.out.printf("Task %d started%n", id);
            long start = System.currentTimeMillis();


            // 模擬工作（檢查中斷狀態(tài)）
            while (System.currentTimeMillis() - start < durationMs) {
                if (respectInterrupt && Thread.interrupted()) {
                    throw new InterruptedException("Task " + id + " interrupted");
                }
                // 不響應中斷的任務會繼續(xù)執(zhí)行
            }


            System.out.printf("Task %d completed%n", id);
            return "Result-" + id;
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        
        List<Callable<String>> tasks = Arrays.asList(
            new Task(1, 2000, true),   // 2秒，響應中斷
            new Task(2, 10000, false)  // 10秒，不響應中斷
        );
        
        System.out.println("Invoking with 3s timeout...");
        try {
            //設置3秒超時
            List<Future<String>> futures = executor.invokeAll(tasks, 3, TimeUnit.SECONDS);


            for (Future<String> f : futures) {
                // 明確處理取消狀態(tài)
                if (f.isCancelled()) {
                    System.out.println("Task was cancelled");  
                } else {
                    try {
                        System.out.println("Result: " + f.get(100, TimeUnit.MILLISECONDS));
                    } catch (TimeoutException | ExecutionException e) {
                        System.out.println("Task failed: " + e.getCause());
                    }
                }
            }
        } finally {
            executor.shutdownNow();
            System.out.println("Executor shutdown");
        }
    }
}

public class ThreadPoolExceptionDemo {
    
    public static void main(String[] args) {
        // 創(chuàng)建單線程線程池（便于觀察異常）
        ExecutorService executor = Executors.newSingleThreadExecutor();
        
        // 場景1：Callable拋出異常（通過Future.get()捕獲）
        Future<String> future1 = executor.submit(() -> {
            System.out.println("[Callable] 開始執(zhí)行");
            Thread.sleep(100);
            throw new RuntimeException("Callable故意拋出的異常");
        });
        
        try {
            System.out.println("Callable結(jié)果: " + future1.get());
        } catch (ExecutionException e) {
            System.err.println("捕獲到Callable異常: " + e.getCause().getMessage());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        // 場景2：Runnable拋出異常（同樣通過Future.get()捕獲）
        Future<?> future2 = executor.submit(() -> {
            System.out.println("[Runnable] 開始執(zhí)行");
            throw new IllegalArgumentException("Runnable故意拋出的異常");
        });
        
        try {
            future2.get(); // Runnable成功時返回null
        } catch (ExecutionException e) {
            System.err.println("捕獲到Runnable異常: " + e.getCause().getMessage());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        // 場景3：未處理的任務異常（需設置異常處理器）
        executor.submit(() -> {
            System.out.println("[未捕獲的任務] 開始執(zhí)行");
            throw new IllegalStateException("這個異常會被默認處理器處理");
        });
        
        executor.shutdown();
    }
}

//自定義當前線程組創(chuàng)建線程的統(tǒng)一異常處理，類似于controller的統(tǒng)一異常處理機制
ThreadFactory myThreadFactory = new ThreadFactory() {
    private final AtomicInteger atomicInteger = new AtomicInteger(0);
    
    private final String threadNamePrefix = "myThreadFactory-";
    
    @Override
    public Thread newThread(Runnable r) {
        Thread t =  new Thread(r,threadNamePrefix + atomicInteger.getAndIncrement());
        t.setUncaughtExceptionHandler((thread, throwable) -> {
        //異常的統(tǒng)一處理，日志打印、兜底處理、監(jiān)控、資源釋放等
        System.err.println("線程[" + thread.getName() + "]異常: " + throwable);});
        return t;
    }
};


//構(gòu)造方法時使用自定義的線程工廠
ExecutorService executor = new ThreadPoolExecutor(
    corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
    threadFactory,
    handler
);

public class MyThreadPoolExecutor extends ThreadPoolExecutor {
    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }
    
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
       //需要特別注意任務是否為submit提交，如果是execute提交的任務，那這里很直接的知道任務是否發(fā)生異常以及后續(xù)去怎么處理
       if(r instanceof Future){
           if(((Future<?>) r).isDone() || ((Future<?>) r).isCancelled()){
               //繼續(xù)使用主線程完成任務,一般不建議，最好使用兜底方式：例如異步發(fā)消息，由后續(xù)的消費組統(tǒng)一處理異常的任務
           }
       }else if( t != null){
            //execute異常處理
       }
    }
}


//FutureTask 把run方法進行了重寫，并且catch住了異常，所以說afterExecute的t 如果是submit提交的方式
//那么t基本上就是null
public void run() {
    //....
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
                set(result);
        }
    } finally {
     //...
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import org.slf4j.*;


public class RobustThreadPool extends ThreadPoolExecutor {
    private static final Logger logger = LoggerFactory.getLogger(RobustThreadPool.class);
    private final AtomicLong failureCounter = new AtomicLong();
    private final RetryPolicy retryPolicy; // 重試策略
    private final ThreadLocal<Long> startTime = new ThreadLocal<>();
    
    public RobustThreadPool(int corePoolSize, int maxPoolSize, 
                          BlockingQueue<Runnable> workQueue,
                          RetryPolicy retryPolicy) {
        super(corePoolSize, maxPoolSize, 60L, TimeUnit.SECONDS, workQueue);
        this.retryPolicy = retryPolicy;
    }
    
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        logger.debug("開始執(zhí)行任務: {}", r);
    }
    
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        // 1. 異常分類處理
         if(r instanceof Future){
           if(((Future<?>) r).isDone()){
             //錯誤記錄以及異常處理
            failureCounter.incrementAndGet();
            handleFailure(r, t, costTime);
           }
       }else if( t != null){
            //execute異常處理
            failureCounter.incrementAndGet();
            handleFailure(r, t, costTime);
       }
        // 2. 資源清理
        cleanThreadLocals();
    }
    
    private void handleFailure(Runnable r, Throwable t) {
        // 1. 異常類型識別
        if (t instanceof OutOfMemoryError) {
            logger.error("JVM內(nèi)存不足，終止任務: {}", t.getMessage());
            System.exit(1); // 嚴重錯誤直接終止
        } 
        // 2. 可重試異常處理
        else if (isRetryable(t)) {
            int retryCount = retryPolicy.getCurrentRetryCount(r);
            if (retryCount < retryPolicy.getMaxRetries()) {
                logger.warn("任務第{}次失敗，準備重試...", 
                    retryCount + 1, t);
                retryPolicy.retry(r, this);
            } else {
                logger.error("任務超過最大重試次數(shù)({})，轉(zhuǎn)入死信隊列", 
                    retryPolicy.getMaxRetries(), t);
                DeadLetterQueue.add(r, t);
            }
        } 
        // 3. 不可重試異常
        else {
            logger.error("不可恢復任務失敗", t);
            Metrics.recordFailure(t.getClass()); // 上報監(jiān)控
        }
    }
    
    private boolean isRetryable(Throwable t) {
        return t instanceof IOException || 
               t instanceof TimeoutException ||
               (t.getCause() != null && isRetryable(t.getCause()));
    }
    
    private void cleanThreadLocals() {
        // 清理可能的內(nèi)存泄漏
        try {
            ThreadLocal<?>[] holders = { /* 其他ThreadLocal */};
            for (ThreadLocal<?> holder : holders) {
                holder.remove();
            }
        } catch (Exception e) {
            logger.warn("清理ThreadLocal失敗", e);
        }
    }
    
    // 重試策略嵌套類
    public static class RetryPolicy {
        private final int maxRetries;
        private final long retryDelayMs;
        private final Map<Runnable, AtomicInteger> retryMap = new ConcurrentHashMap<>();
        
        public RetryPolicy(int maxRetries, long retryDelayMs) {
            this.maxRetries = maxRetries;
            this.retryDelayMs = retryDelayMs;
        }
        
        public void retry(Runnable task, Executor executor) {
            retryMap.computeIfAbsent(task, k -> new AtomicInteger()).incrementAndGet();
            if (retryDelayMs > 0) {
                executor.execute(() -> {
                    try {
                        Thread.sleep(retryDelayMs);
                    } catch (InterruptedException ignored) {}
                    executor.execute(task);
                });
            } else {
                executor.execute(task);
            }
        }
        
        public int getCurrentRetryCount(Runnable task) {
            return retryMap.getOrDefault(task, new AtomicInteger()).get();
        }
        
        public int getMaxRetries() {
            return maxRetries;
        }
    }
}

//添加任務，當不滿足條件時會執(zhí)行拒絕方法reject
public void execute(Runnable command) {
   //...
    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);
    }
    else if (!addWorker(command, false))
        reject(command);
}


//這里就是拒絕的入口。handler是有構(gòu)造方法傳入
final void reject(Runnable command) {
    handler.rejectedExecution(command, this);
}


public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    //....
    //指定拒絕策略
    this.handler = handler;
}

public static class AbortPolicy implements RejectedExecutionHandler {


     //直接拋出RejectedExecutionException
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    //直接在提交任務的線程中執(zhí)行任務
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

public static class DiscardPolicy implements RejectedExecutionHandler {
   //空實現(xiàn)
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    //丟棄隊列中最舊的任務，重試當前任務
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

/**
 * 帶監(jiān)控統(tǒng)計的拒絕策略處理器
 */
public class MetricsRejectedExecutionHandler implements RejectedExecutionHandler {
    private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MetricsRejectedExecutionHandler.class);


    // 統(tǒng)計被拒絕的任務數(shù)量
    private final AtomicLong rejectedCount = new AtomicLong(0);


    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 1. 采集線程池關(guān)鍵指標
        int poolSize = executor.getPoolSize();
        int activeThreads = executor.getActiveCount();
        int corePoolSize = executor.getCorePoolSize();
        int maxPoolSize = executor.getMaximumPoolSize();
        int queueSize = executor.getQueue().size();
        long completedTasks = executor.getCompletedTaskCount();


        // 2. 遞增拒絕計數(shù)器
        long totalRejected = rejectedCount.incrementAndGet();


        // 3. 輸出警告日志（包含完整指標）
        logger.warn("""
             任務被拒絕執(zhí)行！線程池狀態(tài):
            |- 活躍線程數(shù)/當前線程數(shù): {}/{}
            |- 核心/最大線程數(shù): {}/{}
            |- 隊列大小: {}
            |- 已完成任務數(shù): {}
            |- 歷史拒絕總數(shù): {}
            |- 被拒絕任務: {}
            """, 
            activeThreads, poolSize,
            corePoolSize, maxPoolSize,
            queueSize,
            completedTasks,
            totalRejected,
            r.getClass().getName());


        // 4. 可選：降級處理（如存入數(shù)據(jù)庫等待重試）
        // fallbackToDatabase(r);


        // 5. 拋出RejectedExecutionException（保持默認行為）
        throw new RejectedExecutionException("Task " + r.toString() + " rejected");
    }
    
    // 獲取累計拒絕次數(shù)（用于監(jiān)控）
    public long getRejectedCount() {
        return rejectedCount.get();
    }
}

// 創(chuàng)建CPU密集型任務線程池（線程數(shù)=CPU核心數(shù)）
ExecutorService cpuIntensiveExecutor = new ThreadPoolExecutor(
    Runtime.getRuntime().availableProcessors(), // 核心線程數(shù)=CPU核心數(shù)
    Runtime.getRuntime().availableProcessors(), // 最大線程數(shù)=CPU核心數(shù)
    30L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(500),
    new ThreadFactoryBuilder()
        .setNameFormat("cpu-pool-%d")
        .setPriority(Thread.MAX_PRIORITY) // 提高優(yōu)先級
        .build(),
    new ThreadPoolExecutor.AbortPolicy() // 直接拒絕
);


// 使用示例
CompletableFuture.supplyAsync(() -> {
    // 矩陣計算等CPU密集型任務
    double[][] result = matrixMultiply(largeMatrixA, largeMatrixB);
    return result;
}, cpuIntensiveExecutor)
.thenAccept(result -> {
    System.out.println("計算結(jié)果維度: " + result.length + "x" + result[0].length);
});