Executors线程池

一、什么是线程池

线程池是一组已经初始化并等待执行任务的线程集合。通过使用线程池，我们可以避免频繁地创建和销毁线程，从而节省资源和减少系统开销。线程池的核心思想是通过复用线程来提高性能。

二、为什么使用线程池

• 资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。
• 提高响应速度：任务到达时，无需等待线程创建即可立即执行。
• 提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。

三、Java 中的线程池

Java 提供了 java.util.concurrent 包来支持并发编程。其中有几种方法创建线程池

1、通过 Executors 工厂类的静态方法创建线程池：

newFixedThreadPool 固定数量的线程池

 /**
  * 创建一个可重用固定数量线程的线程池，这些线程操作于共享的无界队列。
   * 在任何时候，最多有 {@code nThreads} 个线程会处于活动状态处理任务。
   * 当所有线程都在活动时，如果提交了额外的任务，它们将在队列中等待，
   * 直到有线程可用。
   * 如果在关闭前，任何线程因执行失败而终止，将根据需要创建一个新的线程来执行后续任务。
   * 线程池中的线程将持续存在，直到显式调用 {@link ExecutorService#shutdown shutdown} 方法关闭它。
   *
   * @param nThreads 线程池中的线程数量
   * @return 新创建的线程池
   * @throws IllegalArgumentException 如果 {@code nThreads <= 0}
   */
 public static ExecutorService newFixedThreadPool(int nThreads) {
     return new ThreadPoolExecutor(nThreads, nThreads,
                                   0L, TimeUnit.MILLISECONDS,
                                   new LinkedBlockingQueue<Runnable>());
 }

特点：

• 如果线程数没有达到“固定数量”，每次提交一个任务线程池内就创建一个新线程，直到线程达到线程池固定的数量
• 线程池的大小一旦达到“固定数量”就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程
• 在接收异步任务的执行目标实例时，如果池中的所有线程均在繁忙状态，新任务会进入阻塞队列中（无界的阻塞队列）

适用场景：

• 需要任务长期执行的场景
• CPU密集型任务

缺点：

• 内部使用无界队列来存放排队任务，当大量任务超过线程池最大容量需要处理时，队列无限增大，使服务器资源迅速耗尽

newSingleThreadExecutor 单线程化线程池

/**
 * 创建一个使用单一工作线程的ExecutorService来执行任务。
 * 此执行器服务使用单一线程以顺序方式处理任务，确保每次只执行一个任务。
 * 即使在执行前一个任务时因失败导致线程终止，在关闭前也会创建新的线程继续执行后续任务。
 * 与等效的{@code newFixedThreadPool(1)}不同，返回的执行器保证不能重新配置为使用额外的线程。
 *
 * @return 新创建的单线程ExecutorService
 */
public static ExecutorService newSingleThreadExecutor() {
   return new FinalizableDelegatedExecutorService
       (new ThreadPoolExecutor(1, 1,
                               0L, TimeUnit.MILLISECONDS,
                               new LinkedBlockingQueue<Runnable>()));
}

特点：

• 单线程化的线程池中的任务是按照提交的次序顺序执行的
• 只有一个线程的线程池
• 池中的唯一线程的存活时间是无限的
• 当池中的唯一线程正繁忙时，新提交的任务实例会进入内部的阻塞队列中，并且其阻塞队列是无界的

适用场景：

• 任务按照提交次序，一个任务一个任务地逐个执行的场景

newCachedThreadPool 可缓存线程池

/**
 * 创建一个可根据需要创建新线程的线程池，同时会重用之前已构建的空闲线程。
 * 这种线程池通常能提升执行大量短生命周期异步任务程序的性能。
 * 调用{@code execute}方法时，如果存在可用的先前构建的线程则会重用它们。
 * 若无现有线程可用，将创建新线程并加入到线程池中。
 * 在线程闲置六十秒后，线程将被终止并从缓存中移除。
 * 因此，长时间处于空闲状态的线程池不会消耗任何资源。
 * 注意：可以通过{@link ThreadPoolExecutor}构造器创建具有类似特性但细节不同的（例如超时参数）线程池。
 *
 * @return 新创建的线程池实例
 */
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

特点：

• 在接收新的异步任务target执行目标实例时，如果池内所有线程繁忙，此线程池就会添加新线程来处理任务
• 线程池不会对线程池大小进行限制，线程池大小完全依赖于操作系统（或者说JVM）能够创建的最大线程大小
• 如果部分线程空闲，也就是存量线程的数量超过了处理任务数量，就会回收空闲（60秒不执行任务）线程

适用场景：

• 需要快速处理突发性强、耗时较短的任务场景，如Netty的NIO处理场景、REST API接口的瞬时削峰场景

缺点：

• 线程池没有最大线程数量限制，如果大量的异步任务执行目标实例同时提交，可能会因创建线程过多而导致资源耗尽

一般在工程上不建议使用它们创建线程池，阿里java开发手册

2、ScheduledExecutorService 可调度线程池

ScheduledExecutorService 是 ExecutorService 的子接口，专门用于支持定时及周期性任务执行的线程池。可以通过 Executors.newScheduledThreadPool(int corePoolSize) 方法创建：

 /**
 * 创建一个可以调度命令在指定延迟后运行，或周期性执行的线程池。
 * 
 * @param corePoolSize 即使在空闲时，线程池中也要保持的线程数量
 * @return 新创建的ScheduledExecutorService实例，即计划线程池
 * @throws IllegalArgumentException 如果corePoolSize小于0
 */
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue());
}

