【JavaEE初阶 — 多线程】线程池

1. 线程池的原理

1.1 为什么要有线程池

1.2 线程池的构造方法

1.3 线程池的核心参数

1.4 TimeUnit

1.5 工作队列的类型

1.6 工厂设计模式

1.6.1 工厂模式概念

1.6.2 使用工厂模式的好处

1.6.3 使用工厂模式的典型案例

1.6.4 ThreadFatory

1.7 拒绝策略

1.8 线程池工作原理

2. 并发库提供的线程池实现

3. Executors

4. 模拟实现一个固定线程个数的线程池

4.1 代码讲解

4.2 完整代码

1. 线程池的原理

1.1 为什么要有线程池

线程池是一种池化技术，用于预先创建并管理一组线程，避免频繁创建和销毁线程的开销，提高性能和响应速度。

1.2 线程池的构造方法

1.3 线程池的核心参数

线程池的几个关键的配置：核心线程数、最大线程数、空闲存活时间、工作队列、拒绝策略。

参数属性	说明
int corePoolSize	核心线程数，即线程池中始终保持的线程数量
int maximumPoolSize	最大线程数，即线程池中允许的最大线程数量
long keepAliveTime	线程空闲时间，超过这个时间的非核心线程会被销毁
TimeUnit unit	keepAliveTime 的时间单位，是秒，分钟或者其他值
workQueue	任务队列，存放待执行的任务
threadFactory	线程工厂，用于创建新线程
rejectedExecutionHandler	任务拒绝处理器，当任务无法执行时的处理策略

接下来，我们会详细讲解后面的四个参数；

1.4 TimeUnit

TimeUnit 是一个枚举类型的参数，作为 keepAliveTime 的时间单位，是秒，分钟或者其他值：

1.5 工作队列的类型

工作队列是让程序员根据自己的需求，实例化需要的队列，给程序员更大的自由度，以完成更多的功能；
泛型为 Runnable，说明阻塞队列中的元素是一个个 Runnable 任务；
每次通过调用 submit() 方法，往队列中添加任务，线程池在工作的时候，就会往队列中取走素，并且执行 Runnable 中的 run()方法，这是工作队列所起到的效果：

线程池本质上，也是生产者消费者模型，调用 submit() 就是在生产任务，线程池里的线程就是在消费任务；
因此需要以阻塞队列为基础，进行数据交互，所以工作队列的类型为 BlockingQueue；

工作队列类型	说明
SynchronousQueue	不存储任务，直接将任务提交给线程。
LinkedBlockingQueue	链表结构的阻塞队列，大小无限。
ArrayBlockingQueue	数组结构的有界阻塞队列。
PriorityBlockingQueue	带优先级的无界阻塞队列。

选择使用基于数组，链表，或者优先级队列的阻塞队列，指定capacity，指定是否要带有优先级/比较规则即可；

1.6 工厂设计模式

1.6.1 工厂模式概念

