快捷導(dǎo)航

10分鐘帶你徒手寫個(gè)Java線程池

更新時(shí)間：2023年04月20日 08:43:56 作者：冰?河

我們自己手動(dòng)實(shí)現(xiàn)的線程池要比Java自身的線程池簡單的多，我們?nèi)サ袅烁鞣N復(fù)雜的處理方式，只保留了最核心的原理，感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

Java線程池核心原理

看過Java線程池源碼的小伙伴都知道，在Java線程池中最核心的類就是ThreadPoolExecutor，而在ThreadPoolExecutor類中最核心的構(gòu)造方法就是帶有7個(gè)參數(shù)的構(gòu)造方法，如下所示。

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

各參數(shù)的含義如下所示。

corePoolSize：線程池中的常駐核心線程數(shù)。
maximumPoolSize：線程池能夠容納同時(shí)執(zhí)行的最大線程數(shù)，此值大于等于1。
keepAliveTime：多余的空閑線程存活時(shí)間，當(dāng)空間時(shí)間達(dá)到keepAliveTime值時(shí)，多余的線程會(huì)被銷毀直到只剩下corePoolSize個(gè)線程為止。
unit：keepAliveTime的單位。
workQueue：任務(wù)隊(duì)列，被提交但尚未被執(zhí)行的任務(wù)。
threadFactory：表示生成線程池中工作線程的線程工廠，用戶創(chuàng)建新線程，一般用默認(rèn)即可。
handler：拒絕策略，表示當(dāng)線程隊(duì)列滿了并且工作線程大于等于線程池的最大顯示數(shù)(maxnumPoolSize)時(shí)，如何來拒絕請(qǐng)求執(zhí)行的runnable的策略。

并且Java的線程池是通過 生產(chǎn)者-消費(fèi)者模式 實(shí)現(xiàn)的，線程池的使用方是生產(chǎn)者，而線程池本身就是消費(fèi)者。

Java線程池的核心工作流程如下圖所示。

手?jǐn)]Java線程池

我們自己手動(dòng)實(shí)現(xiàn)的線程池要比Java自身的線程池簡單的多，我們?nèi)サ袅烁鞣N復(fù)雜的處理方式，只保留了最核心的原理：線程池的使用者向任務(wù)隊(duì)列中添加任務(wù)，而線程池本身從任務(wù)隊(duì)列中消費(fèi)任務(wù)并執(zhí)行任務(wù)。

只要理解了這個(gè)核心原理，接下來的代碼就簡單多了。在實(shí)現(xiàn)這個(gè)簡單的線程池時(shí)，我們可以將整個(gè)實(shí)現(xiàn)過程進(jìn)行拆解。拆解后的實(shí)現(xiàn)流程為：定義核心字段、創(chuàng)建內(nèi)部類WorkThread、創(chuàng)建ThreadPool類的構(gòu)造方法和創(chuàng)建執(zhí)行任務(wù)的方法。

定義核心字段

首先，我們創(chuàng)建一個(gè)名稱為ThreadPool的Java類，并在這個(gè)類中定義如下核心字段。

DEFAULT_WORKQUEUE_SIZE：靜態(tài)常量，表示默認(rèn)的阻塞隊(duì)列大小。
workQueue：模擬實(shí)際的線程池使用阻塞隊(duì)列來實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模式。
workThreads：模擬實(shí)際的線程池使用List集合保存線程池內(nèi)部的工作線程。

核心代碼如下所示。

//默認(rèn)阻塞隊(duì)列大小
private static final int DEFAULT_WORKQUEUE_SIZE = 5;

//模擬實(shí)際的線程池使用阻塞隊(duì)列來實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模式
private BlockingQueue<Runnable> workQueue;

//模擬實(shí)際的線程池使用List集合保存線程池內(nèi)部的工作線程
private List<WorkThread> workThreads = new ArrayList<WorkThread>();

創(chuàng)建內(nèi)部類WordThread

在ThreadPool類中創(chuàng)建一個(gè)內(nèi)部類WorkThread，模擬線程池中的工作線程。主要的作用就是消費(fèi)workQueue中的任務(wù)，并執(zhí)行任務(wù)。由于工作線程需要不斷從workQueue中獲取任務(wù)，所以，這里使用了while(true)循環(huán)不斷嘗試消費(fèi)隊(duì)列中的任務(wù)。

核心代碼如下所示。

