重识 java 线程池

作者: __Y_Q | 来源:发表于2021-02-03 00:16 被阅读0次

一. 什么是线程池

线程池就是提前创建若干个线程, 如果有任务需要处理, 线程池里的线程就会处理任务. 处理完之后线程并不会被销毁, 而是等待下一个任务.

二. 为什么要使用线程池

在开发中, 几乎所有需要异步或并发执行任务的程序都可以使用线程池. 合理使用线程池可以带来以下 3 个好处

降低资源消耗. 通过重复利用已创建的线程降低线程创建和销毁造成的消耗.
可以控制最大并发数. 避免大量线程之间因相互抢占系统资源而导致的阻塞现象.
提高线程的可管理性. 线程是稀缺资源, 如果无限制的创建, 不仅会消耗系统资源, 还会降低系统的稳定性. 使用线程池可以进行统一分配, 调优和监控.

三. Executor

在Java 5之后引入了一堆新的启动, 调度和管理线程的API. Executor 框架便是 Java 5 中引入的, 其内部使用了线程池机制, 它在 java.util.cocurrent 包下. 通过该框架来控制线程的启动, 执行和关闭, 可以简化并发编程的操作. 因此, 在 Java 5 之后, 通过使用 Executor 框架来启动线程比使用Thread的start方法更好.

Executor 框架UML 图

Executor
Executor 是一个接口, 它是 Executor 框架的基础, 它将任务的提交与任务的执行分离开来. 内部定义了一个接收 Runnable 对象的方法 executor. 该方法接收一个 Runable 实例, 它用来执行一个任务, 任务即一个实现了 Runnable 接口的类.
ExecutorService
ExecutorService 是一个比 Executor 使用更广泛的子类接口, 其提供了生命周期管理的方法, 以及可跟踪一个或多个异步任务执行状况方法. 所以说 ExecutorService可以说是真正的线程池接口.
ScheduledExecutorService
ScheduledExecutorService 也是一个接口, 继承了 ExecutorService 接口. 内部定义多 4 个带周期执行功能的方法.
ScheduledThreadPoolExecutor
ScheduledThreadPoolExecutor 继承了 ScheduledExecutorService 接口, 并继承了 ThreadPoolExecutor 类. 可以在给定的延迟后运行命令, 或者定期执行命令. ScheduledExecutorService 比 Timer 更灵活, 功能也比较强大.
AbstractExecutorService
AbstractExecutorService 是一个抽象类, 实现了 ExecutorService 接口中的大部分方法.
ThreadPoolExecutor
ThreadPoolExecutor 继承自抽象类 AbstractExecutorService. 是线程池的核心实现类. 它的构造方法提供了一系列参数来配置线程池, 这些参数将会直接影响到线程池的功能特性.

提交任务

Executor 的 execute() 方法用于提交不需要返回值的任务, 所以无法判断任务是否被线程执行成功.
ExecutorService 的 submit() 方法用于提交需要有返回值的任务. 线程池会返回一个 Future 类型的对象, 可以调用 isDone() 方法查询 Future 是否已经完成. 当任务完成时, 它有一个结果. 可以调用 get() 方法来获取该结果. 也可以不用 isDone() 进行检查就直接调用 get() 获取结果, 在这种情况下, get() 将阻塞当前线程, 直至结果准备就绪. 还可以取消任务的执行, Future 提供了 cancel() 方法用来取消执行 pending 中的任务.

关闭线程池

ExecutorService 的 shutdown() 或者 shutdownNow() 方法用来关闭线程池. 它们的原理是遍历线程池中的工作线程, 然后逐个调用线程的 interrupt() 方法来中断线程, 所以无法响应中断的任务可能永远无法停止. 但是他们存在一定的区别.
- shutdownNow() 首先将线程池的状态设置为 STOP, 然后尝试停止所有的正在执行或者暂停任务的线程, 并返回等待执行任务的列表.
- shutdown() 只是将线程池的状态设置成 SHUTDOWN 状态. 然后中断所有没有正在执行任务的线程.
只要调用了这两个方法中的任意一个, ExecutorService.isShutdown() 方法就会返回 true. 当所有的任务都已关闭后, 才表示线程池关闭成功, 这时调用 ExecutorService.isTerminated() 方法会返回 true. 至于应该调用哪一种方法来关闭线程池, 应该由提交到线程池的任务特性决定, 通常调用 ExecutorService.shutdown() 来平滑的关闭线程池, 如果任务不一定要执行完, 则可以使用 ExecutorService.shutdownNow()