工厂模式（Factory Pattern）是一种创建型设计模式；它提供了一种封装对象创建过程的方法
工厂模式通过静态方法，将对象的创建(实例化 & 初始化）和使用分离；
让一个专门的工厂类负责创建对象实例，而不是在代码中直接使用 new 操作符；
每个工厂类中提供多组静态方法，实现不同情况的构造；
工厂模式用于弥补构造方法的缺陷，有助于降低代码的耦合度，提高可维护性和可扩展性。

1.6.2 使用工厂模式的好处

(1) 降低耦合度

工厂模式将对象的创建与使用分离，使得客户端代码不直接依赖具体的类，降低了耦合度。

(2) 提高可扩展性

当需要添加或修改产品类时，只需修改工厂类，而不需要修改客户端代码，提高了系统的可扩展性。

(3) 提高可维护性

通过集中管理对象的创建，提高了代码的可维护性。

(4) 提高代码复用性

工厂类可以被多个客户端代码复用，减少了重复代码。

1.6.3 使用工厂模式的典型案例

(1) 问题描述

我们要定义一个点，可以通过平面直角坐标系来定义，也可以通过极坐标系定义：

通过常规的通过构造方法的方式来构造对象，通过给表示点的类 Point 提供构造方法，来表示一个点：

但是如果构造方法的方法名，和参数的个数和类型相同，则会因为方法重载而报错；

也就是说，如果构造方法的名字是固定的，要想提供不同的版本，就需要通过重载的方式来触发，但是有时候不一定能构成重载；

(2) 解决方法

对于上述问题，我们可以通过工厂模式来解决；

我们不再通过构造方法的方式来构造对象，而是通过提供专门的静态函数，来构造专门的对象，并且返回

用来构造对象的静态方法（makePointByXY & makePointByRA），称为工厂方法；
提供工厂方法的类（PointFactory），就可以称为工厂类；

通过单独的类（PointFactory）提供工厂方法（makePointByXY & makePointByRA），和要构造对象的类（Point）分开，这是更科学的工厂设计模式。

1.6.4 ThreadFatory

Java 说明手册

使用工厂设计模式，最主要的目的，不是提供多种构造模式，而是通过工厂类，来简化初始化操作

一个线程涉及到很多可以设置的属性，如果在线程池中，对这些线程设置多种属性时，我们更希望希望统一使用一样的套路，对这些线程设置属性，这样的做法是比较合适的；

1.7 拒绝策略

AbortPolicy

当任务队列满且没有线程空闲，此时添加任务会直接抛出 RejectedExecutionException 错误，这也是默认的拒绝策略。
适用于必须通知调用者任务未能被执行的场景。

CallerRunsPolicy

当任务队列满且没有线程空闲，此时添加任务由即调用者线程执行。
适用于希望通过减缓任务提交速度来稳定系统的场景。

DiscardOldestPolicy

当任务队列满且没有线程空闲，会删除最早的任务，然后重新提交当前任务。
适用于希望丢弃最旧的任务以保证新的重要任务能够被处理的场景。

DiscardPolicy

直接丢弃当前提交的任务，不会执行任何操作，也不会抛出异常。
适用于对部分任务丢弃没有影响的场景，或系统负载较高时不需要处理所有任务。

自定义拒绝策略

使用拒绝策略

1.8 线程池工作原理

注意：核心线程和非核心线程在线程池中是一样的，并没有特殊的标识区分！图中区分仅为说清楚创建的顺序。

1.默认情况下线程不会预创建，任务提交之后才会创建线程
（不过设置prestartAllCoreThreads可以预创建核心线程）。

2.当核心线程满了之后不会新建线程，而是把任务堆积到工作队列中。

3.如果工作队列放不下了，然后才会新增线程，直至达到最大线程数。

4.如果工作队列满了，然后也已经达到最大线程数了，这时候来任务会执行拒绝策略。

5.如果线程空闲时间，超过空闲存活时间，并且线程线程数是大于核心线程数的，则会销毁线程，直到线程数等于核心线程数（设置allowCoreThreadTimeOut为true可以回收核心线程，默认为false)。

2. 并发库提供的线程池实现

Java 并发库中提供了5种常见的线程池实现，主要通过 Executors 工具类来创建。

FixedThreadPool

创建一个固定数量的线程池

线程池中的线程数是固定的，空闲的线程会被复用。
如果所有线程都在忙，则新任务会放入队列中等待。
适合负载稳定的场景，任务数量确定且不需要动态调整线程数。

CachedThreadPool

一个可以根据需要创建新线程的线程池

线程池的线程数量没有上限，空闲线程会在60秒后被回收；
如果有新任务且没有可用线程，会创建新线程。
适合短期大量并发任务的场景，任务执行时间短且线程数需求变化较大。

SingleThreadExecutor

创建一个只有单个线程的线程池

只有一个线程处理任务，任务会按照提交顺序依次执行。
适用于需要保证任务按顺序执行的场景，或者不需要并发处理任务的情况。

ScheduledThreadPool

支持定时任务和周期性任务的线程池

可以定时或以固定频率执行任务，线程池大小可以由用户指定。
适用于需要周期性任务执行的场景，如定时任务调度器。

WorkStealingPool

