一起學并發(fā)編程 - 簡易線程池實現(xiàn)

Harriet666 發(fā)布于2019-08-15 12:09 / 1104人閱讀

摘要：并且，線程池在某些情況下還能動態(tài)調(diào)整工作線程的數(shù)量，以平衡資源消耗和工作效率。同時線程池還提供了對池中工作線程進行統(tǒng)一的管理的相關(guān)方法。

開發(fā)中經(jīng)常會遇到各種池（如：連接池，線程池），它們的作用就是為了提高性能及減少開銷，在JDK1.5以后的java.util.concurrent包中內(nèi)置了很多不同使用場景的線程池，為了更好的理解它們，自己手寫一個線程池，加深印象。

概述

1.什么是池

它的基本思想是一種對象池，程序初始化的時候開辟一塊內(nèi)存空間，里面存放若干個線程對象，池中線程執(zhí)行調(diào)度由池管理器來處理。當有線程任務時，從池中取一個，執(zhí)行完成后線程對象歸池，這樣可以避免反復創(chuàng)建線程對象所帶來的性能開銷，節(jié)省系統(tǒng)的資源。

2.使用線程池的好處

合理的使用線程池可以重復利用已創(chuàng)建的線程，這樣就可以減少在創(chuàng)建線程和銷毀線程上花費的時間和資源。并且，線程池在某些情況下還能動態(tài)調(diào)整工作線程的數(shù)量，以平衡資源消耗和工作效率。同時線程池還提供了對池中工作線程進行統(tǒng)一的管理的相關(guān)方法。這樣就相當于我們一次創(chuàng)建，就可以多次使用，大量的節(jié)省了系統(tǒng)頻繁的創(chuàng)建和銷毀線程所需要的資源。

簡易版實現(xiàn)

包含功能：

1.創(chuàng)建線程池，銷毀線程池，添加新任務

2.沒有任務進入等待，有任務則處理掉

3.動態(tài)伸縮，擴容

4.拒絕策略

介紹了線程池的原理以及主要組件之后，就讓我們來手動實現(xiàn)一個自己的線程池，以加深理解和深入學習。因為自己實現(xiàn)的簡易版本所以不建議生產(chǎn)中使用，生產(chǎn)中使用java.util.concurrent會更加健壯和優(yōu)雅（后續(xù)文章會介紹）

代碼

以下線程池相關(guān)代碼均在SimpleThreadPoolExecutor.java中，由于為了便于解讀因此以代碼塊的形式呈現(xiàn)

維護一個內(nèi)部枚舉類，用來標記當前任務線程狀態(tài)，在Thread中其實也有.

private enum TaskState {
    FREE, RUNNABLE, BLOCKED, TERMINATED;
}

定義拒絕策略接口，以及默認實現(xiàn)

static class DiscardException extends RuntimeException {
    private static final long serialVersionUID = 8827362380544575914L;

    DiscardException(String message) {
        super(message);
    }
}

interface DiscardPolicy {//拒絕策略接口
    void discard() throws DiscardException;
}

任務線程具體實現(xiàn)

1.繼承Thread，重寫run方法。

2.this.taskState == TaskState.FREE && TASK_QUEUE.isEmpty() 如果當前線程處于空閑狀態(tài)且沒有任何任務了就將它wait住，讓出CPU執(zhí)行權(quán)

3.如果有任務就去執(zhí)行FIFO（先進先出）策略

4.定義close方法，關(guān)閉線程，當然這里不能暴力關(guān)閉，所以這里有需要借助interrupt

public static class WorkerTask extends Thread {
    // 線程狀態(tài)
    private TaskState taskState;
    // 線程編號
    private static int threadInitNumber;
    /**
     * 生成線程名,參考Thread.nextThreadNum();
     *
     * @return
     */
    private static synchronized String nextThreadName() {
        return THREAD_NAME_PREFIX + (++threadInitNumber);
    }

    WorkerTask() {
        super(THREAD_GROUP, nextThreadName());
    }

    @Override
    public void run() {
        Runnable target;
        //說明該線程處于空閑狀態(tài)
        OUTER:
        while (this.taskState != TaskState.TERMINATED) {
            synchronized (TASK_QUEUE) {
                while (this.taskState == TaskState.FREE && TASK_QUEUE.isEmpty()) {
                    try {
                        this.taskState = TaskState.BLOCKED;//此處標記
                        //沒有任務就wait住,讓出CPU執(zhí)行權(quán)
                        TASK_QUEUE.wait();
                        //如果被打斷說明當前線程執(zhí)行了 shutdown() 方法  線程狀態(tài)為 TERMINATED 直接跳到 while 便于退出
                    } catch (InterruptedException e) {
                        break OUTER;
                    }
                }
                target = TASK_QUEUE.removeFirst();//遵循FIFO策略
            }
            if (target != null) {
                this.taskState = TaskState.RUNNABLE;
                target.run();//開始任務了
                this.taskState = TaskState.FREE;
            }
        }
    }