了解了如何提交任务到线程池与如何关闭线程池后, 那么接下来就来看一下 ThreadPoolExecutor 的构造函数中配置线程池的参数都有什么意义吧.

四. ThreadPoolExecutor

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

int corePoolSize
线程池的核心线程数. 当提交一个任务时, 如果线程池中的线程数量未达到核心线程的数量, 那么会直接创建一个线程作为核心线程来执行任务. 如果线程中的线程数量已经达到或者超过核心线程的数量, 那么会将任务放入到阻塞队列 workQueue 中排队等待执行.

如果执行了 ThreadPoolExecutor 的 prestartAllCoreThreads() 方法, 线程池会提前创建并启动所有的核心线程.

默认情况下, 核心线程会在线程池中一直存活, 即使它们处于闲置状态. 如果将 ThreadPoolExecutor 的 allowCoreThreadTimeOut 属性设置为 true, 那么闲置的核心线程在等待新任务到来时会有超时策略, 这个时间间隔由 keepAliveTime 参数指定. 当等待时间超过 keepAliveTime 所指定的时间后, 核心线程就会被终止.
int maximumPoolSize
线程池所能容纳的最大线程数, 如果当前阻塞队列满了, 且又继续提交任务, 则会创建新的线程直接执行新添加的任务, 前提是当前线程池中线程数小于 maximumPoolSize.
long keepAliveTime
非核心线程的闲置时的超时时长. 超过这个时长, 非核心线程就会被回收. 当 ThreadPoolExecutor 的 allowCoreThreadTimeOut 属性设置为 true 的时候, keepAliveTime 同样会作用于核心线程.
TimeUnit unit
用于指定 keepAliveTime 参数的时间单位, 是一个枚举, 常用的有 TimeUnit.MILLISECONDS(毫秒), TimeUnit.SECONDS(秒) 以及 TimeUnit.MINUTES(分钟) 等 .
BlockingQueue<Runnable> workQueue
线程池中的阻塞任务队列. 通过线程池的 execute 方法提交的 Runnable 对象会存储在这个对列中.

上一章的阻塞队列中有说过阻塞队列分为有界与无界的, 在这里我们尽量使用有界队列, 如果使用的是无界队列可能会耗尽系统资源, 及时使用有界队列, 也要尽量控制队列的大小在一个合适的范围.
ThreadFactory threadFactory
线程工厂(可缺省), 通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名.
默认的线程命名规则为 pool-数字-thread-数字
RejectedExecutionHandler handler
线程池拒绝策略(可缺省), 当阻塞队列满了, 并且当前线程池内的线程数量大于 maximumPoolSize 最大线程数量, 如果继续提交任务, 就必须采取一种策略处理该任务. 线程池默认提供了 4 种策略.
AbortPolicy: 直接抛出异常. 线程池中默认的拒绝策略.
DiscardOldestPolicy : 直接丢弃阻塞队列中最老的任务, 也就是最前面的任务, 并执行当前任务.
CallerRunsPolicy: 提交任务所在的线程来执行这个要提交的任务.
DiscardPolicy: 直接丢弃最新的任务, 也就是最后面的.
也可以根据场景来实现 RejectedExecutionHandler 接口, 自定义拒绝策略, 比如记录日志或者持久化存储被拒绝的任务.

根据以上参数理解, 可以总结出当有新任务需要处理的时候 ThreadPoolExecuto 执行任务的流程大致如下

先看线程池中的线程数量是否小于核心线程 corePoolSize 的数量
- 小于核心线程数, 那么会直接启动一个核心线程来执行任务.
大于或等于核心线程数, 接着看任务队列 workQueue是否满了.
- 任务队列未满, 那么就直接插入到任务队列中等待执行
任务队列 workQueue 满了, 最后看线程池中的线程数量是否小于线程池最大线程数 maximumPoolSize.
- 小于线程池最大线程数, 立刻启动一个非核心线程来执行任务.
大于或等于线程池最大线程数, 则执行拒绝策略.

ThreadPoolExecuto 执行任务流程图

下面以一个示例来使用一下 ThreadPoolExecuto

