003. 线程池应用及实现原理剖析

in 编程
关注公众号【好便宜】( ID:haopianyi222 ),领红包啦~
阿里云,国内最大的云服务商,注册就送数千元优惠券:https://t.cn/AiQe5A0g
腾讯云,良心云,价格优惠: https://t.cn/AieHwwKl
搬瓦工,CN2 GIA 优质线路,搭梯子、海外建站推荐: https://t.cn/AieHwfX9

1. 为什么使用线程池?线程池是不是越多越好?


2. 线程池原理 - 概念


img

3. 线程池 API - 接口定义和实现类


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

4. 线程池 API - 方法定义


ExecutorService

ExecutorService

ScheduledExecutorService

ScheduledExecutorService

5. 线程池 API - Executors 工具类


方法 描述
newFixedThreadPool(int nThreads) 创建一个固定大小、任务队列容量无界的线程池。核心线程数=最大线程数。
newCachedThreadPool() 创建的是一个大小无界的缓冲线程池。它的任务队列是一个同步队列。任务加入到池中,如果池中有空闲线程,则用空闲线程执行,如无则创建新线程执行。池中的线程空闲超过60秒,将被销毁释放。线程数随任务的多少变化。适用于执行耗时较小的异步任务。池的核心线程数=0,最大线程数=Integer.MAX_VALUE。
newSingleThreadExecutor() 只有一个线程来执行无界任务队列的单一线程池。该线程池确保任务按加入的顺序一个一个依次执行。当唯一的线程因任务异常中止时,将创建一个新的线程来继续执行后续的任务。与 newFixedThreadPool(1) 的却别在于,单一线程池的池大小在 newSingleThreadExecutor 方法中硬编码,不能再改变的。
newScheduledThreadPool(int corePoolSize) 能定时执行任务的线程池。该池的核心线程数由参数指定,最大线程数=Integer.MAX_VALUE。

6. 线程池原理 - 任务 execute 过程


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

任务 execute 过程

7. 线程数量


如何确定合适数量的线程?

8. 代码演示


package com.study.hc.thread.chapter1.thread;

import sun.nio.ch.ThreadPool;

import java.util.List;
import java.util.concurrent.*;

/**
 * 线程池的使用
 */
public class Demo9 {

