/**
 * @author 一燈架構(gòu)
 * @apiNote 線程池示例
 **/
public class ThreadPoolDemo {

    public static void main(String[] args) {
        // 1. 創(chuàng)建線程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                3,
                3,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
      
        // 2. 往線程池中提交3個(gè)任務(wù)
        for (int i = 0; i < 3; i++) {
            threadPoolExecutor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + " 關(guān)注公眾號(hào)：一燈架構(gòu)");
            });
        }
      
        // 3. 關(guān)閉線程池
        threadPoolExecutor.shutdown();
    }
}

public class ThreadPoolExecutor extends AbstractExecutorService {

    // 線程池的控制狀態(tài)，Integer長(zhǎng)度是32位，前3位用來(lái)存儲(chǔ)線程池狀態(tài)，后29位用來(lái)存儲(chǔ)線程數(shù)量
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // 線程個(gè)數(shù)所占的位數(shù)
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 線程池的最大容量，2^29-1，約5億個(gè)線程
    private static final int CAPACITY = (1 << COUNT_BITS) - 1;

    // 獨(dú)占鎖，用來(lái)控制多線程下的并發(fā)操作
    private final ReentrantLock mainLock = new ReentrantLock();
    // 工作線程的集合
    private final HashSet<Worker> workers = new HashSet<>();
    // 等待條件，用來(lái)響應(yīng)中斷
    private final Condition termination = mainLock.newCondition();
    // 是否允許回收核心線程
    private volatile boolean allowCoreThreadTimeOut;
    // 線程數(shù)的歷史峰值
    private int largestPoolSize;

    /**
     * 以下是線程池的七大核心參數(shù)
     */
    private volatile int corePoolSize;
    private volatile int maximumPoolSize;
    private volatile long keepAliveTime;
    private final BlockingQueue<Runnable> workQueue;
    private volatile ThreadFactory threadFactory;
    private volatile RejectedExecutionHandler handler;

}

// 往線程池中提交任務(wù)
public void execute(Runnable command) {
    // 1. 判斷提交的任務(wù)是否為null
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    // 2. 判斷線程數(shù)是否小于核心線程數(shù)
    if (workerCountOf(c) < corePoolSize) {
        // 3. 把任務(wù)包裝成worker，添加到worker集合中
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 4. 判斷如果線程數(shù)不小于corePoolSize，并且可以添加到阻塞隊(duì)列
    if (isRunning(c) && workQueue.offer(command)) {
        // 5. 重新檢查線程池狀態(tài)，如果線程池不是運(yùn)行狀態(tài)，就移除剛才添加的任務(wù)，并執(zhí)行拒絕策略
        int recheck = ctl.get();
        if (!isRunning(recheck) && remove(command))
            reject(command);
        // 6. 判斷如果線程數(shù)是0，就創(chuàng)建非核心線程（任務(wù)是null，會(huì)從阻塞隊(duì)列中拉取任務(wù)）
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 7. 如果添加阻塞隊(duì)列失敗，就創(chuàng)建一個(gè)Worker
    else if (!addWorker(command, false))
        // 8. 如果創(chuàng)建Worker失敗說(shuō)明已經(jīng)達(dá)到最大線程數(shù)了，則執(zhí)行拒絕策略
        reject(command);
}

// 添加worker
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 1. 檢查是否允許提交任務(wù)
        if (rs >= SHUTDOWN &&
                !(rs == SHUTDOWN &&
                        firstTask == null &&
                        !workQueue.isEmpty()))
            return false;
        // 2. 使用死循環(huán)保證添加線程成功
        for (; ; ) {
            int wc = workerCountOf(c);
            // 3. 校驗(yàn)線程數(shù)是否超過(guò)容量限制
            if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 4. 使用CAS修改線程數(shù)
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();
            // 5. 如果線程池狀態(tài)變了，則從頭再來(lái)
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 6. 把任務(wù)和新線程包裝成一個(gè)worker
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 7. 加鎖，控制并發(fā)
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 8. 再次校驗(yàn)線程池狀態(tài)是否異常
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                    // 9. 如果線程已經(jīng)啟動(dòng)，就拋出異常
                    if (t.isAlive())
                        throw new IllegalThreadStateException();
                    // 10. 添加到worker集合中
                    workers.add(w);
                    int s = workers.size();
                    // 11. 記錄線程數(shù)歷史峰值
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                // 12. 啟動(dòng)線程
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (!workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable {
    // 工作線程
    final Thread thread;
    // 任務(wù)
    Runnable firstTask;

    // 創(chuàng)建worker，并創(chuàng)建一個(gè)新線程（用來(lái)執(zhí)行任務(wù)）
    Worker(Runnable firstTask) {
        setState(-1);
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
}

// 線程執(zhí)行入口
public void run() {
    runWorker(this);
}

// 線程運(yùn)行核心方法
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock();
    boolean completedAbruptly = true;
    try {
        // 1. 如果當(dāng)前worker中任務(wù)是null，就從阻塞隊(duì)列中獲取任務(wù)
        while (task != null || (task = getTask()) != null) {
            // 加鎖，保證thread不被其他線程中斷（除非線程池被中斷）
            w.lock();
            // 2. 校驗(yàn)線程池狀態(tài)，是否需要中斷當(dāng)前線程
            if ((runStateAtLeast(ctl.get(), STOP) ||
                    (Thread.interrupted() &&
                            runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 3. 執(zhí)行run方法
                    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 {
        // 4. 從worker集合刪除當(dāng)前worker
        processWorkerExit(w, completedAbruptly);
    }
}

// 從阻塞隊(duì)列中拉取任務(wù)
private Runnable getTask() {
    boolean timedOut = false;
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 1. 如果線程池已經(jīng)停了，或者阻塞隊(duì)列是空，就回收當(dāng)前線程
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // 2. 再次判斷是否需要回收線程
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            // 3. 從阻塞隊(duì)列中拉取任務(wù)
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}