3、ThreadPoolExecutor构造器

使用 ThreadPoolExecutor 的构造函数可以更加灵活地配置线程池，例如指定核心线程数、最大线程数、线程存活时间、工作队列等参数，如下所示：

 public ThreadPoolExecutor(
 	int corePoolSize, //核心线程数，即使线程空闲也不会被收回
	int maximumPoolSize, // 最大线程数
	long keepAliveTime, //线程的最大空闲时长
	TimeUnit unit, // 线程存活时间单位
	BlockingQueue<Runnable> workQueue, // 阻塞队列
	ThreadFactory threadFactory, // 线程工厂
	RejectedExecutionHandler handler //拒绝策略
)

核心线程和最大线程数量

• corePoolSize用于设置核心（Core）线程池数量，参数maximumPoolSize用于设置最大线程数量
• 线程池接收到新任务，当前工作线程数少于corePoolSize, 即使有空闲的工作线程，也会创建新的线程来处理该请求，直到线程数达到corePoolSize
• 当前工作线程数多于corePoolSize数量，但小于maximumPoolSize数量，那么仅当任务排队队列已满时才会创建新线程
• maximumPoolSize被设置为无界值（如Integer.MAX_VALUE）时，线程池可以接收任意数量的并发任务

BlockingQueue

• BlockingQueue（阻塞队列）的实例用于暂时接收到的异步任务，如果线程池的核心线程都在忙，那么所接收到的目标任务缓存在阻塞队列中

keepAliveTime

• 空闲线程存活时间
• 用于设置池内线程最大Idle（空闲）时长（或者说保活时长）
• 超过这个时间，默认情况下Idle、非Core线程会被回收

ThreadPoolExecutor 线程池的执行过程

四、线程池的拒绝策略

名称	描述
AbortPolicy	丢弃任务并抛出RejectedExecutionException异常。这是线程池默认的拒绝策略，在任务不能再提交的时候，抛出异常，及时反馈程序运行状态。如果是比较关键的业务推荐使用此拒绝策略，这样子在系统不能承载更大的并发量的时候，能够及时的通过异常发现。
CallerRunsPolicy	由调用线程(提交任务的线程)处理该任务。这种情况是需要让所有任务都执行完毕那么就适合大量计算的任务类型去执行，多线程仅仅是增大吞吐量的手段，最终必须要让每个任务都执行完毕。
DiscardPolicy	丢弃任务，但是不抛出异常。 使用此策略，可能会使我们无法发现系统的异常状态。建议是一些无关紧要的业务采用此策略。
DiscardOldestPolicy	丢弃队列最前面的任务，然后重新提交被拒绝的任务。是否要采用此种拒绝策略，还得根据实际业务是否允许丢弃老任务来认真衡量。

五、向线程池提交任务的两种方式

1. execute方法

   • void execute(Runnable command): Executor接口中的方法

2. submit方法

   • <T> Future<T> submit(Callable<T> task);
   • <T> Future<T> submit(Runnable task, T result);
   • Future<?> submit(Runnable task);

   这3个submit方法都是ExecutorService接口中的方法

两种方法的区别：

• execute()方法只能接收Runnable类型的参数，而submit()方法可以接收Callable、Runnable两种类型的参数
• Callable类型的任务是可以返回执行结果的，而Runnable类型的任务不可以返回执行结果
• submit()提交任务后会有返回值，而execute()没有
• submit()方便Exception处理

1、通过submit()返回的Future对象获取结果

public class CreateThreadPollDemo {
    public static void main(String[] args) throws InterruptedException {
       ScheduledExecutorService pool=Executors.newScheduledThreadPool(2);
       Future<Integer> future=pool.submit(new Callable<Integer>() {

        @Override
        public Integer call() throws Exception {
            return 123;
        }
           
       });
       try {
        Integer result=future.get();
        System.out.println("result:"+result);//123
    } catch (ExecutionException e) {
        e.printStackTrace();
    }
    Thread.sleep(1000);
    pool.shutdown();

    }
}

2、通过submit()返回的Future对象捕获异常

public class CreateThreadPollDemo {
    public static final int SLEEP_GAP=1000;
    static class TargetTask implements Runnable{
        static AtomicInteger taskNo=new AtomicInteger(1);
        String taskName;
        public TargetTask()
        {
            taskName="task-"+taskNo;
            taskNo.incrementAndGet();
        }
        public void run()
        {
            System.out.println(taskName+" is doing...");
            try {
                Thread.sleep(SLEEP_GAP);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(taskName+" end...");
        }
    }
    static class TargetTaskWithError extends TargetTask{
        public void run()
        {
            super.run();//执行父类的run方法
            throw new RuntimeException("Error from "+taskName);
        }
    }

    public static void main(String[] args) throws InterruptedException {
       ScheduledExecutorService pool=Executors.newScheduledThreadPool(2);
      pool.execute(new TargetTaskWithError());
      Future future=pool.submit(new TargetTaskWithError());
      try {
        if(future.get()==null)
          {
              System.out.println("No Exception");
          }
    } catch (ExecutionException e) {
      
        e.printStackTrace();
    }
    Thread.sleep(1000);
    pool.shutdown();

    }
  
}

execute()方法在启动任务执行后，任务执行过程中可能发生的异常调用者并不关心。而通过submit()方法返回的Future对象（异步执行实例），可以进行异步执行过程中的异常捕获