    /**
     * 测试,提交15个执行时间需要3秒的任务,看线程池的状况
     * @param threadPoolExecutor 传入不同的线程池,看不同的结果
     * @throws Exception
     */
    private void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception {
        // 测试,提交15个执行时间需要3秒的任务,看超过大小的2个,对应的处理情况
        for (int i = 0; i < 15; i ++) {
            int n = i;
            threadPoolExecutor.submit(() -> {
                try {
                    System.out.println("开始执行:" + n);
                    Thread.sleep(3000);
                    System.out.println("执行结束:" + n);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            System.out.println("任务提交成功:" + n);
        }
        // 查看线程数量,查看队列等待数量
        Thread.sleep(500);
        System.out.println("当前线程池线程数量为:" + threadPoolExecutor.getPoolSize());
        System.out.println("当前线程池等待的数量为:" + threadPoolExecutor.getQueue().size());
        // 等待15秒,查看线程数量和等待数量(理论上,超出核心线程数量的线程自动销毁)
        Thread.sleep(15000);
        System.out.println("当前线程池线程数量为:" + threadPoolExecutor.getPoolSize());
        System.out.println("当前线程池等待的数量为:" + threadPoolExecutor.getQueue().size());
    }

    /**
     * 1. 线程池信息:核心线程数量5,最大数量10,无界队列,超出核心线程数量的线程存活时间为:5秒,指定拒绝策略
     * @throws Exception
     */
    private void threadPoolExecutorTest1() throws Exception {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10,
                5, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
        testCommon(threadPoolExecutor);
        // 预计结果:线程池线程数量为5,超出数量的任务,其他的进入队列中等待被执行
    }

    /**
     * 2. 线程池信息:核心线程数量5,最大数量10,队列大小3,超出核心线程数量的线程存活时间为:5秒,指定拒绝策略
     * @throws Exception
     */
    private void threadPoolExecutorTest2() throws Exception {
        // 创建一个 核心线程数量5,最大数量10,等待队列最大是3的线程池,也就是最大容纳13个任务
        // 默认的策略是抛出 RejectedExecutionException 异常,java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(3), (r, executor) -> System.out.println("有任务被拒绝执行了"));
        testCommon(threadPoolExecutor);
        // 预计结果:
        // 1. 5个任务直接分配线程开始执行
        // 2. 3个任务进入等待队列
        // 3. 队列不够用,临时加开5个线程来执行任务(5秒没活干就销毁)
        // 4. 队列和线程池都满了,剩下2个任务,没资源了,被拒绝执行
        // 5. 任务执行,5秒后,如果无任务可执行,销毁临时创建的5个线程
    }

    /**
     * 3. 线程池信息:核心线程数量5,最大数量5,无界队列,超出核心线程数量的线程存活时间:0秒
     * @throws Exception
     */
    private void threadPoolExecutorTest3() throws Exception {
        // 和 Executors.newFixedThreadPool(int nThreads) 一样的
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>());
        testCommon(threadPoolExecutor);
        // 预计结果:线程池线程数量为5,超出数量的任务,其他的进入队列中等待被执行
    }

    /**
     * 4. 线程池信息:核心线程数量0,最大数量 Integer.MAX_VALUE,SynchronousQueue 队列,超出核心线程数量的线程存活时间:60秒
     * @throws Exception
     */
    private void threadPoolExecutorTest4() throws Exception {
        /*
        SynchronousQueue,实际上它不是一个真正的队列,因为它不会为队列中元素维护存储空间,与其他队列不同的是,它维护一组线程,这些线程在等待着把元素加入或移出队列
        在使用 SynchronousQueue 作为工作队列的前提下,客户端代码向线程池提交任务时,
        而线程中又没有空闲的线程能够从 SynchronousQueue 队列中取出一个任务
        那么相应的 offer 方法调用就会失败(即任务没有被存入工作队列)
        此时,ThreadPoolExecutor 会新建一个新的工作线程用于对这个入队列失败的任务进行处理(假设此时线程池大小还未达到其最大线程池大小 maximumPoolSize)
         */

        // 和 Executors.newCachedThreadPool() 一样的
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
                new SynchronousQueue<>());
        testCommon(threadPoolExecutor);
        // 预计结果:
        // 1. 线程池线程数量为15,超出数量的任务,其他的进入队列中等待被执行
        // 2. 所有任务执行结束,60秒后,如果无任务可执行,所有线程全部被销毁,池的大小恢复为0
        Thread.sleep(60000);
        System.out.println("60秒后,再看看线程池中的数量:" + threadPoolExecutor.getPoolSize());
    }

    /**
     * 5. 定时执行线程池信息:3秒后执行,一次性任务,到点就执行
     * 核心线程数量是5,最大数量 Integer.MAX_VALUE,DelayedWorkQueue 延时队列,超出核心线程数量的线程存活时间:0秒
     */
    private void threadPoolExecutorTest5() {
        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        scheduledThreadPoolExecutor.schedule(() -> {
            System.out.println("任务被执行,现在时间:" + System.currentTimeMillis());
        }, 3, TimeUnit.SECONDS);
        System.out.println("定时任务,提交成功,时间是:" + System.currentTimeMillis() + ",当前线程池数量:" + scheduledThreadPoolExecutor.getPoolSize());
        // 预计结果:任务在3秒后被执行一次
    }

    /**
     * 6. 定时执行线程池信息:线程固定数量5
     * 核心线程数量5,最大数量 Integer.MAX_VALUE,DelayedWorkQueue 延时队列,超出核心线程数量的线程存活时间:0秒
     */
    private void threadPoolExecutorTest6() {
        ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        // 周期性执行一个任务,线程池提供了两周调度方式
        // 测试场景:提交的任务需要3秒才能执行完毕
        // 效果1:提交后,2秒后开始第一次执行,之后每隔一秒,固定执行一次(如果发现上次执行还未完毕,则等待完毕,完毕后立刻执行)
        // 也就是说,这个代码中,3秒执行一次(计算方式:每次执行3秒,间隔时间1秒,执行结束后马上开始下一次执行,无需等待)
        threadPoolExecutor.scheduleAtFixedRate(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("任务-1被执行,现在时间:" + System.currentTimeMillis());
        }, 2000, 1000, TimeUnit.MILLISECONDS);

        // 效果2:提交后,2秒后开始第一次执行,之后每隔1秒,固定执行一次(如果发现上次执行还未完毕,则等待完毕,等上一次执行完毕后再开始计时,等待1秒)
        // 也就是说这个代码的效果看到的是:4秒执行一次(计算方式:每次执行3秒,间隔时间1秒,执行完以后再等待1秒,所以是3+1)
        threadPoolExecutor.scheduleWithFixedDelay(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("任务-2被执行,现在时间:" + System.currentTimeMillis());
        }, 2000, 1000, TimeUnit.MILLISECONDS);
    }

    /**
     * 7. 终止线程:线程池信息:核心线程数量5,最大数量10,队列大小3,超出核心线程数量的线程存活时间:5秒,指定拒绝策略的
     */
    private void threadPoolExecutorTest7() throws InterruptedException {
        // 创建一个 核心线程数量5,最大数量10,等待队列最大是3的线程池,也就是最大容纳13个任务
        // 默认的策略是抛出 RejectedExecutionException 异常,java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(3), (r, executor) -> {
                    System.out.println("有任务被拒绝执行了");
                });
        // 测试:提交15个执行时间需要3秒的任务,看超过大小的2个,对应的处理情况
        for (int i = 0; i < 15; i ++) {
            int n = i;
            threadPoolExecutor.submit(() -> {
                try {
                    System.out.println("开始执行:" + n);
                    Thread.sleep(3000);
                    System.out.println("执行结束: " + n);
                } catch (InterruptedException e) {
                    System.out.println("异常:" + e.getMessage());
                }
            });
            System.out.println("任务提交成功:" + n);
        }

        // 1秒后终止线程池
        Thread.sleep(1000);
        threadPoolExecutor.shutdown();
        // 再次提交提示失败
        threadPoolExecutor.submit(() -> {
            System.out.println("追加一个任务");
        });

        // 结果分析:
        // 1. 10个任务被执行,3个任务进入等待队列,2个任务被拒绝执行
        // 2. 调用 shutdown 后,不接收新的任务,等待13任务执行结束
        // 3. 追加的任务在线程池关闭后,无法再提交,会被拒绝执行
    }

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

        // 1秒后终止线程池
        Thread.sleep(1000);
        List<Runnable> shutdownNow = threadPoolExecutor.shutdownNow();
        // 再次提交提示失败
        threadPoolExecutor.submit(() -> {
            System.out.println("追加一个任务");
        });
        System.out.println("未结束的任务有:" + shutdownNow.size());

        // 结果分析:
        // 1. 10个任务被执行,3个任务进入等待队列,2个任务被拒绝执行
        // 2. 调用 shutdownNow 后,队列中的三个线程不再执行,10个线程被终止

    }

    public static void main(String[] args) throws Exception {
//        new Demo9().threadPoolExecutorTest1();
//        new Demo9().threadPoolExecutorTest2();
//        new Demo9().threadPoolExecutorTest3();
//        new Demo9().threadPoolExecutorTest4();
//        new Demo9().threadPoolExecutorTest5();
//        new Demo9().threadPoolExecutorTest6();
//        new Demo9().threadPoolExecutorTest7();
        new Demo9().threadPoolExecutorTest8();
    }
}
关注公众号【好便宜】( ID:haopianyi222 ),领红包啦~
阿里云,国内最大的云服务商,注册就送数千元优惠券:https://t.cn/AiQe5A0g
腾讯云,良心云,价格优惠: https://t.cn/AieHwwKl
搬瓦工,CN2 GIA 优质线路,搭梯子、海外建站推荐: https://t.cn/AieHwfX9
扫一扫关注公众号添加购物返利助手,领红包
Comments are closed.

推荐使用阿里云服务器

超多优惠券

服务器最低一折,一年不到100!

朕已阅去看看