jdk線程池的實(shí)現(xiàn)

更新時(shí)間：2023年04月04日 15:51:07 作者：ThinkPet

本文主要介紹了jdk線程池的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

jdk線程池ThreadPoolExecutor的7個(gè)參數(shù)

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

corePoolSize

核心線程個(gè)數(shù) ,int類型

maximunPoolSize

最大線程數(shù) ,int類型

keepAliveTime存活時(shí)間

傳long類型的值,

當(dāng)線程池中的線程數(shù)大于corePoolSize核心線程個(gè)數(shù),且線程是閑置狀態(tài),則這些空閑線程的最大存活時(shí)間是KeepAliveTime

TimeUnit

存活時(shí)間的單位, 有時(shí)/分/秒/毫秒等可選配置

workQueue

存放待執(zhí)行任務(wù)的阻塞隊(duì)列, 可傳入

arrayBlockingQueue 基于數(shù)組的有界阻塞隊(duì)列;

linkedBlockingQueue基于鏈表的無界阻塞隊(duì)列;

synchronousQueue最多只有1個(gè)元素的同步隊(duì)列, 隊(duì)列容量是1;

priorityBlockingQueue帶優(yōu)先級(jí)的無界阻塞隊(duì)列,出隊(duì)元素是優(yōu)先級(jí)最高或最低的元素;

DelayQueue 帶延遲功能的無界阻塞隊(duì)列, 過期元素才會(huì)出隊(duì),隊(duì)頭元素是快要過期的元素.

以上幾個(gè)Queue都是BlockingQueue的實(shí)現(xiàn)類

threadFactory

創(chuàng)建線程的工廠,

jdk提供了DefaultThreadFactory默認(rèn)工廠,

用Executors.defaultThreadFactory()就行.

RejectedExecutionHandler拒絕策略

當(dāng)隊(duì)列滿且線程數(shù)達(dá)到maximunPoolSize最大線程數(shù)后采取的策略, 可傳入

AbortPolicy 拋出異常,這個(gè)是默認(rèn)策略.

CallersRunPolicy 由調(diào)用者所在的線程執(zhí)行任務(wù)

DiscardOldestPolicy 丟棄最老的任務(wù)

DiscardPolicy 丟棄新任務(wù),不拋出異常

jdk提供的Executors快速創(chuàng)建線程池的用法

jdk封裝了一個(gè)Executors類可以直接創(chuàng)建各種線程池,

用法形如

ExecutorService pool = Executors.newXXXXXPool()

可以用Executors類創(chuàng)建業(yè)務(wù)常用的3種線程池

固定線程池

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

創(chuàng)建一個(gè)核心線程數(shù)和最大線程數(shù)相同的線程池,都為nThreads,

且線程池的阻塞隊(duì)列長(zhǎng)度是Integer.MAX_VALUE,

且keepAliveTime=0,說明只要線程個(gè)數(shù)比核心線程個(gè)數(shù)多并且當(dāng)前空閑則回收.

單線程線程池

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

創(chuàng)建一個(gè)核心線程數(shù)和最大線程數(shù)都是1的線程池,

且線程池的阻塞隊(duì)列長(zhǎng)度是Integer.MAX_VALUE,

且keepAliveTime=0,說明只要線程個(gè)數(shù)比核心線程個(gè)數(shù)多并且當(dāng)前空閑則回收.

已緩存的線程池

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

創(chuàng)建一個(gè)按需創(chuàng)建線程的線程池,初始線程個(gè)數(shù)為0,最多線程個(gè)數(shù)為

Integer.MAX_VALUE,并且阻塞隊(duì)列為同步隊(duì)列.

keepAliveTime=60,說明當(dāng)前線程在60s內(nèi)空閑則回收.

CachedThreadPool的特殊之處在于,加入同步隊(duì)列的任務(wù)會(huì)被馬上執(zhí)行,同步隊(duì)列里邊最多只有1個(gè)任務(wù).

使用創(chuàng)建好的ExecutorService 線程池執(zhí)行異步任務(wù)

submit操作

提交一個(gè)任務(wù), 任務(wù)參數(shù)可以是 Runnable實(shí)現(xiàn)類或 Callable 實(shí)現(xiàn)類.

返回的類型是Future 表示異步計(jì)算的結(jié)果, 可以用future.get()方法拿到數(shù)據(jù).

shutdown操作

調(diào)用shutdown方法后,線程池就不會(huì)再接受新的任務(wù)了,但是工作隊(duì)列里邊的任務(wù)還是要執(zhí)行的, 該方法會(huì)立刻返回,不等待隊(duì)列任務(wù)完成再返回.

使用線程池的情況下當(dāng)程序結(jié)束時(shí)記得調(diào)用shutdown關(guān)閉線程池, 如果不關(guān)閉線程池,則會(huì)導(dǎo)致線程池資源一直不被釋放.

shutdownNow操作

調(diào)用shutdownNow方法后,線程池就不會(huì)再接受新的任務(wù)了,并且會(huì)丟棄工作隊(duì)列里邊的任務(wù),正在執(zhí)行的任務(wù)會(huì)被中斷,該方法會(huì)立刻返回,并不等待激活的任務(wù)執(zhí)行完成. 返回值為這時(shí)候隊(duì)列里面被丟棄的任務(wù)列表.

awaitTermination操作

當(dāng)線程調(diào)用awaitTermination方法后,當(dāng)前線程會(huì)被阻塞, 直到線程池狀態(tài)變?yōu)門ERMINATED 才返回,或者等待時(shí)間超時(shí)才返回.

