Java線程池實現(xiàn)原理是什么及怎么使用

這篇文章主要講解了“Java線程池實現(xiàn)原理是什么及怎么使用”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“Java線程池實現(xiàn)原理是什么及怎么使用”吧！

1. 為什么要使用線程池

使用線程池通常由以下兩個原因：

• 頻繁創(chuàng)建銷毀線程需要消耗系統(tǒng)資源，使用線程池可以復(fù)用線程。

• 使用線程池可以更容易管理線程，線程池可以動態(tài)管理線程個數(shù)、具有阻塞隊列、定時周期執(zhí)行任務(wù)、環(huán)境隔離等。

2. 線程池的使用

/**
 * @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個任務(wù)
        for (int i = 0; i < 3; i++) {
            threadPoolExecutor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + " 關(guān)注公眾號：一燈架構(gòu)");
            });
        }
      
        // 3. 關(guān)閉線程池
        threadPoolExecutor.shutdown();
    }
}

線程池的使用非常簡單：

• 調(diào)用new ThreadPoolExecutor()構(gòu)造方法，指定核心參數(shù)，創(chuàng)建線程池。

• 調(diào)用execute()方法提交Runnable任務(wù)

• 使用結(jié)束后，調(diào)用shutdown()方法，關(guān)閉線程池。

再看一下線程池構(gòu)造方法中核心參數(shù)的作用。

3. 線程池核心參數(shù)

線程池共有七大核心參數(shù)：

1.corePoolSize 核心線程數(shù)

當(dāng)往線程池中提交任務(wù)，會創(chuàng)建線程去處理任務(wù)，直到線程數(shù)達(dá)到corePoolSize，才會往阻塞隊列中添加任務(wù)。默認(rèn)情況下，空閑的核心線程并不會被回收，除非配置了allowCoreThreadTimeOut=true。

2.maximumPoolSize 最大線程數(shù)

當(dāng)線程池中的線程數(shù)達(dá)到corePoolSize，阻塞隊列又滿了之后，才會繼續(xù)創(chuàng)建線程，直到達(dá)到maximumPoolSize，另外空閑的非核心線程會被回收。

3.keepAliveTime 線程存活時間

非核心線程的空閑時間達(dá)到了keepAliveTime，將會被回收。

4.TimeUnit 時間單位

線程存活時間的單位，默認(rèn)是TimeUnit.MILLISECONDS（毫秒），可選擇的有：

• TimeUnit.NANOSECONDS（納秒）

• TimeUnit.MICROSECONDS（微秒）

• TimeUnit.MILLISECONDS（毫秒）

• TimeUnit.SECONDS（秒）

• TimeUnit.MINUTES（分鐘）

• TimeUnit.HOURS（小時）

• TimeUnit.DAYS（天）

5.workQueue 阻塞隊列

當(dāng)線程池中的線程數(shù)達(dá)到corePoolSize，再提交的任務(wù)就會放到阻塞隊列的等待，默認(rèn)使用的是LinkedBlockingQueue，可選擇的有：

• LinkedBlockingQueue（基于鏈表實現(xiàn)的阻塞隊列）

• ArrayBlockingQueue（基于數(shù)組實現(xiàn)的阻塞隊列）

• SynchronousQueue（只有一個元素的阻塞隊列）

• PriorityBlockingQueue（實現(xiàn)了優(yōu)先級的阻塞隊列）

• DelayQueue（實現(xiàn)了延遲功能的阻塞隊列）

6.threadFactory 線程創(chuàng)建工廠

用來創(chuàng)建線程的工廠，默認(rèn)的是Executors.defaultThreadFactory()，可選擇的還有Executors.privilegedThreadFactory()實現(xiàn)了線程優(yōu)先級。當(dāng)然也可以自定義線程創(chuàng)建工廠，創(chuàng)建線程的時候最好指定線程名稱，便于排查問題。

7.RejectedExecutionHandler 拒絕策略

當(dāng)線程池中的線程數(shù)達(dá)到maximumPoolSize，阻塞隊列也滿了之后，再往線程池中提交任務(wù)，就會觸發(fā)執(zhí)行拒絕策略，默認(rèn)的是AbortPolicy（直接終止，拋出異常），可選擇的有：

• AbortPolicy（直接終止，拋出異常）

• DiscardPolicy（默默丟棄，不拋出異常）

• DiscardOldestPolicy（丟棄隊列中最舊的任務(wù)，執(zhí)行當(dāng)前任務(wù)）

• CallerRunsPolicy（返回給調(diào)用者執(zhí)行）

4. 線程池工作原理

線程池的工作原理，簡單理解如下：

• 當(dāng)往線程池中提交任務(wù)的時候，會先判斷線程池中線程數(shù)是否核心線程數(shù)，如果小于，會創(chuàng)建核心線程并執(zhí)行任務(wù)。

• 如果線程數(shù)大于核心線程數(shù)，會判斷阻塞隊列是否已滿，如果沒有滿，會把任務(wù)添加到阻塞隊列中等待調(diào)度執(zhí)行。

• 如果阻塞隊列已滿，會判斷線程數(shù)是否小于最大線程數(shù)，如果小于，會繼續(xù)創(chuàng)建最大線程數(shù)并執(zhí)行任務(wù)。

• 如果線程數(shù)大于最大線程數(shù)，會執(zhí)行拒絕策略，然后結(jié)束。

5. 線程池源碼剖析

5.1 線程池的屬性

public class ThreadPoolExecutor extends AbstractExecutorService {

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

    // 獨占鎖，用來控制多線程下的并發(fā)操作
    private final ReentrantLock mainLock = new ReentrantLock();
    // 工作線程的集合
    private final HashSet<Worker> workers = new HashSet<>();
    // 等待條件，用來響應(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;

}

線程池的控制狀態(tài)ctl用來存儲線程池狀態(tài)和線程個數(shù)，前3位用來存儲線程池狀態(tài)，后29位用來存儲線程數(shù)量。

設(shè)計者多聰明，用一個變量存儲了兩塊內(nèi)容。

5.2 線程池狀態(tài)

線程池共有5種狀態(tài)：

5.3 execute源碼

看一下往線程池中提交任務(wù)的源碼，這是線程池的核心邏輯：

// 往線程池中提交任務(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，并且可以添加到阻塞隊列
    if (isRunning(c) && workQueue.offer(command)) {
        // 5. 重新檢查線程池狀態(tài)，如果線程池不是運行狀態(tài)，就移除剛才添加的任務(wù)，并執(zhí)行拒絕策略
        int recheck = ctl.get();
        if (!isRunning(recheck) && remove(command))
            reject(command);
        // 6. 判斷如果線程數(shù)是0，就創(chuàng)建非核心線程（任務(wù)是null，會從阻塞隊列中拉取任務(wù)）
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 7. 如果添加阻塞隊列失敗，就創(chuàng)建一個Worker
    else if (!addWorker(command, false))
        // 8. 如果創(chuàng)建Worker失敗說明已經(jīng)達(dá)到最大線程數(shù)了，則執(zhí)行拒絕策略
        reject(command);
}

execute方法的邏輯也很簡單，最終就是調(diào)用addWorker方法，把任務(wù)添加到worker集合中，再看一下addWorker方法的源碼：

// 添加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. 校驗線程數(shù)是否超過容量限制
            if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 4. 使用CAS修改線程數(shù)
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();
            // 5. 如果線程池狀態(tài)變了，則從頭再來
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 6. 把任務(wù)和新線程包裝成一個worker
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 7. 加鎖，控制并發(fā)
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 8. 再次校驗線程池狀態(tài)是否異常
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                    // 9. 如果線程已經(jī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. 啟動線程
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (!workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

方法雖然很長，但是邏輯很清晰。就是把任務(wù)和線程包裝成worker，添加到worker集合，并啟動線程。

5.4 worker源碼

再看一下worker類的結(jié)構(gòu)：

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

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

5.5 runWorker源碼

再看一下run方法的源碼：

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

// 線程運行核心方法
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，就從阻塞隊列中獲取任務(wù)
        while (task != null || (task = getTask()) != null) {
            // 加鎖，保證thread不被其他線程中斷（除非線程池被中斷）
            w.lock();
            // 2. 校驗線程池狀態(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);
    }
}

runWorker方法邏輯也很簡單，就是不斷從阻塞隊列中拉取任務(wù)并執(zhí)行。

再看一下從阻塞隊列中拉取任務(wù)的邏輯：

// 從阻塞隊列中拉取任務(wù)
private Runnable getTask() {
    boolean timedOut = false;
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 1. 如果線程池已經(jīng)停了，或者阻塞隊列是空，就回收當(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. 從阻塞隊列中拉取任務(wù)
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

感謝各位的閱讀，以上就是“Java線程池實現(xiàn)原理是什么及怎么使用”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對Java線程池實現(xiàn)原理是什么及怎么使用這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關(guān)知識點的文章，歡迎關(guān)注！