    void close() {//優(yōu)雅關(guān)閉線程
        this.taskState = TaskState.TERMINATED;
        this.interrupt();
    }
}

簡易版線程池，主要就是維護了一個任務隊列和線程集，為了動態(tài)擴容，自己也繼承了Thread去做監(jiān)聽操作，對外提供submit()提交執(zhí)行任務、shutdown()等待所有任務工作完畢，關(guān)閉線程池

public class SimpleThreadPoolExecutor extends Thread {

    // 線程池大小
    private int threadPoolSize;
    // 最大接收任務
    private int queueSize;
    // 拒絕策略
    private DiscardPolicy discardPolicy;
    // 是否被銷毀
    private volatile boolean destroy = false;

    private final static int DEFAULT_MIN_THREAD_SIZE = 2;// 默認最小線程數(shù)
    private final static int DEFAULT_ACTIVE_THREAD_SIZE = 5;// 活躍線程
    private final static int DEFAULT_MAX_THREAD_SIZE = 10;// 最大線程
    private final static int DEFAULT_WORKER_QUEUE_SIZE = 100;// 最多執(zhí)行多少任務
    private final static String THREAD_NAME_PREFIX = "MY-THREAD-NAME-";//線程名前綴
    private final static String THREAD_POOL_NAME = "SIMPLE-POOL";//線程組的名稱
    private final static ThreadGroup THREAD_GROUP = new ThreadGroup(THREAD_POOL_NAME);//線程組
    private final static List WORKER_TASKS = new ArrayList<>();// 線程容器
    // 任務隊列容器,也可以用Queue 遵循 FIFO 規(guī)則
    private final static LinkedList TASK_QUEUE = new LinkedList<>();
    // 拒絕策略
    private final static DiscardPolicy DEFAULT_DISCARD_POLICY = () -> {
        throw new DiscardException("[拒絕執(zhí)行] - [任務隊列溢出...]");
    };

    private int minSize;//最小線程
    private int maxSize;//最大線程
    private int activeSize;//活躍線程

    SimpleThreadPoolExecutor() {
        this(DEFAULT_MIN_THREAD_SIZE, DEFAULT_ACTIVE_THREAD_SIZE, DEFAULT_MAX_THREAD_SIZE, DEFAULT_WORKER_QUEUE_SIZE, DEFAULT_DISCARD_POLICY);
    }

    SimpleThreadPoolExecutor(int minSize, int activeSize, int maxSize, int queueSize, DiscardPolicy discardPolicy){
        this.minSize = minSize;
        this.activeSize = activeSize;
        this.maxSize = maxSize;
        this.queueSize = queueSize;
        this.discardPolicy = discardPolicy;
        initPool();
    }

    void submit(Runnable runnable) {
        if (destroy) {
            throw new IllegalStateException("線程池已銷毀...");
        }
        synchronized (TASK_QUEUE) {
            if (TASK_QUEUE.size() > queueSize) {//如果當前任務隊超出隊列限制,后續(xù)任務拒絕執(zhí)行
                discardPolicy.discard();
            }
            // 1.將任務添加到隊列
            TASK_QUEUE.addLast(runnable);
            // 2.喚醒等待的線程去執(zhí)行任務
            TASK_QUEUE.notifyAll();
        }
    }

    void shutdown() throws InterruptedException {
        int activeCount = THREAD_GROUP.activeCount();
        while (!TASK_QUEUE.isEmpty() && activeCount > 0) {
            // 如果還有任務,那就休息一會
            Thread.sleep(100);
        }
        int intVal = WORKER_TASKS.size();//如果線程池中沒有線程,那就不用關(guān)了
        while (intVal > 0) {
            for (WorkerTask task : WORKER_TASKS) {
                //當任務隊列為空的時候，線程狀態(tài)才會為 BLOCKED ，所以可以打斷掉，相反等任務執(zhí)行完在關(guān)閉
                if (task.taskState == TaskState.BLOCKED) {
                    task.close();
                    intVal--;
                } else {
                    Thread.sleep(50);
                }
            }
        }
        this.destroy = true;
        //資源回收
        TASK_QUEUE.clear();
        WORKER_TASKS.clear();
        this.interrupt();
        System.out.println("線程關(guān)閉");
    }

