线程的池

2024-01-30 08:58:59

1. 线程池

线程池和数据库的连接池是同样意思，把多个线程放在一个集合里，有任务时从集合里分配线程，当该线程完成任务后不是销毁，放入线程池等待下次任务，减少了创建和销毁线程的次数，提高系统效率，因为创建和销毁属于重操作。如果每有一个任务就创建一个线程，大量任务涌进会导致创建过多线程而内存溢出

2. Excutor

java.util.concurrent.Executor 提供一系列与线程池相关的接口，及其实现类，其中抽取常用的部分来讲解

UML图：

ExecutorService接口方法：

AbstractExecutorService类方法：

Executor 接口里定义了 execute(Runnable command)方法
ExecutorService定义了线程生命周期的相关方法
AbstractExecutorService提供了默认实现
ThreadPoolExecutor推荐使用的线程池类
后面发现有个ForkJoinPool 线程池类，从1.7开始有的，不做讨论了

3. ThreadPoolExecutor

这个常用的类提供了创建线程池的方法，根据传入的参数不同，创建不同的线程池，先来看看构造方法

public ThreadPoolExecutor(
    
    int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler) {
    
    //省略具体逻辑，单看参数
}

corePoolSize：核心线程数量
- 若现有的线程数量少于核心线程数量，则创建新的线程处理请求
- 若现有的线程数量多于核心线程数量，但小于最大线程数量，则队满时才创建新线程
- 若核心线程数量等于最大线程数量，则创建大小固定的线程池
- 若最大线程数量为无限，则线程池任意大小
maximumPoolSize：最大线程数量
keepAliveTime：空闲保持时间
- 若现有线程数量多于核心线程数，且超出空闲保持时间，则多余的线程会销毁
unit：空闲时间的单位
workQueue：排队策略
- 同步移交：不放入队列，而是等待线程执行它。如果当前线程没有执行，很可能会新开一个线程执行。
- *限策略：如果核心线程都在工作，该线程会放到队列中。所以线程数不会超过核心线程数
- 有界限策略：可以避免资源耗尽，但是一定程度上减低了吞吐量
threadFactory：创建线程的工厂
handler：拒绝策略
- 直接抛出异常
- 使用调用者的线程来处理（多出的相当于没使用线程池）
- 直接丢掉这个任务
- 丢掉最老的任务

4. 线程池的状态

RUNNING：线程池能接受新任务，以及对新添加的任务进行处理
SHUTDOWN：线程池不接受新任务，但会对已添加的任务进行处理
STOP：线程池不接收新任务，不处理已添加的任务，并且会中断正在处理的任务
TIDYING：所有的任务已终止，ctl记录的"任务数量"为0，线程池会变为TIDYING状态，当线程池变为TIDYING状态时，会执行钩子函数terminated()，terminated()在ThreadPoolExecutor类中是空的，若用户想在线程池变为TIDYING时，进行相应的处理，可以通过重载terminated()函数来实现
TERMINATED：线程池真正的终止

5. 线程池方法

execute()，shutdown()，shutdownNow() 是开启和关闭线程池的方法，前者进行相应逻辑判断再考虑是否创建新线程，两者是关闭线程池，区别于后者不等其任务完成就中断线程

6. 快捷创建线程池

在Executor类中，有下面几个静态方法来快捷创建线程池，下面写三个：

newFixedThreadPool：corePoolSize和maximumPoolSize相等
newCachedThreadPool：若新任务进来，没空闲进程会立马创建
SingleThreadExecutor：单线程，从队列中取任务执行

7. 线程任务

获取线程的结果：
- Future：Futrue模式就是Action先给Invoker一个未来（future），其实也就是票据，Invoker就可以继续接下来的步骤而无需等待Action结果的返回，通过 future.get() 可以获得返回值

提交任务的种类：
- Runnable：重写里面的run方法
- Callable（可以认为是Runnable的扩展，多了返回值或异常）：重写call方法

public static void main(String[] args) {
    
    //快捷线程池
    ExecutorService pool = Executors.newFixedThreadPool(10);
    
    //没有返回值
    Runnable runnable = () -> System.out.println("Runnable任务");
    
    //有返回值
    Callable callable = () -> {
        String msg = "Callable任务";
        System.out.println(msg);
        return msg;
    };
    
    //提交任务，并获取返回值
    Future f1 = pool.submit(runnable);
    Future f2 = pool.submit(callable);
    
    try {
        System.out.println("f1:" + f1.get());
        System.out.println("f2:" + f2.get());
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
    
    //关闭线程池
    pool.shutdown();
}

Callable任务
Runnable任务
f1:null
f2:Callable任务

提交任务的方式：
- submit：可以接收Runnable，和Callable，有返回值和异常，底层还是用excute
- execute，只能接收Runnable，没有返回值

submit底层是用execute实现的：

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);     //把Callable变成RunnableFuture
    execute(ftask);         //把RunnableFuture传入execute
    return ftask;
}

Callable会被包装成RunnableFuture，在RunnableFuture的run中会调用callable的call方法，然后把返回值或异常放入该类的静态变量中

8.线程池实现

public static void main(String[] args) {
    
    //核心线程
    int corePoolSize = 5;
    //最大线程
    int maximumPoolSize = 10;
    //保持空闲时间
    long keepAliveTime = 10;
    //空闲时间单位，秒
    TimeUnit unit = TimeUnit.SECONDS;
    //排队策略，基于数组结构的有界阻塞队列
    BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(5);
    //设置创建线程的工厂，可以预设部分内容：eg:守护进程，优先级
    ThreadFactory threadFactory = Executors.defaultThreadFactory();
    //拒绝策略
    RejectedExecutionHandler handler = new AbortPolicy();;
    
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize,
                                                                   maximumPoolSize,
                                                                   keepAliveTime,
                                                                   unit,
                                                                   workQueue,
                                                                   threadFactory,
                                                                   handler);
    
    Callable callable = () -> {
        String msg = "Callable任务";
        System.out.println(msg);
        return msg;
    };
    
    threadPoolExecutor.submit(() -> System.out.println("线程测试1"));
    threadPoolExecutor.submit(callable);

}

<!-- 打印 -->
线程测试1
Callable任务

码农公寓