案例1-測(cè)試FixedThreadPool執(zhí)行CallableTask任務(wù)

package cn.demo;

import cn.hutool.core.util.RandomUtil;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ExecutorTestsForCallableTask {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        String res1 = "";
        String res2 = "";
        String res3 = "";
        String res4 = "";

        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);

        //submit 提交4個(gè)任務(wù), 實(shí)際執(zhí)行時(shí),任務(wù)是并發(fā)執(zhí)行的,執(zhí)行順序不固定
        Future<String> submit1 = fixedThreadPool.submit(
                new TestCallableTask(RandomUtil.randomInt(30,1000),"t1"));
        Future<String> submit2 = fixedThreadPool.submit(
                new TestCallableTask(RandomUtil.randomInt(100,400),"t2"));
        Future<String> submit3 = fixedThreadPool.submit(
                new TestCallableTask(RandomUtil.randomInt(30,350),"t3"));
        Future<String> submit4 = fixedThreadPool.submit(
                new TestCallableTask(RandomUtil.randomInt(310,500),"t4"));

        res1 = submit1.get();
        System.out.println(res1);
        res2 = submit2.get();
        System.out.println(res2);
        res3 = submit3.get();
        System.out.println(res3);
        res4 = submit4.get();
        System.out.println(res4);

        fixedThreadPool.shutdown();
    }
}

package cn.demo;


import cn.hutool.core.util.RandomUtil;

import java.time.LocalDateTime;
import java.util.concurrent.Callable;

public class TestCallableTask implements Callable<String> {

    private int testIntVal;
    private String taskSeq;

    public TestCallableTask(int testIntVal, String taskSeq) {
        this.testIntVal = testIntVal;
        this.taskSeq = taskSeq;
    }

    @Override
    public  String call() throws Exception {
        String s = LocalDateTime.now().toString();
        System.out.println(s+"->"+taskSeq+" run ....");

        int i = testIntVal;
        System.out.println(i);

        try {
            Thread.sleep(RandomUtil.randomInt(100,300));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        if (i>300){
            return "300more";
        }else {
            return "300less";
        }
    }
}

案例2-測(cè)試FixedThreadPool執(zhí)行RunnableTask任務(wù)

package cn.demo;

import java.util.concurrent.*;

public class ExecutorTestsForRunnableTask {

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

        String res1 = "";
        String res2 = "";
        String res3 = "";
        String res4 = "";
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);

        //submit 提交4個(gè)任務(wù), 實(shí)際執(zhí)行時(shí),任務(wù)是并發(fā)執(zhí)行的,執(zhí)行順序不固定
        Task1Param task1Param = new Task1Param();
        task1Param.setUrl("f23r3r");
        task1Param.setName("1heg43t34t34t");
        Future<String> stringFuture = fixedThreadPool.submit(
            new TestTask1Runnable(task1Param), "success1 ok");

        Task1Param t2 = new Task1Param();
        t2.setUrl("gnsg2323");
        t2.setName("2wwswer2r1asdaaws");
        Future<String> f2 = fixedThreadPool.submit(new TestTask1Runnable(t2), "success2 ok");

        Task1Param t3 = new Task1Param();
        t3.setUrl("thwasr23r");
        t3.setName("3erzawfe23rawsf");
        Future<String> f3 = fixedThreadPool.submit(new TestTask1Runnable(t3), "success3 ok");

        Task1Param t4 = new Task1Param();
        t4.setUrl("mjkdsragt");
        t4.setName("4tbertydraewrsfk");
        Future<String> f4 = fixedThreadPool.submit(new TestTask1Runnable(t4), "success4 ok");


        res1 = stringFuture.get();
        System.out.println(res1);
        res2 = f2.get();
        System.out.println(res2);
        res3 = f3.get();
        System.out.println(res3);
        res4 = f4.get();
        System.out.println(res4);

        fixedThreadPool.shutdown();
    }
}

package cn.demo;

import cn.hutool.core.util.RandomUtil;
import java.time.LocalDateTime;

public class TestTask1Runnable implements Runnable{

    private Task1Param task1Param;

    public TestTask1Runnable(Task1Param task1Param) {
        this.task1Param = task1Param;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(RandomUtil.randomInt(200,600));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(task1Param.getName());
        System.out.println(task1Param.getUrl());
        String s = LocalDateTime.now().toString();
        System.out.println(s+" TestTask1Runnable run ....");
    }
}

使用自定義的ThreadPoolExecutor來執(zhí)行異步任務(wù)

package cn.demo;


import cn.hutool.core.util.RandomUtil;
import java.util.concurrent.*;

public class TpeTest {

    private final static ThreadPoolExecutor pool =
            new ThreadPoolExecutor(
                    1,1,
                    1L, TimeUnit.MINUTES,
                    new ArrayBlockingQueue<Runnable>(1),
                    new ThreadPoolExecutor.CallerRunsPolicy());
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Future<String> submit1 = pool.submit(
                new TestCallableTask(RandomUtil.randomInt(30,1000),"t1"));
        Future<String> submit2 = pool.submit(
                new TestCallableTask(RandomUtil.randomInt(100,400),"t2"));
        Future<String> submit3 = pool.submit(
                new TestCallableTask(RandomUtil.randomInt(30,350),"t3"));
        Future<String> submit4 = pool.submit(
                new TestCallableTask(RandomUtil.randomInt(310,500),"t4"));
        System.out.println("task1-"+submit1.get());
        System.out.println("task2-"+submit2.get());
        System.out.println("task3-"+submit3.get());
        System.out.println("task4-"+submit4.get());

        pool.shutdown();
    }
}