jdk线程池技术

电玩女神 2024-03-24 10:16 75阅读 0赞

#### jdk线程池ThreadPoolExecutor的7个参数 ####

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

核心线程个数 ,int类型

> maximunPoolSize

最大线程数 ,int类型

> keepAliveTime存活时间

传long类型的值,

当线程池中的线程数大于corePoolSize核心线程个数,且线程是闲置状态,则这些空闲线程的最大存活时间是KeepAliveTime

> TimeUnit

存活时间的单位, 有时/分/秒/毫秒等可选配置

> workQueue

存放待执行任务的阻塞队列, 可传入

arrayBlockingQueue 基于数组的有界阻塞队列;

linkedBlockingQueue基于链表的无界阻塞队列;

synchronousQueue最多只有1个元素的同步队列, 队列容量是1;

priorityBlockingQueue带优先级的无界阻塞队列,出队元素是优先级最高或最低的元素;

DelayQueue 带延迟功能的无界阻塞队列, 过期元素才会出队,队头元素是快要过期的元素.

以上几个Queue都是BlockingQueue的实现类

> threadFactory

创建线程的工厂,

jdk提供了DefaultThreadFactory默认工厂,

用Executors.defaultThreadFactory()就行.

> RejectedExecutionHandler拒绝策略

当队列满且线程数达到maximunPoolSize最大线程数后采取的策略, 可传入

AbortPolicy 抛出异常,这个是默认策略.

CallersRunPolicy 由调用者所在的线程执行任务

DiscardOldestPolicy 丢弃最老的任务

DiscardPolicy 丢弃新任务,不抛出异常

#### jdk提供的Executors快速创建线程池的用法 ####

jdk封装了一个Executors类可以直接创建各种线程池,

用法形如

ExecutorService pool = Executors.newXXXXXPool()

可以用Executors类创建业务常用的3种线程池

> 固定线程池

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

创建一个核心线程数和最大线程数相同的线程池,都为nThreads,

且线程池的阻塞队列长度是Integer.MAX\_VALUE,

且keepAliveTime=0,说明只要线程个数比核心线程个数多并且当前空闲则回收.

> 单线程线程池

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

创建一个核心线程数和最大线程数都是1的线程池,

且线程池的阻塞队列长度是Integer.MAX\_VALUE,

且keepAliveTime=0,说明只要线程个数比核心线程个数多并且当前空闲则回收.

> 已缓存的线程池

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

创建一个按需创建线程的线程池,初始线程个数为0,最多线程个数为

Integer.MAX\_VALUE,并且阻塞队列为同步队列.

keepAliveTime=60,说明当前线程在60s内空闲则回收.

CachedThreadPool的特殊之处在于,加入同步队列的任务会被马上执行,同步队列里边最多只有1个任务.

#### 使用创建好的ExecutorService 线程池执行异步任务 ####

![pp6qEJ1.png][]

> submit操作

提交一个任务, 任务参数可以是 Runnable实现类 或 Callable 实现类.

返回的类型是Future 表示异步计算的结果, 可以用future.get()方法拿到数据.

> shutdown操作

调用shutdown方法后,线程池就不会再接受新的任务了,但是工作队列里边的任务还是要执行的, 该方法会立刻返回,不等待队列任务完成再返回.

使用线程池的情况下当程序结束时记得调用shutdown关闭线程池, 如果不关闭线程池,则会导致 线程池资源一直不被释放.

> shutdownNow操作

调用shutdownNow方法后,线程池就不会再接受新的任务了,并且会丢弃工作队列里边的任务,正在执行的任务会被中断,该方法会立刻返回,并不等待激活的任务执行完成. 返回值为这时候队列里面被丢弃的任务列表.

> awaitTermination操作

当线程调用awaitTermination方法后,当前线程会被阻塞, 直到线程池状态变为TERMINATED 才返回,或者等待时间超时才返回.

--------------------

案例1-测试FixedThreadPool执行CallableTask任务

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个任务, 实际执行时,任务是并发执行的,执行顺序不固定
            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-测试FixedThreadPool执行RunnableTask任务

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个任务, 实际执行时,任务是并发执行的,执行顺序不固定
            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来执行异步任务 ####

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();
        }
    }

> 线程池使用FutureTask时需要注意的事情

线程池使用FutureTask时,如果把拒绝策略设置为 DiscardPolicy 和 DiscardOldestPolicy,并且在被拒绝的任务的Future对象上调用了无参get方法,那么调用线程会一直被阻塞.

如上面的代码,如果把CallerRunsPolicy替换成 DiscardPolicy 或 DiscardOldestPolicy ,就会导致任务一直被阻塞,一直无法取到future.get()的值.

[pp6qEJ1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/595b05c12ea64587a07dbd7fddce1e3c.png