class MyTask implements Runnable {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "正在执行。。。");
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class TestType5 {
    public static void main(String[] args) throws Exception {
        //定义一个容量为 2 的有界阻塞队列
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);
        //创建一个线程池, 核心线程 3 个, 最大线程数 5 个, 60 秒钟超时
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 5, 60, TimeUnit.SECONDS, queue);
        Runnable r1= new MyTask();
        Runnable r2= new MyTask();
        Runnable r3= new MyTask();
        Runnable r4= new MyTask();
        Runnable r5= new MyTask();
        Runnable r6= new MyTask();
        Runnable r7= new MyTask();
        Runnable r8= new MyTask();

        threadPoolExecutor.execute(r1);
        threadPoolExecutor.execute(r2);
        threadPoolExecutor.execute(r3);
        threadPoolExecutor.execute(r4);
        threadPoolExecutor.execute(r5);
        threadPoolExecutor.execute(r6);
        threadPoolExecutor.execute(r7);
        //threadPoolExecutor.execute(r8);
        threadPoolExecutor.shutdown();
    }
}

输出结果为:

pool-1-thread-1正在执行。。。
pool-1-thread-2正在执行。。。
pool-1-thread-3正在执行。。。
pool-1-thread-4正在执行。。。
pool-1-thread-5正在执行。。。
pool-1-thread-1正在执行。。。
pool-1-thread-5正在执行。。。

可是, 如果把 r8 也执行了, 输出结果就会变成下面这样

