以下代碼的實作邏輯出自于公眾號碼農翻身

《你管這破玩意叫執行緒池？》

- PS:劉欣老師在我心中是軟體技術行業的大劉，

執行緒池介面

public interface Executor {
    public void execute(Runnable r);
}

View Code

介面中只有一個抽象方法，execute(Runnable r);它接收一個Runnable,無回傳值實作它的子類只需要將傳入的Runnable執行即可，

NewsThreadExecutor

package com.datang.bingxiang.run;

import com.datang.bingxiang.run.intr.Executor;

public class NewsThreadExecutor implements Executor {
    //每次呼叫都創建一個新的執行緒
    @Override
    public void execute(Runnable r) {
        new Thread(r).start();
    }

}

View Code

這個實作類最簡單也最明白,真的每次呼叫我們都創建一個Thread將引數Runnable執行，這么做的弊端就是每個呼叫者發布一個任務都需要創建一個新的執行緒，執行緒使用后就被銷毀了，對記憶體造成了很大的浪費，

SingThreadExecutor

package com.datang.bingxiang.run;

import java.util.concurrent.ArrayBlockingQueue;

import com.datang.bingxiang.run.intr.Executor;

//只有一個執行緒，在實體化后就啟動執行緒，用戶呼叫execute()傳遞的Runnable會添加到佇列中，
//佇列有一個固定的容量3，如果佇列滿則拋棄任務，
//執行緒的run方法不停的回圈，從佇列里取Runnable然后執行其run()方法，
public class SingThreadExecutor implements Executor {

    // ArrayBlockingQueue 陣列型別的有界佇列
    // LinkedBlockingDeque 鏈表型別的有界雙端佇列
    // LinkedBlockingQueue 鏈表型別的有界單向佇列
    private ArrayBlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1);

    //執行緒不停的從佇列獲取任務
    private Thread worker = new Thread(() -> {
        while (true) {
            try {
                //take會在獲取不到任務時阻塞，并且也有Lock鎖
                Runnable r = queue.take();
                r.run();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }

        }
    });

    // 建構式啟動執行緒
    public SingThreadExecutor() {
        worker.start();
    }

    @Override
    public void execute(Runnable r) {
        // 這個offer和add不同的是offer有Lock鎖，如果佇列滿則回傳false，
        // add則是佇列滿拋出例外，并且沒有Lock鎖，
        if (!queue.offer(r)) {
            System.out.println("執行緒等待佇列已滿，不可加入，本次任務丟棄！");
        }
    }

}

View Code

改變下思路,這次執行緒池實作類只創建一個執行緒,呼叫者發布的任務都存放到一個佇列中(佇列符合先進先出的需求)但是注意我們設計執行緒池一定要選擇有界佇列，因為我們不能無限制的往佇列中添加任務，在佇列滿后，在進來的任務就要被拒絕掉，ArrayBlockingQueue

是一個底層有陣列實作的有界阻塞佇列，實體化一個ArrayBlockingQueue傳遞引數為1，表示佇列長度最大為1.唯一的一個作業執行緒也是成員變數，執行緒執行后不斷的自旋從佇列中獲取任務，take()方法將佇列頭的元素出隊，若佇列為空則阻塞，這個方法是執行緒安全的，

execute(r)方法接收到任務后，將任務添加到佇列中，offer()方法將元素添加到佇列若佇列已滿則回傳false，execute(r)則直接拒絕掉本次任務，

CorePollThreadExecutor

SingThreadExecutor執行緒池的缺點是只有一個作業執行緒，這樣顯然是不夠靈活，CorePollThreadExecutor中增加了corePollSize核心執行緒數引數，由用戶規定有需要幾個作業執行緒，這次我們選用的佇列為LinkedBlockingQueue這是一個資料結構為鏈表的有界阻塞單向佇列，

