ThreadPoolExecutor引數含義及原始碼執行流程詳解

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        // maximumPoolSize 必須大於 0，且必須大於 corePoolSize
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    // Executors.defaultThreadFactory() 為預設的執行緒建立工廠
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}
public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}
// 預設的執行緒建立工廠，需要實現 ThreadFactory 介面
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;
    }
}

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    // 當前工作的執行緒數小於核心執行緒數
    if (workerCountOf(c) < corePoolSize) {
        // 建立新的執行緒執行此任務
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 檢查執行緒池是否處於執行狀態，如果是則把任務新增到佇列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 再出檢查執行緒池是否處於執行狀態，防止在第一次校驗通過後執行緒池關閉
        // 如果是非執行狀態，則將剛加入佇列的任務移除
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 如果執行緒池的執行緒數為 0 時（當 corePoolSize 設定為 0 時會發生）
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false); // 新建執行緒執行任務
    }
    // 核心執行緒都在忙且佇列都已爆滿，嘗試新啟動一個執行緒執行失敗
    else if (!addWorker(command, false)) 
        // 執行拒絕策略
        reject(command);
}

ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 10L,
        TimeUnit.SECONDS, new LinkedBlockingQueue(20));
// execute 使用
executor.execute(new Runnable() {
    @Override
    public void run() {
        System.out.println("Hello, execute.");
    }
});
// submit 使用
Future<String> future = executor.submit(new Callable<String>() {
    @Override
    public String call() throws Exception {
        System.out.println("Hello, submit.");
        return "Success";
    }
});
System.out.println(future.get());

ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 3, 10,
        TimeUnit.SECONDS, new LinkedBlockingQueue<>(2),
        new ThreadPoolExecutor.AbortPolicy()); // 新增 AbortPolicy 拒絕策略
for (int i = 0; i < 6; i++) {
    executor.execute(() -> {
        System.out.println(Thread.currentThread().getName());
    });
}

ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 3, 10,
        TimeUnit.SECONDS, new LinkedBlockingQueue<>(2),
        new RejectedExecutionHandler() {  // 新增自定義拒絕策略
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                // 業務處理方法
                System.out.println("執行自定義拒絕策略");
            }
        });
for (int i = 0; i < 6; i++) {
    executor.execute(() -> {
        System.out.println(Thread.currentThread().getName());
    });
}

public class ThreadPoolExtend {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 執行緒池擴充套件呼叫
        MyThreadPoolExecutor executor = new MyThreadPoolExecutor(2, 4, 10,
                TimeUnit.SECONDS, new LinkedBlockingQueue());
        for (int i = 0; i < 3; i++) {
            executor.execute(() -> {
                Thread.currentThread().getName();
            });
        }
    }
   /**
     * 執行緒池擴充套件
     */
    static class MyThreadPoolExecutor extends ThreadPoolExecutor {
        // 儲存執行緒執行開始時間
        private final ThreadLocal<Long> localTime = new ThreadLocal<>();
        public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                            TimeUnit unit, BlockingQueue<Runnable> workQueue) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
        /**
         * 開始執行之前
         * @param t 執行緒
         * @param r 任務
         */
        @Override
        protected void beforeExecute(Thread t, Runnable r) {
            Long sTime = System.nanoTime(); // 開始時間 (單位：納秒)
            localTime.set(sTime);
            System.out.println(String.format("%s | before | time=%s",
                    t.getName(), sTime));
            super.beforeExecute(t, r);
        }
        /**
         * 執行完成之後
         * @param r 任務
         * @param t 丟擲的異常
         */
        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            Long eTime = System.nanoTime(); // 結束時間 (單位：納秒)
            Long totalTime = eTime - localTime.get(); // 執行總時間
            System.out.println(String.format("%s | after | time=%s | 耗時：%s 毫秒",
                    Thread.currentThread().getName(), eTime, (totalTime / 1000000.0)));
            super.afterExecute(r, t);
        }
    }
}