pool-1-thread-1正在执行。。。
pool-1-thread-3正在执行。。。
pool-1-thread-4正在执行。。。
pool-1-thread-2正在执行。。。
pool-1-thread-5正在执行。。。

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task org.study.MyTask@6f94fa3e rejected from java.util.concurrent.ThreadPoolExecutor@5e481248
[Running, pool size = 5, active threads = 5, queued tasks = 2, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    at org.study.TestType5.main(TestType5.java:42)

pool-1-thread-2正在执行。。。
pool-1-thread-3正在执行。。。

可以看到执行了拒绝策略, 抛出了默认的异常. 为什么呢? 我们来屡屡. 一共创建了 8 个Runnable, 分别是 r1 r2 r3 ... r8 .
创建的线程池核心线程为 3 个, 最大线程总数为 5 个. 阻塞队列长度为 2.

r1, r2, r3 这三个分别在开始执行时发现当前线程池内线程总数(0)小于核心线程总数(3), 那么就会启动三个核心线程来执行r1 - r3.
执行 r4 的时候, 发现线程池内的线程总数已经等于了核心线程数, 那么看到任务队列未满, 就放入到阻塞队列中.
r5的执行与 r4 流程相同., r5 执行完后, 此时线程池中线程数量为 3, 阻塞队列也满了.
执行 r6 的时候发现当前线程池内的线程总数等于核心线程, 并且阻塞队列也满了, 最后拿当前线程池中的线程总数(3)与我们设置的线程池最大线程数(5)做对比, 发现小于我们设置的最大线程数, 那么就新启动一个线程来执行 r6.
r7 的执行与 r6 相同, 也是新启动一个线程来执行 r7. 这时候, 核心线程总数为 3, 阻塞队列满了, 线程池中的线程总数为 5 个.
当要执行 r8 的时候, 发现核心线程数满了, 阻塞队列也满了, 同时线程池中的线程总数也达标了, 那么就会执行拒绝策略, 抛出默认的异常.
目前正在执行的任务有 r1, r2, r3, r6, r7, 阻塞队列中的任务是 r4, r5. 当 r1, r2, r3, r6, r7 中任何一个任务执行完了, 就会从阻塞队列中取出任务来执行, 所以最后又打印了两个 log. 执行的就是阻塞队列中的 r4 与 r5.

看到这里的同学是不是有点蒙圈了, 通过 ThreadPoolExecutor 配置的线程池, 那些参数要如何填写呢? 核心线程要分配几个呢? 怎么配置线程池算是合理呢? 其实想要合理的配置线程池, 需要先分析任务的特性. 而任务的特性可以从以下几个角度来分析

任务的性质: CPU 密集型任务, IO 密集型任务和混合型任务.
任务的优先级: 高, 中, 低.
任务的执行时间: 长, 中, 短
任务的依赖性: 是否依赖其他系统资源, 如数据库连接.

CPU 密集型任务的特点是要进行大量的计算, 消耗 CPU 资源, 比如计算圆周率, 对视频进行高清解码等, 全靠 CPU 的运算能力.
IO 密集型是指设计到网络, 磁盘 IO 的任务都是 IO 密集型任务, 这类任务特点是 CPU 消耗很少, 任务的大部分时间都在等待 IO 操作完成 (因为 IO 的速度远远低于 CPU 和内存的速度).

性质不能同的任务可以用不同规模的线程池分开处理.

CPU 密集型的任务应配置尽可能小的线程, 因为CPU密集型任务使得CPU使用率很高, 若开过多的线程数, 只能增加上下文切换的次数, 因此会带来额外的开销. 如配置 CPU +1 个线程的线程池.
IO密集型任务CPU使用率并不高, 因此可以让 CPU 在等待 IO 的时候去处理别的任务, 充分利用 CPU, 则应该配置尽可能多的线程, 如 2*CPU.
混合型任务, 如果可以拆分, 将其拆分成一个 CPU 密集型任务和一个 IO 密集型任务, 只要这两个任务执行的时间相差不是太大, 那么分解后执行的吞吐量将高于串行执行的吞吐量. 如果这两个任务执行的时间相差太大, 则没有必要进行分解.
优先级不同的任务可以使用优先级队列 PriorityBlockingQueue 来处理, 它可以让优先级高的任务优先执行.
执行时间不同的任务可以交给不同规模的线程池来处理, 或者可以使用优先级队列, 让执行时间短的优先执行.
建议使用有界队列, 有界队列能增加系统的稳定性和预警能力, 可以根据需要设置. 如果设置为无界队列, 那么线程池的队列内的任务就会越来越多, 有可能会撑满内存, 导致整个系统不可用等各种异常.

Ps: 通过 Runtime.getRuntime().availableProcessors() 方法获得当前设备的 CPU 个数
Ps: CPU+1 是为了防止页缺失.也叫硬中断. 当计算密集型的线程偶尔由于缺失故障或者其他原因而暂停时, 这个额外的线程也能确保 CPU 的时钟周期不会被浪费.

这里了解了通过 ThreadPoolExecutor 实现线程池方式, 而在 Java 中, 使用 Executors 类创建线程池也是一种常用的创建线程池的方式. 下面再来了解一下 Executors类.

五. Executors

Executors 类, 提供了一系列工厂方法用于创建线程池，返回的线程池都实现了 ExecutorService 接口.
通过 Executors 类创建的 4 类线程池都是直接或间接的通过配置ThreadPoolExecutor 来实现自己的功能特性, 下面分别来了解一下通过 Executors 创建的这 4 类线程池.

1. newCachedThreadPool 缓存型线程池

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

通过 newCachedThreadPool 创建的是一种线程数量不定的线程池, 它只有又非核心线程, 并且设置了最大线程池数为 Integer.MAX_VALUE. 由于 Integer.MAX_VALUE 是一个很大的数, 实际上就相当于最大线程数可以任意大.
当线程池中的线程都处于活动状态时, 线程池会创建新的线程来处理新任务, 否则就会利用空闲的线程来处理新任务.
线程池内的空闲线程超时时间为 60 秒, 超过 60 秒闲置的线程就会被回收.
线程池内的任务队列其实相当于一个空集合, 这将导致任何任务都会被立即执行, 因为在这种场景下 SynchronousQueue 阻塞队列是无法插入任务的. 详情见上一章的
「java 中的阻塞队列」

从 newCachedThreadPool 创建出的线程池特性来看, 这类线程池比较适合执行大量的耗时较少的任务, 当整个线程池都处于闲置状态时, 线程池中的线程都会因为超时而被停止, 这时候线程池内部是没有任何线程的, 几乎不占用任何系统资源.

2. newFixedThreadPool 全核心型线程池

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

通过 newFixedThreadPool 创建的是一种线程数量固定的线程池.
当线程池内的线程处于空闲状态时, 并不会被回收, 除非线程池被关闭了.
当所有的线程都处于活动状态时, 新任务都会处于等待状态, 直到有线程空闲出来.
由于只有核心线程并且核心线程不会被回收, 这意味着它能够更加快速的响应外接的请求.

这类型的线程池多数针对一些很稳定很固定的正规并发线程, 多用于服务器.

3. newScheduledThreadPool 调度型线程池

    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }
    // 上面的 Executor 框架 UML 图 中 ScheduledThreadPoolExecutor  继承了 ThreadPoolExecutor 类.
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue());
    }

核心线程数固定, 而非核心线程池数没有限制.
当非核心线程闲置时会被立即回收.
这个池子里的线程可以按 schedule 依次 delay 执行, 或周期执行.

这类型的线程池主要用于执行定时任务和具有固定周期的重复任务.

4. newSingleThreadExecutor 单例型线程池

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

线程池内部只有一个核心线程, 它确保所有的任务都在同一个线程中按顺序执行.
newSingleThreadExecutor 的意义在于统一所有的外界任务到一个线程中, 这使的这些任务之间不需要处理线程同步的问题.

