Java 多线程（七）：Executor 线程池框架

迈不过友情╰ 2021-08-12 00:39 498阅读 0赞

*  线程的频繁创建在高并发及大数据量时是非常消耗资源的
 *  因此 Java 提供了线程池，Java5 在 java.util.concurrent 中添加了 Exectuor **异步执行框架**
 *  通过线程池减少线程创建和销毁的次数，重复利用，根据系统情况调整线程池数量，防止创建过多线程
 *  java.util.concurrent 中包含了几个比较重要的类：

<table> 
 <thead> 
  <tr> 
   <th style="text-align:center">类名</th> 
   <th style="text-align:center">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="text-align:center">Executor</td> 
   <td style="text-align:center">线程池接口</td> 
  </tr> 
  <tr> 
   <td style="text-align:center">ExecutorService</td> 
   <td style="text-align:center">在基础 Executor 线程池接口上生命了生命周期管理方法、任务执行状况跟踪方法</td> 
  </tr> 
  <tr> 
   <td style="text-align:center">ScheduledExecutorService</td> 
   <td style="text-align:center">一个定时调度任务的接口</td> 
  </tr> 
  <tr> 
   <td style="text-align:center">ScheduledThreadPoolExecutor</td> 
   <td style="text-align:center">ScheduledExecutorService 的实现，实现了可定时调度任务的线程池</td> 
  </tr> 
  <tr> 
   <td style="text-align:center">ThreadPoolExecutor</td> 
   <td style="text-align:center">线程池，可以通过调用 Executors 以下静态工厂方法来创建线程池并返回一个 ExecutorService 对象</td> 
  </tr> 
 </tbody> 
</table>

## ThreadPoolExecutor 构造函数 ##

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) //后两个参数为可选参数

*  **corePoolSize**：核心线程数，如果运行的线程少于 corePoolSize 则创建新线程来执行新任务，即使线程池中的其他线程是空闲的
 *  **maximumPoolSize**：最大线程数，corePoolSize和maximumPoolSize设置的边界自动调整池大小：
    
     *  当前线程数 >= corePoolSize（且任务队列已满时，线程会创建新线程来处理任务）
     *  当前线程数 = maxPoolSize（且任务队列已满时，线程池会拒绝处理任务而抛出异常）
 *  **keepAliveTime**：如果线程数多于 corePoolSize 则这些多余的线程的空闲时间超过 keepAliveTime 时将被终止
 *  **unit**：keepAliveTime 参数的时间单位
 *  **workQueue**：保存任务的阻塞队列，与线程池的大小有关：
    
     *  当运行的线程数少于 corePoolSize 时，新任务直接创建新线程来执行任务而无需再进队列
     *  当运行的线程数等于或多于 corePoolSize 新任务添加时则选加入队列，不直接创建线程
     *  当队列满时，新任务就会创建新线程
 *  **threadFactory**：使用 ThreadFactory 创建新线程，默认使用 defaultThreadFactory 创建线程
 *  **handle**：定义处理被拒绝任务的策略，默认使用 ThreadPoolExecutor.AbortPolicy 任务被拒绝时将抛出 RejectExecutorException

## Executors ##

*  Executors 提供了一系列静态工厂方法用于创建各种线程池，根据具体应用场景而选择不同的线程池

### newFixedThreadPool ###

*  创建可重用且固定线程数的线程池，**如果线程池中的所有线程都处于活动状态，此时再提交任务就在队列中等待，直到有可用线程**
 *  如果线程池中的某个线程由于异常而结束时，线程池就会再添加一个新线程

> 方法定义

//使用一个基于FIFO排序的阻塞队列，在所有corePoolSize线程都忙时新任务将在队列中等待
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

> 使用例子

public class Main {
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            IntStream.range(0, 6).forEach(i -> executorService.execute(() -> {
                try {
                    TimeUnit.SECONDS.sleep(1);
                    String threadName = Thread.currentThread().getName();
                    System.out.println("finished: " + threadName);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }));
        }
    }
    
    //输出结果为：
    //finished: pool-1-thread-1
    //finished: pool-1-thread-2
    //finished: pool-1-thread-3
    //finished: pool-1-thread-4
    //finished: pool-1-thread-5
    //finished: pool-1-thread-1

### newSingleThreadExecutor ###

*  创建一个单线程的 Executor，它只会用唯一的线程来执行任务，**保证所有任务按照指定顺序（FIFO, LIFO, 优先级）执行**
 *  如果该线程因为异常而结束就新建一条线程来继续执行后续的任务

> 方法定义

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        //corePoolSize和maximumPoolSize都等于，表示固定线程池大小为1
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }

> 使用例子

public class Main {
        private static int num = 100;
    
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            IntStream.range(0, 100).forEach(i -> executorService.execute(() -> {
                String runnableName = "Runnable"+i+":";
                System.out.println(runnableName + num--);
            }));
            try {
                executorService.shutdown();
                executorService.awaitTermination(5, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (!executorService.isTerminated()) {
                    executorService.shutdownNow();
                }
            }
            System.out.println("执行结束");
        }
    }
    
    /** 输出结果为：
        Runnable0:100
        Runnable1:99
        Runnable2:98
        .....
        Runnable98:2
        Runnable99:1
        执行结束
    **/

### newScheduledThreadPool ###

*  创建一个可延迟执行或定期执行的线程池

> 方法定义

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

> 使用例子

public class Main {
        public static void main(String[] args) {
            ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
            IntStream.range(0, 2).forEach(i -> executorService.scheduleAtFixedRate(() -> {
                    String threadName = Thread.currentThread().getName();
                    System.out.println("finished: " + threadName);
            },1000, 2000, TimeUnit.MILLISECONDS));
        }
    }
    
    //输出结果为：
    //finished: pool-1-thread-1
    //finished: pool-1-thread-2
    //finished: pool-1-thread-1
    //finished: pool-1-thread-2
    //finished: pool-1-thread-2
    //finished: pool-1-thread-1

### newCachedThreadPool ###

*  创建可缓存的线程池，如果线程池中的线程在 60 秒未被使用就将被移除，**在执行新的任务时，当线程池中有之前创建的可用线程就重用可用线程，否则就新建一个线程**

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

public class Main {
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newCachedThreadPool();
            IntStream.range(0, 6).forEach(i -> executorService.execute(() -> {
                    String threadName = Thread.currentThread().getName();
                    System.out.println("finished: " + threadName);
            }));
        }
    }
    
    //输出结果为：
    //finished: pool-1-thread-1
    //finished: pool-1-thread-2
    //finished: pool-1-thread-2
    //finished: pool-1-thread-2
    //finished: pool-1-thread-2
    //finished: pool-1-thread-2
    //线程池会缓存线程，尽可能的利用线程资源

## ExecutorService 常用方法 ##

*  **shutdown**：方法等待提交的任务执行完成并不再接受新任务，在完成全部提交的任务后关闭
 *  **shutdownNow**：方法将强制终止所有运行中的任务并不再允许提交新任务
 *  **awaitTermination**：阻塞，直到所有任务在关闭请求之后完成执行，或发生超时，或当前线程中断
 *  **isTerminated**：当所有任务关闭后返回 true，除非 shutdown、shutdownNow 被调用，否则永远不为 true
 *  **submit**：提交可执行的任务（Runnable、Thread、Callable）并执行，返回一个 Future 对象