initThread()方法根據corePollSize回圈創建N個執行緒，執行緒創建后同樣呼叫take()方法從阻塞佇列中獲取元素，若獲取成功則執行Runnable的run()方法，若獲取佇列中沒有元素則阻塞，execute(r)則還是負責將任務添加到佇列中，

CountCorePollThreadExecutor

CorePollThreadExecutor中有三個問題

1 當佇列滿時執行緒池直接拒絕了任務，這應該讓用戶決定被拒絕的任務如何處理，

2 執行緒的創建策略也應該交給用戶做處理，

3 初始化后就創建了N個核心執行緒數，但是這些執行緒可能會用不到而造成浪費，

RejectedExecutionHandler介面的實作應該讓用戶決定如何處理佇列滿的例外情況，

package com.datang.bingxiang.run.intr;

public interface RejectedExecutionHandler {
    public void rejectedExecution();
}

View Code

package com.datang.bingxiang.run;

import com.datang.bingxiang.run.intr.RejectedExecutionHandler;

public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {

    @Override
    public void rejectedExecution() {
        System.out.println("佇列已經滿了！！！！當前task被拒絕");
    }

}

View Code

ThreadFactory介面的實作應該讓用戶決定創建執行緒的方法，

package com.datang.bingxiang.run.intr;

public interface ThreadFactory {
    public Thread newThread(Runnable r);
}

View Code

package com.datang.bingxiang.run;

import com.datang.bingxiang.run.intr.ThreadFactory;

public class CustomThreadFactory implements ThreadFactory {
    @Override
    public Thread newThread(Runnable r) {
        System.out.println("創建了新的核心執行緒");
        return new Thread(r);
    }

}

View Code

CountCorePollThreadExecutor的建構式接收三個引數corePollSize，rejectedExecutionHandler，threadFactory，因為現在我們需要按需創建核心執行緒，所以需要一個變數workCount記錄當前已經創建的作業執行緒，為了保證執行緒之間拿到的workCount是最新的(可見性)，我們需要給變數workCount加上volatile修飾，保證改變了的修改能被所有執行緒看到，execute(r)首先要呼叫initThread(r)判斷是否有執行緒被創建，如果沒有執行緒創建則表示作業執行緒數已經和核心執行緒數相同了，此時需要將新的任務添加到佇列中，如果佇列滿，則執行傳入的拒絕策略，重要的方法在于initThread（r)，initThread(r)方法回傳true表示有作業執行緒被創建任務將被作業執行緒直接執行，無需入佇列，回傳false則將任務入隊，佇列滿則執行拒絕策略，

fill變數表示核心執行緒數是否全部創建，為了保證多執行緒的環境下不會創建多于corePoolSize個數的執行緒，所以需要使用同步鎖，initThread(r)都要使用鎖則會降低效率，尤其是當作業執行緒數已經到達核心執行緒數后，所以這一塊代碼使用到了雙重判斷，當加鎖后在此判斷作業執行緒是否已滿，如果已滿回傳false，接下來使用threadFactory工廠創建執行緒，在執行緒中使用代碼塊，保證當前任務可以被新創建的作業執行緒執行，新的作業執行緒依然是從佇列中獲取任務并執行，執行緒開啟后作業執行緒++，如果作業執行緒數等于核心執行緒數則改變fill標記，回傳true，成功創建執行緒，不要忘記在finally中釋放鎖，

package com.datang.bingxiang.run;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.datang.bingxiang.run.intr.Executor;
import com.datang.bingxiang.run.intr.RejectedExecutionHandler;
import com.datang.bingxiang.run.intr.ThreadFactory;

public class CountCorePollThreadExecutor implements Executor {

    // 核心執行緒數
    private Integer corePollSize;

    // 作業執行緒數，也就是執行緒實體的數量
    private volatile Integer workCount = 0;

    // 執行緒是否已滿
    private volatile boolean fill = false;

    // 拒絕策略，由呼叫者傳入，當佇列滿時，執行自定義策略
    private RejectedExecutionHandler rejectedExecutionHandler;

    // 執行緒工廠，由呼叫者傳入
    private ThreadFactory threadFactory;

