线程的池

Posted 2021-03-13 howlet

---

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()，submit()，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任务