Java如何手動創(chuàng)建線程池

更新時間：2022年08月26日 09:41:50 作者：記憶的深藍

這篇文章主要介紹了Java如何手動創(chuàng)建線程池，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

如何手動創(chuàng)建線程池

jdk提供了一個通過ThreadPoolExecutor創(chuàng)建一個線程池的類

構(gòu)造器

使用給定的參數(shù)和默認的飽和策略、默認的工廠方法創(chuàng)建線程池

ThreadPoolExecutor(int corePoolSize,?
int maximumPoolSize,?
long keepAliveTime,?
TimeUnit unit,?
BlockingQueue<Runnable> workQueue)

使用給定的參數(shù)和默認的工廠方法創(chuàng)建線程池

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

使用給定的參數(shù)和默認的飽和策略（AbortPolicy）創(chuàng)建線程池

ThreadPoolExecutor(int corePoolSize,?
int maximumPoolSize,
long keepAliveTime,
BlockingQueue<Runnable> workQueue,?
ThreadFactory threadFactory)

使用指定的參數(shù)創(chuàng)建線程池

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

參數(shù)說明

corePoolSize 線程池的基本大小, 當提交一個任務到線程池的時候，線程池會創(chuàng)建一個線程來執(zhí)行任務，即使當前線程池已經(jīng)存在空閑線程，仍然會創(chuàng)建一個線程，等到需要執(zhí)行的任務數(shù)大于線程池基本大小時就不再創(chuàng)建。如果調(diào)用線程池的prestartAllCoreThreads()方法，線程池會提前創(chuàng)建并啟動所有的基本線程。
maximumPoolSizeSize 線程池最大數(shù)量，線程池允許創(chuàng)建的最大線程數(shù)，如果隊列滿了，并且已創(chuàng)建的線程數(shù)小于最大線程數(shù)，則線程池會再創(chuàng)建新的線程執(zhí)行任務。值得注意的是，如果使用了無界的任務隊列這個參數(shù)就沒什么效果。
keepAliveTime 線程活動保持時間，線程池的工作線程空閑后，保持存活的時間，所以，如果任務很多，并且每個任務執(zhí)行的時間比較短，可以調(diào)大時間，提高線程的利用率。
unit 線程活動保持時間的單位，可選擇的單位有時分秒等等。
workQueue 任務隊列。用來暫時保存任務的工作隊列
threadFactory 用于創(chuàng)建線程的工廠

隊列

ArrayBlockingQueue：是一個基于數(shù)組結(jié)構(gòu)的有界阻塞隊列，此隊列按照FIFO（先進先出）原則對元素進行排序
DelayQueue
LinkedBlockingDeque
LinkedBlockingQueue：是一個基于鏈表結(jié)構(gòu)的有界阻塞隊列，此隊列按照FIFO排序元素，吞吐量高于ArrayBlockingQueue。靜態(tài)工廠方法Executors.newFixedThreadPool(n)使用了此隊列
LinkedTransferQueue
PriorityBlockingQueue：一個具有優(yōu)先級的無限阻塞隊列
SynchronousQueue：一個不存儲元素的阻塞隊列。每個插入操作必須等待另一個線程調(diào)用移除操作，否則插入操作一直處于阻塞狀態(tài)，吞吐量要高于LinkedBlockingQueue，靜態(tài)工廠方法Executors.newCachedThreadPool()使用了此隊列

飽和策略

當隊列和線程池都滿了，說明線程池處于飽和的狀態(tài)，那么必須采取一種策略處理提交的新任務。這個策略默認是AbortPolicy，表示無法處理新任務時拋出異常

ThreadPoolExecutor.AbortPolicy：直接拋出異常
ThreadPoolExecutor.CallerRunsPolicy：只用調(diào)用這所在的線程來運行任務
ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊列里最近的一個任務，并執(zhí)行當前任務
ThreadPoolExecutor.DiscardPolicy：不處理，丟棄掉

示例

public class ThreadPool {
? ? /**
? ? ?* 線程池的基本大小
? ? ?*/
? ? static int corePoolSize = 10;
? ? /**
? ? ?* 線程池最大數(shù)量
? ? ?*/
? ? static int maximumPoolSizeSize = 100;
? ? /**
? ? ?* 線程活動保持時間
? ? ?*/
? ? static long keepAliveTime = 1;
? ? /**
? ? ?* 任務隊列
? ? ?*/
? ? static ArrayBlockingQueue workQueue = new ArrayBlockingQueue(10);
? ? public static void main(String[] args) {
? ? ? ? ThreadPoolExecutor executor = new ThreadPoolExecutor(
? ? ? ? ? ? ? ? corePoolSize,
? ? ? ? ? ? ? ? maximumPoolSizeSize,
? ? ? ? ? ? ? ? keepAliveTime,
? ? ? ? ? ? ? ? TimeUnit.SECONDS,
? ? ? ? ? ? ? ? workQueue,
? ? ? ? ? ? ? ? new ThreadFactoryBuilder().setNameFormat("XX-task-%d").build());
? ? ? ? //提交一個任務
? ? ? ? executor.execute(() -> System.out.println("ok"));
? ? }
}