基于任务窃取算法的线程池。

线程池中的每个线程维护一个双端队列（deque），线程可以从自己的队列中取任务执行。
如果线程的任务队列为空，它可以从其他线程的队列中“窃取”任务来执行，达到负载均衡的效果。
适合大量小任务并行执行，特别是递归算法或大任务分解成小任务的场景。

不同线程池的选择总结

FixedThreadPool	适合任务数量相对固定，且需要限制线程数的场景，避免线程过多占用系统资源。
CachedThreadPool	更适合大量短期任务或任务数量不确定的场景，能够根据任务量动态调整线程数。
SingleThreadExecutor	保证任务按顺序执行，适合要求严格顺序执行的场景。
ScheduledThreadPool	是定时任务的最佳选择，能够轻松实现周期性任务调度。
WorkStealingPool	适合处理大量的小任务，能更好地利用CPU资源。

3. Executors

为了简化线程池的使用步骤， Java标准库提供了另一组类 Executors，针对ThreadPoolExecutor进行了进一步封装，Executors 也是基于工厂设计模式实现的类。

最主要的是上面两种方法

newFixedThreadPool	创建一个固定线程数量的线程池（核心线程数和最大线程数一样）
newCachedThreadPool	最大线程数是一个很大的数字（线程数可以无限增加）

使用案例

(1) newFixedThreadPool

(2) newCachedThreadPool()

值得一提

两个线程池完成打印任务后，都不会马上终止进程，而是停留一段时间，直到确认工作队列中没有新的任务需要执行，进程才会结束；
Executors 的线程数目，拒绝策略等信息都是隐式的，可能不好控制，所以更推荐 ThreadPoolExecutor 来创建线程池。

4. 模拟实现一个固定线程个数的线程池

4.1 代码讲解

使用一个 BlockingQueue组织所有的任务；

使用Worker类描述一个工作线程，使用Runnable 描述一个任务；

核心操作为submit()，将任务加入线程池中；

得有线程来执行队列中的任务，在构造方法中，把线程创建出来；

每个worker线程要做的事情：不停地从BlockingQueue中取任务并执行，没取到任务就阻塞等待；

指定一下线程池中的最大线程数maxWorkerCount，如果当前线程数超过这个最大值时，就不再新增线程了；

向线程池提交任务（注意往线程池 submit() 方法的写法）

执行结果

这10个线程还在执行，只是因为阻塞队列中的任务都已经被 take() 完了，因此所有线程都进入阻塞状态，并且这些线程是由我们程序员设置的，默认都是前台线程，会阻止进程结束；

强制终止进程是一种异常的情况，退出码不为0

可以在线程创建好时，把这些线程全部设置成后台线程；

shutdown() 能够把线程池里的线程全部关闭，但是不能保证线程池内的任务一定能全部执行完毕;

所以，如果需要等待线程池内的任务全部执行完毕，需要调用 awaitTermination() 方法;

4.2 完整代码

package Thread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class MyThreadPool{

    private BlockingQueue<Runnable> queue = null;

    public MyThreadPool(int n){
        //初始化线程池，创建固定个数的线程
        queue = new ArrayBlockingQueue<>(1000);
        //使用 ArrayBlockingQueue 为任务队列，容量1000

        //创建 N 个线程
        for (int i = 0; i < n; i++) {
            Thread t = new Thread(() -> {
                try {
                    while (true){
                        Runnable task = queue.take();
                        task.run();
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            });
            //t.setDaemon(true);
            t.start();
        }
    }

    public void submit(Runnable task) throws InterruptedException {
        //把任务放入队列中
        queue.put(task);
    }
}
public class Demo32 {

    public static void main(String[] args) {
        ExecutorService threadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 1000; i++) {
            int id = i;
            threadPool.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " id = " + id );
            });
        }
        threadPool.shutdown();
    }
    public static void main1(String[] args) throws InterruptedException {
        MyThreadPool pool = new MyThreadPool(10);

        for (int i = 0; i < 100; i++) {
            int id = i;
            pool.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " id = " + id );
            });
        }
    }
}