下面只挑出两个具有代表性的来看一下是如何使用的, newCachedThreadPool 与 newScheduledThreadPool

带有返回值的任务提交与 newCachedThreadPool 的使用.

class MyTask implements Callable<String>{

    private int id;

    public MyTask(int id) {
        this.id = id;
    }

    @Override
    public String call() throws Exception {
        System.out.println("id:"+ id + " - threadName:"+Thread.currentThread().getName() + "调用 call 方法");
        //这里返回的结果, 会被 Future 的 get 方法得到.
        return "任务返回结果为:" +  id +" - "+ Thread.currentThread().getName();
    }
}

public class TestType5 {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newCachedThreadPool();
        //存储返回的结果
        ArrayList<Future<String>> list = new ArrayList<>();

        //提交 10 个任务, 并将返回的 future 存储
        for (int i = 0; i < 10; i++) {
            Future<String> future = pool.submit(new MyTask(i));
            list.add(future);
        }

        for (Future<String> fs : list) {
            //Future 返回如果没有完成, 则一直循环,
            while (!fs.isDone());
            System.out.println(fs.get());
            pool.shutdown();
        }

    }
}

输出结果

id:0 - threadName:pool-1-thread-1调用 call 方法
id:3 - threadName:pool-1-thread-4调用 call 方法
id:2 - threadName:pool-1-thread-3调用 call 方法
id:1 - threadName:pool-1-thread-2调用 call 方法
id:5 - threadName:pool-1-thread-6调用 call 方法
id:4 - threadName:pool-1-thread-5调用 call 方法
id:6 - threadName:pool-1-thread-7调用 call 方法
id:7 - threadName:pool-1-thread-6调用 call 方法
id:9 - threadName:pool-1-thread-5调用 call 方法
id:8 - threadName:pool-1-thread-7调用 call 方法
任务返回结果为:0 - pool-1-thread-1
任务返回结果为:1 - pool-1-thread-2
任务返回结果为:2 - pool-1-thread-3
任务返回结果为:3 - pool-1-thread-4
任务返回结果为:4 - pool-1-thread-5
任务返回结果为:5 - pool-1-thread-6
任务返回结果为:6 - pool-1-thread-7
任务返回结果为:7 - pool-1-thread-6
任务返回结果为:8 - pool-1-thread-7
任务返回结果为:9 - pool-1-thread-5

newScheduledThreadPool 调度型线程池的使用

class ThreadPoolUtil implements Runnable{

    private Integer index;

    public ThreadPoolUtil(Integer index) {
        this.index = index;
    }

    @Override
    public void run() {
        try {
            System.out.println(index+"开始处理线程！");
            Thread.sleep(5000);
            System.out.println("线程标识是："+this.toString());
            System.out.println(index+"处理结束！");
        }
        catch(InterruptedException e) {
            e.printStackTrace();
        }
    }

}

public class TestType5 {
    public static void main(String[] args) throws Exception {
        //核心线程为 2. 一次执行 2 个任务.剩下的放入到队列
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
        for (int i = 0; i < 4; i++) {
            //延迟 2 秒执行
            pool.schedule(new ThreadPoolUtil(i), 2 , TimeUnit.SECONDS);
        }
        pool.shutdown();
    }
}

输出结果

0开始处理线程！
1开始处理线程！
线程标识是：org.study.ThreadPoolUtil@30f68028
线程标识是：org.study.ThreadPoolUtil@37c3a03e
0处理结束！
1处理结束！
2开始处理线程！
3开始处理线程！
线程标识是：org.study.ThreadPoolUtil@34c912ee
线程标识是：org.study.ThreadPoolUtil@390458b6
2处理结束！
3处理结束！

线程池到这里就结束了, 我们在 Android 中有时也会需要用到线程池. 所以了解类之间的关系以及线程池的优化也是必不可少的.

重识 java 线程池

一. 什么是线程池

二. 为什么要使用线程池

三. Executor

提交任务

关闭线程池

四. ThreadPoolExecutor

五. Executors

1. newCachedThreadPool 缓存型线程池

2. newFixedThreadPool 全核心型线程池

3. newScheduledThreadPool 调度型线程池

4. newSingleThreadExecutor 单例型线程池

带有返回值的任务提交与 newCachedThreadPool 的使用.

newScheduledThreadPool 调度型线程池的使用

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Android开发

程序员

Android开发

Android技术知识

Android开发经验谈