    private void createWorkerTask() {
        WorkerTask task = new WorkerTask();
        //剛創(chuàng)建出來的線程應該是未使用的
        task.taskState = TaskState.FREE;
        WORKER_TASKS.add(task);
        task.start();
    }

    /**
     * 初始化操作
     */
    private void initPool() {
        for (int i = 0; i < this.minSize; i++) {
            this.createWorkerTask();
        }
        this.threadPoolSize = minSize;
        this.start();//自己啟動自己
    }

    @Override
    public void run() {
        while (!destroy) {
            try {
                Thread.sleep(5_000L);
                if (TASK_QUEUE.size() > activeSize && threadPoolSize < activeSize) { // 第一次擴容到 activeSize 大小
                    for (int i = threadPoolSize; i < activeSize; i++) {
                        createWorkerTask();
                    }
                    this.threadPoolSize = activeSize;
                    System.out.println("[初次擴充] - [" + toString() + "]");
                } else if (TASK_QUEUE.size() > maxSize && threadPoolSize < maxSize) {// 第二次擴容到最大線程
                    System.out.println();
                    for (int i = threadPoolSize; i < maxSize; i++) {
                        createWorkerTask();
                    }
                    this.threadPoolSize = maxSize;
                    System.out.println("[再次擴充] - [" + toString() + "]");
                } else {
                    //防止線程在submit的時候，其他線程獲取到鎖干壞事
                    synchronized (WORKER_TASKS) {
                        int releaseSize = threadPoolSize - activeSize;
                        Iterator iterator = WORKER_TASKS.iterator();// List不允許在for中刪除集合元素,所以這里需要使用迭代器
                        while (iterator.hasNext()) {
                            if (releaseSize <= 0) {
                                break;
                            }
                            WorkerTask task = iterator.next();
                            //不能回收正在運行的線程,只回收空閑線程
                            if (task.taskState == TaskState.FREE) {
                                task.close();
                                iterator.remove();
                                releaseSize--;
                            }
                        }
                        System.out.println("[資源回收] - [" + toString() + "]");
                    }
                    threadPoolSize = activeSize;
                }
            } catch (InterruptedException e) {
                System.out.println("資源釋放");
            }
        }
    }

    @Override
    public String toString() {
        return "SimpleThreadPoolExecutor{" +
                "threadPoolSize=" + threadPoolSize +
                ", taskQueueSize=" + TASK_QUEUE.size() +
                ", minSize=" + minSize +
                ", maxSize=" + maxSize +
                ", activeSize=" + activeSize +
                "}";
    }
}

測試一把

創(chuàng)建一個測試類

public class SimpleExecutorTest {