    public CountCorePollThreadExecutor(Integer corePollSize, RejectedExecutionHandler rejectedExecutionHandler,
            ThreadFactory threadFactory) {
        this.corePollSize = corePollSize;
        this.rejectedExecutionHandler = rejectedExecutionHandler;
        this.threadFactory = threadFactory;
    }

    // 這次使用鏈表型別的單向佇列
    LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(1);

    @Override
    public void execute(Runnable r) {
        // 如果沒有創建執行緒
        if (!initThread(r)) {
            // offer和ArrayBlockingQueue的offer相同的作用
            if (!queue.offer(r)) {
                rejectedExecutionHandler.rejectedExecution();
            }
        }

    }

    // 同步鎖，因為判斷核心執行緒數和作業執行緒數的操作需要執行緒安全
    Lock lock = new ReentrantLock();

    public boolean initThread(Runnable r) {
        // 如果作業執行緒沒有創建滿則需要創建，
        if (!fill) {
            try {
                lock.lock();// 把鎖 加在判斷里邊是為了不讓每次initThread方法執行時都加鎖
                // 此處進行雙重判斷，因為可能因為多執行緒原因多個執行緒都判斷作業執行緒沒有創建滿，但是不要緊
                // 只有一個執行緒可以進來，如果后續執行緒二次判斷已經滿了就直接回傳，
                if (fill) {
                    return false;
                }
                Thread newThread = threadFactory.newThread(() -> {
                    // 因為執行緒是由任務觸發創建的，所以先把觸發執行緒創建的任務執行掉，
                    {
                        r.run();
                    }

                    while (true) {
                        // 然后該執行緒則不停的從佇列中獲取任務
                        try {
                            Runnable task = queue.take();
                            task.run();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                });
                newThread.start();
                // 作業執行緒數+1
                workCount++;
                // 如果作業執行緒數已經與核心執行緒數相等，則不可創建
                if (workCount == corePollSize) {
                    fill = true;
                }
                return true;
            } finally {
                lock.unlock();// 釋放鎖
            }
        } else {
            // 作業執行緒已滿則不創建
            return false;
        }

    }

}

View Code

ThreadPoolExecutor

最后考慮下，當作業執行緒數到達核心執行緒數后，佇列也滿了以后，任務就被拒絕了，能不能想個辦法，當作業執行緒滿后，多增加幾個執行緒作業，當任務不多時在將擴展的執行緒銷毀，ThreadPoolExecutor的建構式中新增三個引數maximumPoolSize最大執行緒數keepAliveTime空閑時間，unit空閑時間的單位，

和CountCorePollThreadExecutor相比較在流程上講我們只需要在佇列滿時判斷作業執行緒是否和最大執行緒數相等，如果不相等則創建備用執行緒，并且在備用執行緒長時間不作業時需要銷毀掉作業執行緒，create()方法雙重判斷workCount==maximumPoolSize如果已經相等表示已經不能創建執行緒了，此時只能執行拒絕策略，否則創建備用執行緒，備用執行緒創建后自旋的執行poll(l,u)方法，該方法也是取出佇列頭元素，和take()不同的是，poll如果一段時間后仍然從佇列中拿不到元素（佇列為空）則回傳null，此時我們需要將該備用執行緒銷毀，在創建執行緒后將workCount++，此外需要注意，因為當前佇列滿了，所以才會創建備用執行緒所以不要將當前的任務給忘了，LinkedBlockingQueue的put(r)方法會阻塞的添加元素，直到添加成功，最后 stop()判讀如果workCount>corePollSize則在執行緒安全的環境下將執行緒停止，并且將workCount--，

package com.datang.bingxiang.run;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.datang.bingxiang.run.intr.Executor;
import com.datang.bingxiang.run.intr.RejectedExecutionHandler;
import com.datang.bingxiang.run.intr.ThreadFactory;

public class ThreadPoolExecutor implements Executor {

    // 核心執行緒數
    private Integer corePollSize;

    // 作業執行緒數，也就是執行緒實體的數量
    private Integer workCount = 0;

    // 當佇列滿時，需要創建新的Thread,maximumPoolSize為最大執行緒數
    private Integer maximumPoolSize;

    // 當任務不多時，需要洗掉多余的執行緒，keepAliveTime為空閑時間
    private long keepAliveTime;

    // unit為空閑時間的單位
    private TimeUnit unit;

    // 執行緒是否已滿
    private boolean fill = false;

    // 拒絕策略，由呼叫者傳入，當佇列滿時，執行自定義策略
    private RejectedExecutionHandler rejectedExecutionHandler;

    // 執行緒工廠，由呼叫者傳入
    private ThreadFactory threadFactory;

    // 這次使用鏈表型別的單向佇列
    BlockingQueue<Runnable> workQueue;

    public ThreadPoolExecutor(Integer corePollSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
            RejectedExecutionHandler rejectedExecutionHandler) {
        this.corePollSize = corePollSize;
        this.rejectedExecutionHandler = rejectedExecutionHandler;
        this.threadFactory = threadFactory;
        this.workQueue = workQueue;
        this.maximumPoolSize = maximumPoolSize;
        this.keepAliveTime = keepAliveTime;
        this.unit = unit;
    }

    @Override
    public void execute(Runnable r) {
        // 如果沒有創建執行緒
        if (!initThread(r)) {
            // offer和ArrayBlockingQueue的offer相同的作用
            if (!workQueue.offer(r)) {
                // 佇列滿了以后先不走拒絕策略而是查詢執行緒數是否到達最大執行緒數
                if (create()) {
                    Thread newThread = threadFactory.newThread(() -> {
                        while (true) {
                            // 然后該執行緒則不停的從佇列中獲取任務
                            try {
                                Runnable task = workQueue.poll(keepAliveTime, unit);
                                if (task == null) {
                                    stop();
                                } else {
                                    task.run();
                                }
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                    });
                    newThread.start();
                    // 作業執行緒數+1
                    workCount++;
                    // 增加執行緒后，還需要將本應該被拒絕的任務添加到佇列
                    try {
                        // 這個put()方法會在佇列滿時阻塞添加，直到添加成功
                        workQueue.put(r);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    rejectedExecutionHandler.rejectedExecution();
                }
            }
        }

    }

    Lock clock = new ReentrantLock();

    private boolean create() {
        //雙重檢查
        if (workCount == maximumPoolSize) {
            return false;
        }
        try {
            clock.lock();
            if (workCount < maximumPoolSize) {
                return true;
            } else {
                return false;
            }
        } finally {
            clock.unlock();
        }

    }

    Lock slock = new ReentrantLock();

    // 銷毀執行緒
    private void stop() {
        slock.lock();
        try {
            if (workCount > corePollSize) {
                System.out.println(Thread.currentThread().getName() + "執行緒被銷毀");
                workCount--;
                Thread.currentThread().stop();
            }
        } finally {
            slock.unlock();
        }

    }

    // 獲取當前的作業執行緒數
    public Integer getworkCount() {
        return workCount;
    }

    // 同步鎖，因為判斷核心執行緒數和作業執行緒數的操作需要執行緒安全
    Lock lock = new ReentrantLock();

    public boolean initThread(Runnable r) {
        // 如果作業執行緒沒有創建滿則需要創建，
        if (!fill) {
            try {
                lock.lock();// 把鎖 加在判斷里邊是為了不讓每次initThread方法執行時都加鎖
                // 此處進行雙重判斷，因為可能因為多執行緒原因多個執行緒都判斷作業執行緒沒有創建滿，但是不要緊
                // 只有一個執行緒可以進來，如果后續執行緒二次判斷已經滿了就直接回傳，
                if (fill) {
                    return false;
                }
                Thread newThread = threadFactory.newThread(() -> {
                    // 因為執行緒是由任務觸發創建的，所以先把觸發執行緒創建的任務執行掉，
                    {
                        r.run();
                    }

                    while (true) {
                        // 然后該執行緒則不停的從佇列中獲取任務
                        try {
                            Runnable task = workQueue.take();
                            task.run();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }

                    }
                });
                newThread.start();
                // 作業執行緒數+1
                workCount++;
                // 如果作業執行緒數已經與核心執行緒數相等，則不可創建
                if (workCount == corePollSize) {
                    fill = true;
                }
                return true;
            } finally {
                lock.unlock();// 釋放鎖
            }
        } else {
            // 作業執行緒已滿則不創建
            return false;
        }
    }

}

View Code

測驗代碼

package com.datang.bingxiang.run.test;

import java.time.LocalDateTime;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import com.datang.bingxiang.run.CorePollThreadExecutor;
import com.datang.bingxiang.run.CountCorePollThreadExecutor;
import com.datang.bingxiang.run.CustomRejectedExecutionHandler;
import com.datang.bingxiang.run.CustomThreadFactory;
import com.datang.bingxiang.run.NewsThreadExecutor;
import com.datang.bingxiang.run.SingThreadExecutor;
import com.datang.bingxiang.run.ThreadPoolExecutor;
import com.datang.bingxiang.run.intr.Executor;

@RestController
public class TestController {



    private int exe1Count = 1;
    Executor newsThreadExecutor = new NewsThreadExecutor();

    // 每次都創建新的執行緒執行
    @GetMapping(value = "https://www.cnblogs.com/zumengjie/p/exe1")
    public String exe1() {
        newsThreadExecutor.execute(() -> {
            System.out.println("正在執行" + exe1Count++);
        });
        return "success";
    }

    /*
     * 等待佇列長度為1，三個執行緒加入，第一個加入后會迅速的出佇列，剩下兩個只有一個可以成功 加入，另一個 則會被丟棄
     */
    private int exe2Count = 1;
    Executor singThreadExecutor = new SingThreadExecutor();

    @GetMapping(value = "exe2")
    public String exe2() {
        singThreadExecutor.execute(() -> {
            System.out.println("正在執行" + exe2Count++);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        return "success";
    }

    private int exe3Count = 1;
    Executor corePollThreadExecutor = new CorePollThreadExecutor(2);

    @GetMapping(value = "exe3")
    public String exe3() {
        corePollThreadExecutor.execute(() -> {
            System.out.println("正在執行" + exe3Count++);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        return "success";
    }

    private int exe4Count = 1;
    Executor countCorePollThreadExecutor = new CountCorePollThreadExecutor(2, new CustomRejectedExecutionHandler(),
            new CustomThreadFactory());

    @GetMapping(value = "exe4")
    public String exe4() {
        countCorePollThreadExecutor.execute(() -> {
            System.out.println("正在執行" + exe4Count++);
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        return "success";
    }

    // 第一次創建執行緒并執行 1
    // 第二次進入佇列 2
    // 第三次創建執行緒取出佇列中的2，將3添加到佇列
    // 第四次拒絕
    // 等待3秒后只剩下一個佇列
    private int exe5Count = 1;
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 3, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1), new CustomThreadFactory(), new CustomRejectedExecutionHandler());

    @GetMapping(value = "exe5")
    public String exe5() {
        threadPoolExecutor.execute(() -> {
            System.out.println("正在執行" + exe5Count++);
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        return "success";
    }

    @GetMapping(value = "workCount")
    public Integer getWorkCount() {
        return threadPoolExecutor.getworkCount();
    }
}

View Code

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/266582.html

標籤：Java

上一篇：Spring Security 整合微信小程式登錄的思路探討

下一篇：Shell腳本控制docker容器啟動順序

造輪子-ThreadPoolExecutor

執行緒池介面

NewsThreadExecutor

SingThreadExecutor

CorePollThreadExecutor

CountCorePollThreadExecutor

ThreadPoolExecutor

測驗代碼