//內(nèi)部類WorkThread，模擬線程池中的工作線程
//主要的作用就是消費(fèi)workQueue中的任務(wù)，并執(zhí)行
//由于工作線程需要不斷從workQueue中獲取任務(wù)，使用了while(true)循環(huán)不斷嘗試消費(fèi)隊(duì)列中的任務(wù)
class WorkThread extends Thread{
    @Override
    public void run() {
        //不斷循環(huán)獲取隊(duì)列中的任務(wù)
        while (true){
            //當(dāng)沒有任務(wù)時(shí)，會(huì)阻塞
            try {
                Runnable workTask = workQueue.take();
                workTask.run();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

創(chuàng)建ThreadPool類的構(gòu)造方法

這里，我們?yōu)門hreadPool類創(chuàng)建兩個(gè)構(gòu)造方法，一個(gè)構(gòu)造方法中傳入線程池的容量大小和阻塞隊(duì)列，另一個(gè)構(gòu)造方法中只傳入線程池的容量大小。

核心代碼如下所示。

//在ThreadPool的構(gòu)造方法中傳入線程池的大小和阻塞隊(duì)列
public ThreadPool(int poolSize, BlockingQueue<Runnable> workQueue){
    this.workQueue = workQueue;
    //創(chuàng)建poolSize個(gè)工作線程并將其加入到workThreads集合中
    IntStream.range(0, poolSize).forEach((i) -> {
        WorkThread workThread = new WorkThread();
        workThread.start();
        workThreads.add(workThread);
    });
}

//在ThreadPool的構(gòu)造方法中傳入線程池的大小
public ThreadPool(int poolSize){
    this(poolSize, new LinkedBlockingQueue<>(DEFAULT_WORKQUEUE_SIZE));
}

創(chuàng)建執(zhí)行任務(wù)的方法

在ThreadPool類中創(chuàng)建執(zhí)行任務(wù)的方法execute()，execute()方法的實(shí)現(xiàn)比較簡單，就是將方法接收到的Runnable任務(wù)加入到workQueue隊(duì)列中。

核心代碼如下所示。

//通過線程池執(zhí)行任務(wù)
public void execute(Runnable task){
    try {
        workQueue.put(task);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

完整源碼

這里，我們給出手動(dòng)實(shí)現(xiàn)的ThreadPool線程池的完整源代碼，如下所示。

package io.binghe.thread.pool;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.IntStream;

/**
 * @author binghe
 * @version 1.0.0
 * @description 自定義線程池
 */
public class ThreadPool {

    //默認(rèn)阻塞隊(duì)列大小
    private static final int DEFAULT_WORKQUEUE_SIZE = 5;

    //模擬實(shí)際的線程池使用阻塞隊(duì)列來實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模式
    private BlockingQueue<Runnable> workQueue;

    //模擬實(shí)際的線程池使用List集合保存線程池內(nèi)部的工作線程
    private List<WorkThread> workThreads = new ArrayList<WorkThread>();

    //在ThreadPool的構(gòu)造方法中傳入線程池的大小和阻塞隊(duì)列
    public ThreadPool(int poolSize, BlockingQueue<Runnable> workQueue){
        this.workQueue = workQueue;
        //創(chuàng)建poolSize個(gè)工作線程并將其加入到workThreads集合中
        IntStream.range(0, poolSize).forEach((i) -> {
            WorkThread workThread = new WorkThread();
            workThread.start();
            workThreads.add(workThread);
        });
    }

    //在ThreadPool的構(gòu)造方法中傳入線程池的大小
    public ThreadPool(int poolSize){
        this(poolSize, new LinkedBlockingQueue<>(DEFAULT_WORKQUEUE_SIZE));
    }

 //通過線程池執(zhí)行任務(wù)
    public void execute(Runnable task){
        try {
            workQueue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //內(nèi)部類WorkThread，模擬線程池中的工作線程
    //主要的作用就是消費(fèi)workQueue中的任務(wù)，并執(zhí)行
    //由于工作線程需要不斷從workQueue中獲取任務(wù)，使用了while(true)循環(huán)不斷嘗試消費(fèi)隊(duì)列中的任務(wù)
    class WorkThread extends Thread{
        @Override
        public void run() {
            //不斷循環(huán)獲取隊(duì)列中的任務(wù)
            while (true){
                //當(dāng)沒有任務(wù)時(shí)，會(huì)阻塞
                try {
                    Runnable workTask = workQueue.take();
                    workTask.run();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

沒錯(cuò)，我們僅僅用了幾十行Java代碼就實(shí)現(xiàn)了一個(gè)極簡版的Java線程池，沒錯(cuò)，這個(gè)極簡版的Java線程池的代碼卻體現(xiàn)了Java線程池的核心原理。

接下來，我們測試下這個(gè)極簡版的Java線程池。

編寫測試程序

測試程序也比較簡單，就是通過在main()方法中調(diào)用ThreadPool類的構(gòu)造方法，傳入線程池的大小，創(chuàng)建一個(gè)ThreadPool類的實(shí)例，然后循環(huán)10次調(diào)用ThreadPool類的execute()方法，向線程池中提交的任務(wù)為：打印當(dāng)前線程的名稱--->> Hello ThreadPool。

整體測試代碼如下所示。

package io.binghe.thread.pool.test;

import io.binghe.thread.pool.ThreadPool;

import java.util.stream.IntStream;

/**
 * @author binghe
 * @version 1.0.0
 * @description 測試自定義線程池
 */
public class ThreadPoolTest {

    public static void main(String[] args){
        ThreadPool threadPool = new ThreadPool(10);
        IntStream.range(0, 10).forEach((i) -> {
            threadPool.execute(() -> {
                System.out.println(Thread.currentThread().getName() + "--->> Hello ThreadPool");
            });
        });
    }
}

接下來，運(yùn)行ThreadPoolTest類的main()方法，會(huì)輸出如下信息。

Thread-0--->> Hello ThreadPool
Thread-9--->> Hello ThreadPool
Thread-5--->> Hello ThreadPool
Thread-8--->> Hello ThreadPool
Thread-4--->> Hello ThreadPool
Thread-1--->> Hello ThreadPool
Thread-2--->> Hello ThreadPool
Thread-5--->> Hello ThreadPool
Thread-9--->> Hello ThreadPool
Thread-0--->> Hello ThreadPool

至此，我們自定義的Java線程池就開發(fā)完成了。