    public static void main(String[] args) throws InterruptedException {

        SimpleThreadPoolExecutor executor = new SimpleThreadPoolExecutor();
        IntStream.range(0, 30).forEach(i ->
                executor.submit(() -> {
                    System.out.printf("[線程] - [%s] 開始工作...
", Thread.currentThread().getName());
                    try {
                        Thread.sleep(2_000L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.printf("[線程] - [%s] 工作完畢...
", Thread.currentThread().getName());
                })
        );
        //executor.shutdown();如果放開注釋即會執(zhí)行完所有任務關(guān)閉線程池
    }
}

日志分析： 從日志中可以看到，初始化的時候是2個線程在工作，執(zhí)行速度較為緩慢，當經(jīng)過第一次擴容后，會觀察到線程池里線程個數(shù)增加了，執(zhí)行任務的速度就越來越快了，本文一共擴容了2次，第一次是擴容到activeSize的大小，第二次是擴容到maxSize，在執(zhí)行任務的過程中，當線程數(shù)過多的時候就會觸發(fā)回收機制...

[線程] - [MY-THREAD-NAME-1] 開始工作...
[線程] - [MY-THREAD-NAME-2] 開始工作...
[線程] - [MY-THREAD-NAME-1] 工作完畢...
[線程] - [MY-THREAD-NAME-1] 開始工作...
[線程] - [MY-THREAD-NAME-2] 工作完畢...
[線程] - [MY-THREAD-NAME-2] 開始工作...
[線程] - [MY-THREAD-NAME-1] 工作完畢...
[線程] - [MY-THREAD-NAME-1] 開始工作...
[線程] - [MY-THREAD-NAME-2] 工作完畢...
[線程] - [MY-THREAD-NAME-2] 開始工作...
[初次擴充] - [SimpleThreadPoolExecutor{threadPoolSize=5, taskQueueSize=44, minSize=2, maxSize=10, activeSize=5}]
[線程] - [MY-THREAD-NAME-3] 開始工作...
...
[線程] - [MY-THREAD-NAME-6] 開始工作...
[線程] - [MY-THREAD-NAME-7] 開始工作...
[再次擴充] - [SimpleThreadPoolExecutor{threadPoolSize=10, taskQueueSize=30, minSize=2, maxSize=10, activeSize=5}]
[線程] - [MY-THREAD-NAME-10] 開始工作...
...
[線程] - [MY-THREAD-NAME-5] 開始工作...
[資源回收] - [SimpleThreadPoolExecutor{threadPoolSize=10, taskQueueSize=4, minSize=2, maxSize=10, activeSize=5}]
[線程] - [MY-THREAD-NAME-4] 工作完畢...
...
[線程] - [MY-THREAD-NAME-7] 工作完畢...
[資源回收] - [SimpleThreadPoolExecutor{threadPoolSize=5, taskQueueSize=0, minSize=2, maxSize=10, activeSize=5}]
[資源回收] - [SimpleThreadPoolExecutor{threadPoolSize=5, taskQueueSize=0, minSize=2, maxSize=10, activeSize=5}]

總結(jié)

通過本文，大致可以了解線程池的工作原理和實現(xiàn)方式，學習的過程中，就是要知其然知其所以然。這樣才能更好地駕馭它，更好地去理解和使用，也能更好地幫助我們觸類旁通，后面的文章中會詳細介紹java.util.concurrent中的線程池。

- 說點什么

全文代碼：https://gitee.com/battcn/battcn-concurent/tree/master/Chapter1-1/battcn-thread/src/main/java/com/battcn/chapter12

個人QQ：1837307557

battcn開源群（適合新手）：391619659

微信公眾號：battcn（歡迎調(diào)戲）

云服務器 GPU云服務器簡易java線程池線程池并發(fā) linux線程池并發(fā) java線程池并發(fā)

文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/67758.html

發(fā)表評論

登陸后可評論

0條評論

Harriet666

男|高級講師

我要關(guān)注我要私信

TA的文章

tensor

閱讀 1801·2023-04-26 00:20
景安快云服務器怎么樣（雙11景安快云VPS和服務器優(yōu)惠促銷）

閱讀 1851·2021-11-08 13:21
首次公開，整理12年積累的博客收藏夾，零距離展示《收藏夾吃灰》系列博客

閱讀 2049·2021-09-10 10:51
ESP32開源驅(qū)動庫Easyio的使用，基于ESP-IDF開發(fā)框架，非Arduino

閱讀 1606·2021-09-10 10:50
前端進階系列(五):flex布局

閱讀 3334·2019-08-30 15:54
瀏覽器對于前端的作用

閱讀 2161·2019-08-30 14:22
響應式可彈出橫向?qū)Ш綑?/a>

閱讀 1457·2019-08-29 16:10

js數(shù)組用法

閱讀 3119·2019-08-26 11:50

成人国产在线小视频_日韩寡妇人妻调教在线播放_色成人www永久在线观看_2018国产精品久久_亚洲欧美高清在线30p_亚洲少妇综合一区_黄色在线播放国产_亚洲另类技巧小说校园_国产主播xx日韩_a级毛片在线免费

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

一起學并發(fā)編程 - 簡易線程池實現(xiàn)

相關(guān)文章

**一起學并發(fā)編程 - 處理異常中止的線程**

**一起學并發(fā)編程 - 等待與通知**

java并發(fā)編程學習之線程池-ThreadPoolExecutor(三)

Python

**一起學并發(fā)編程 - 守護線程**

發(fā)表評論

0條評論

Harriet666

男|高級講師

TA的文章

tensor

景安快云服務器怎么樣（雙11景安快云VPS和服務器優(yōu)惠促銷）

首次公開，整理12年積累的博客收藏夾，零距離展示《收藏夾吃灰》系列博客

ESP32開源驅(qū)動庫Easyio的使用，基于ESP-IDF開發(fā)框架，非Arduino

前端進階系列(五):flex布局

瀏覽器對于前端的作用

響應式可彈出橫向?qū)Ш綑?/a>

js數(shù)組用法

最新活動

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

一起學并發(fā)編程 - 簡易線程池實現(xiàn)

相關(guān)文章

發(fā)表評論

0條評論

男|高級講師

TA的文章

最新活動

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！