源碼分析

任務執(zhí)行

public void execute(Runnable command) {
? ? ? ? if (command == null)
? ? ? ? ? ? throw new NullPointerException();
? ? ? ? /*
? ? ? ? ?* Proceed in 3 steps:
? ? ? ? ?*
? ? ? ? ?* 1. If fewer than corePoolSize threads are running, try to
? ? ? ? ?* start a new thread with the given command as its first
? ? ? ? ?* task. ?The call to addWorker atomically checks runState and
? ? ? ? ?* workerCount, and so prevents false alarms that would add
? ? ? ? ?* threads when it shouldn't, by returning false.
? ? ? ? ?*
? ? ? ? ?* 2. If a task can be successfully queued, then we still need
? ? ? ? ?* to double-check whether we should have added a thread
? ? ? ? ?* (because existing ones died since last checking) or that
? ? ? ? ?* the pool shut down since entry into this method. So we
? ? ? ? ?* recheck state and if necessary roll back the enqueuing if
? ? ? ? ?* stopped, or start a new thread if there are none.
? ? ? ? ?*
? ? ? ? ?* 3. If we cannot queue task, then we try to add a new
? ? ? ? ?* thread. ?If it fails, we know we are shut down or saturated
? ? ? ? ?* and so reject the task.
? ? ? ? ?*/
? ? ? ? int c = ctl.get();
? ? ? ? if (workerCountOf(c) < corePoolSize) {
? ? ? ? ? ? if (addWorker(command, true))
? ? ? ? ? ? ? ? return;
? ? ? ? ? ? c = ctl.get();
? ? ? ? }
? ? ? ? if (isRunning(c) && workQueue.offer(command)) {
? ? ? ? ? ? int recheck = ctl.get();
? ? ? ? ? ? if (! isRunning(recheck) && remove(command))
? ? ? ? ? ? ? ? reject(command);
? ? ? ? ? ? else if (workerCountOf(recheck) == 0)
? ? ? ? ? ? ? ? addWorker(null, false);
? ? ? ? }
? ? ? ? else if (!addWorker(command, false))
? ? ? ? ? ? reject(command);
? ? }

參考文檔

https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ThreadPoolExecutor.html

線程池工具類

實現(xiàn)線程的三種方式

1.繼承 Thread 類

2.實現(xiàn)Runnable 接口

3.實現(xiàn) Callbale接口和Future接口實現(xiàn)

4.三種方式比較：

繼承Thread 類編程簡單，可擴展性差。

實現(xiàn)接口方式可擴展性高，編程復雜。

使用ThreadPoolExecutor編寫線程池工具類

1.線程創(chuàng)建方式，實例化賢臣池時，創(chuàng)建核心線程，

2.當任務大于核心線程時將進入阻塞隊列

3.當阻塞隊列滿時，任務沒有超過最大線程時創(chuàng)建新的線程

4.當任務 > 最大線程數(shù)+阻塞隊列時，執(zhí)行拒絕策略。

public class ThreadPoolUtils {
    public static ThreadPoolExecutor pool=null;
    // 無響應執(zhí)行
    public static void execute(Runnable runnable){
        getThreadPool().execute(runnable);
    }
    // 有響應執(zhí)行
    public static<T> Future<T> submit(Callable<T> callable){
        return getThreadPool().submit(callable);
    }
    // 創(chuàng)造線程池
    private static synchronized ThreadPoolExecutor getThreadPool(){
        if(pool==null){
            // 獲取處理器數(shù)量
            int cpuNum = Runtime.getRuntime().availableProcessors();
            // 根據(jù)cpu數(shù)量,計算出合理的線程并發(fā)數(shù)
            // 最佳線程數(shù)目 = （（線程等待時間+線程CPU時間）/線程CPU時間 ）* CPU數(shù)目
            int maximumPoolSize = cpuNum * 2 + 1;
            // 七個參數(shù)
            // 1. 核心線程數(shù)
            // 2. 最大線程數(shù)
            // 3. 空閑線程最大存活時間
            // 4. 時間單位
            // 5. 阻塞隊列
            // 6. 創(chuàng)建線程工廠
            // 7. 拒絕策略
            pool=new ThreadPoolExecutor(maximumPoolSize-1,
            maximumPoolSize,
            5,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(50),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());
        }
        return pool;
    }
}

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: