JAVA 线程池

概述

• JAVA 中线程创建和销毁需要时间，如果创建和销毁的时间比执行时间还多就不划算
• 过多的线程占用系统内存
• 频繁切换上下文影响系统性能

线程池原理

• 线程池管理器
  用于创建并管理线程池，包含创建和销毁线程池，添加新的任务。

• 工作线程
  线程池中的线程，循环执行任务，没有任务时处理等待状态

• 任务接口
  每个任务必须实现的接口，以供工作线程调度任务的执行，主要规定的任务的入口，任务执行完后的收尾工作，任务的执行状态等

• 任务队列
  用于存放没有处理的任务，提供一种缓冲机制。

• 执行过程

  1. 是否达到核心线程数量？没达到，创建一个工作线程来执行任务
  2. 任务队列是否已满？没满，则将新提交的任务存储在任务队列中
  3. 是否达到线程池最大数量？没达到，创建新的工作线程来执行任务
  4. 最后执行拒绝策略来处理这个任务
  image.png

线程池接口 API

类型	名称	描述
接口	Executor	最上层的接口，定义了执行任务的发放 execute
接口	ExcutorService	继承了 Executor 接口，拓展了Callable,Future,关闭方法
接口	ScheduledExcutorService	继承了 ExcutorService 接口，增加了定时任务相关方法
实现类	ThreadPoolExecutor	基础，标准的线程实现类
实现类	ScheduledThreadPoolExecutor	继承了 ThreadPoolExecutor ，实现了ScheduledExcutorService 中定时任务相关方法

代码实现

• 线程池信息： 核心线程数量5，最大数量10，无界队列

  结果： 线程池线程数量为：5,超出数量的任务，存入队列中等待被执行

public class Demo8 {

    private void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception{
        for (int i=0; i<15; i++) {
            threadPoolExecutor.submit(new MyTask(i));
            System.out.println("任务提交成功 :" + i);
        }

        // 查看线程数量，查看队列等待数量
        Thread.sleep(500L);
        System.out.println("当前线程池线程数量为：" + threadPoolExecutor.getPoolSize());
        System.out.println("当前线程池等待的数量为：" + threadPoolExecutor.getQueue().size());
        // 等待15秒，查看线程数量和队列数量（理论上，会被超出核心线程数量的线程自动销毁）
        Thread.sleep(15000L);
        System.out.println("当前线程池线程数量为：" + threadPoolExecutor.getPoolSize());
        System.out.println("当前线程池等待的数量为：" + threadPoolExecutor.getQueue().size());
    }

    private void threadPoolExecutorTest1() throws Exception{
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5,10,5,
                TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());

        testCommon(threadPoolExecutor);
    }

    public static void main(String[] args) throws Exception{
        new Demo8().threadPoolExecutorTest1();
    }
}

• 线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的

  结果： 5个线程直接分配任务开始执行，3个任务存入队列，队列够用，临时加开5个线程（5秒没活干就销毁），仍有2个任务没有资源去执行，拒绝执行。

private void threadPoolExecutorTest2() throws Exception{
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5,
                10,
                5,
                TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(3),
                new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.err.println("任务被拒绝");
            }
        });

        testCommon(threadPoolExecutor);
    }

• 核心线程数量5，最大数量10，无界队列

  结果：线程池线程数量为：5,超出数量的任务，存入队列中等待被执行

private void threadPoolExecutorTest3() throws Exception{
        // 和Executors.newFixedThreadPool(int nThreads)一样的
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5,
                5,
                0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingDeque<Runnable>());

        testCommon(threadPoolExecutor);
    }

• 缓存线程池，适合任务大小无法预估的情况，线程池有就复用，没有就加开。

  结果：创建任务数相同的线程，执行任务，60
  秒后如果没有任务，所有线程销毁

private void threadPoolExecutorTest4() throws Exception{
        // 相当于 Executors.newCachedThreadPool();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(0,
                Integer.MAX_VALUE,
                60L,
                TimeUnit.SECONDS,
                new SynchronousQueue<>());

        testCommon(threadPoolExecutor);

        Thread.sleep(60000L);
        System.out.println("60秒后，再看线程池中的数量：" + threadPoolExecutor.getPoolSize());
    }

• 延时任务，核心线程5个，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒

  结果：3秒执行一次

 private void threadPoolExecutorTest5() throws Exception{
        // 和Executors.newScheduledThreadPool()一样的
        ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        threadPoolExecutor.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("任务被执行,当前时间为:" + System.currentTimeMillis());
            }
        },3000,TimeUnit.MILLISECONDS);

        System.out.println("定时任务执行成功，时间为:" + System.currentTimeMillis() + "当前线程池中线程数量:" + threadPoolExecutor.getPoolSize());
    }

• 周期执行任务，
  方式1: 任务执行时间超出周期时间，立即执行
  方式2: 任务执行时间超出周期时间，等待周期时间后执行

 private void threadPoolExecutorTest6() throws Exception{
        ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        // 效果1： 提交后，2秒后开始第一次执行，之后每间隔1秒，固定执行一次(如果发现上次执行还未完毕，则等待完毕，完毕后立刻执行)。
        // threadPoolExecutor.scheduleAtFixedRate()
        //效果2：提交后，2秒后开始第一次执行，之后每间隔1秒，固定执行一次(如果发现上次执行还未完毕，则等待完毕，等上一次执行完毕后再开始计时，等待1秒)。
        threadPoolExecutor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(3000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务被执行,当前时间为:" + System.currentTimeMillis());
            }
        },2000,1000,TimeUnit.MILLISECONDS);
    }

• 线程终止
  shutdown方式 不接收新的任务提交，已提交任务会执行完
  shutdownNow方式 所有任务终止

/**
     * 7、 终止线程：线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
     * 
     * @throws Exception
     */
    private void threadPoolExecutorTest7() throws Exception {
        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
        // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.err.println("有任务被拒绝执行了");
                    }
                });
        // 测试： 提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
        for (int i = 0; i < 15; i++) {
            int n = i;
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("开始执行：" + n);
                        Thread.sleep(3000L);
                        System.err.println("执行结束:" + n);
                    } catch (InterruptedException e) {
                        System.out.println("异常：" + e.getMessage());
                    }
                }
            });
            System.out.println("任务提交成功 :" + i);
        }
        // 1秒后终止线程池
        Thread.sleep(1000L);
        threadPoolExecutor.shutdown();
        // 再次提交提示失败
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("追加一个任务");
            }
        });
        // 结果分析
        // 1、 10个任务被执行，3个任务进入队列等待，2个任务被拒绝执行
        // 2、调用shutdown后，不接收新的任务，等待13任务执行结束
        // 3、 追加的任务在线程池关闭后，无法再提交，会被拒绝执行
    }

/**
     * 8、 立刻终止线程：线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
     * 
     * @throws Exception
     */
    private void threadPoolExecutorTest8() throws Exception {
        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
        // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.err.println("有任务被拒绝执行了");
                    }
                });
        // 测试： 提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
        for (int i = 0; i < 15; i++) {
            int n = i;
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("开始执行：" + n);
                        Thread.sleep(3000L);
                        System.err.println("执行结束:" + n);
                    } catch (InterruptedException e) {
                        System.out.println("异常：" + e.getMessage());
                    }
                }
            });
            System.out.println("任务提交成功 :" + i);
        }
        // 1秒后终止线程池
        Thread.sleep(1000L);
        List<Runnable> shutdownNow = threadPoolExecutor.shutdownNow();
        // 再次提交提示失败
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("追加一个任务");
            }
        });
        System.out.println("未结束的任务有：" + shutdownNow.size());

        // 结果分析
        // 1、 10个任务被执行，3个任务进入队列等待，2个任务被拒绝执行
        // 2、调用shutdownnow后，队列中的3个线程不再执行，10个线程被终止
        // 3、 追加的任务在线程池关闭后，无法再提交，会被拒绝执行
    }