线程池1

作者: GeekerLou | 来源:发表于2018-10-17 09:46 被阅读0次

构造一个线程池为什么需要几个参数？如果避免线程池出现OOM？Runnable和Callable的区别是什么？

一、背景

1.1 线程与线程池的基本概念

线程：进程中负责程序执行的执行单元。一个进程中至少有一个线程。
多线程：解决多任务同时执行的需求，合理使用CPU资源。多线程的运行是根据CPU切换完成，如何切换由CPU决定，因此多线程运行具有不确定性。
线程池：基本思想还是一种对象池的思想，开辟一块内存空间，里面存放了众多(未死亡)的线程，池中线程执行调度由池管理器来处理。当有线程任务时，从池中取一个，执行完成后线程对象归池，这样可以避免反复创建线程对象所带来的性能开销，节省了系统的资源。

1.2 直接创建线程的缺点

每次new thread新建对象，性能较差
线程缺乏统一管理，可能无限制地创建新的线程，相互竞争，有可能占用过多的系统资源导致死机或者OOM。
缺乏更多管理功能：更多执行？、定期执行、线程中断
开发实践过程中发现，Thread类提供那么多方法，说起来很容易，但是使用起来很容易出错。

1.3 使用封装的线程池的优点

重用存在的线程，减少对象的创建、消亡的开销、性能佳
可有效地控制最大并发线程数提供系统资源利用率，同时可以避免过多的资源竞争，避免发生阻塞。
提供定时执行、定期执行、单线程、并发数控制等功能。

二、线程池

2.1 线程池的定义

Java里面线程池的顶级接口是 Executor，不过真正的线程池接口是 ExecutorService， ExecutorService 的默认实现是 ThreadPoolExecutor；普通类 Executors 里面调用的就是 ThreadPoolExecutor。

照例看一下各个接口的源码：

public interface Executor {
    void execute(Runnable command);
}

public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    
    boolean isShutdown();
    boolean isTerminated();
    
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
    Future<?> submit(Runnable task);
    ...
}

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

2.2 线程池的创建

Java中创建线程池很简单，只需要调用Executors中相应的便捷方法即可，比如Executors.newFixedThreadPool(int nThreads)。一个简单的创建固定数目线程的线程池的用法如下：

ExecutorService pool = Executors.newCachedThreadPool();

看起来非常简单，但是便捷不仅隐藏了复杂性，也为我们埋下了潜在的隐患（OOM，线程耗尽）。

Executors创建线程池便捷方法列表：

方法名	功能
newCachedThreadPool	创建固定大小的线程池
newSingleThreadExecutor()	创建只有一个线程的线程池
newCachedThreadPool()	创建一个不限线程数上限的线程池，任何提交的任务都将立即执行
newScheduledThreadPool	创建一个支持周期性调度的线程池

Executors 提供四种线程池：
1）newCachedThreadPool：是一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们。对于执行很多短期异步任务的程序而言，这些线程池通常可提高程序性能。调用 execute() 将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。因此，长时间保持空闲的线程池不会使用任何资源。注意，可以使用 ThreadPoolExecutor 构造方法创建具有类似属性但细节不同（例如超时参数）的线程池。

2）newSingleThreadExecutor ：创建是一个单线程池，也就是该线程池只有一个线程在工作，所有的任务是串行执行的，如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它，此线程池保证所有任务的执行顺序按照任务的提交顺序执行。

3）newFixedThreadPool ：创建固定大小的线程池，每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小，线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。

4）newScheduledThreadPool ：创建一个大小无限的线程池，此线程池支持定时以及周期性执行任务的需求。

根据《阿里巴巴Java开发手册》中的约定，非常不建议直接使用Executor去创建线程池，具体原因如下：

线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。说明：Executors各个方法的弊端：
1）newFixedThreadPool和newSingleThreadExecutor:
主要问题是默认使用无界队列，堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。
2）newCachedThreadPool和newScheduledThreadPool:
主要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。

2.3 线程池的参数

通过 ThreadPoolExecutor 的构造函数，撸一撸线程池相关参数的概念：

// Java线程池的完整构造函数
public ThreadPoolExecutor(
  int corePoolSize, // 线程池长期维持的线程数，即使线程处于Idle状态，也不会回收。
  int maximumPoolSize, // 线程数的上限
  long keepAliveTime, TimeUnit unit, // 超过corePoolSize的线程的idle时长，
                                     // 超过这个时间，多余的线程会被回收。
  BlockingQueue<Runnable> workQueue, // 任务的排队队列
  ThreadFactory threadFactory, // 新线程的产生方式
  RejectedExecutionHandler handler) // 拒绝策略;
}

1）corePoolSize：线程池的核心线程数，一般情况下不管有没有任务都会一直在线程池中一直存活，只有在 ThreadPoolExecutor 中的方法 allowCoreThreadTimeOut(boolean value) 设置为 true 时，闲置的核心线程会存在超时机制，如果在指定时间没有新任务来时，核心线程也会被终止，而这个时间间隔由第3个属性 keepAliveTime 指定。

2）maximumPoolSize：线程池所能容纳的最大线程数，当活动的线程数达到这个值后，后续的新任务将会被阻塞。

3）keepAliveTime：控制线程闲置时的超时时长，超过则终止该线程。一般情况下用于非核心线程，只有在 ThreadPoolExecutor 中的方法 allowCoreThreadTimeOut(boolean value) 设置为 true时，也作用于核心线程。

4）unit：用于指定 keepAliveTime 参数的时间单位，TimeUnit 是个 enum 枚举类型，常用的有：TimeUnit.HOURS(小时)、TimeUnit.MINUTES(分钟)、TimeUnit.SECONDS(秒) 和 TimeUnit.MILLISECONDS(毫秒)等。

5）workQueue：线程池的任务队列，通过线程池的 execute(Runnable command) 方法会将任务 Runnable 存储在队列中。

6）threadFactory：线程工厂，它是一个接口，用来为线程池创建新线程的。一般用来定义线程池中的线程名称。

7）RejectedExecutionHandler：拒绝策略，即当任务等待队列塞满情况时的处理策略。

竟然有7个参数，构造一个线程池确实需要这么多参数。这些参数中，比较容易引起问题的有corePoolSize, maximumPoolSize, workQueue以及handler，这些参数设置不当将会导致如下的问题：

corePoolSize和maximumPoolSize设置不当会影响效率，甚至耗尽线程；
workQueue设置不当容易导致OOM；
handler设置不当会导致提交任务时抛出异常。
正确的参数设置方式会在下文给出。

2.4 线程池的工作原理

官方文档中是这么说明线程池的工作原理的：

If fewer than corePoolSize threads are running, the Executor always prefers adding a new thread rather than queuing.
If corePoolSize or more threads are running, the Executor always prefers queuing a request rather than adding a new thread.
If a request cannot be queued, a new thread is created unless this would exceed maximumPoolSize, in which case, the task will be rejected.

我们以ThreadPoolExecutor为例展示下线程池的工作流程图:

1.png

2.png

如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（注意，执行这一步骤需要获取全局锁）。
如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue。
如果无法将任务加入BlockingQueue（队列已满），则在非corePool中创建新的线程来处理任务（注意，执行这一步骤需要获取全局锁）。
如果创建新线程将使当前运行的线程超出maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

ThreadPoolExecutor采取上述步骤的总体设计思路，是为了在执行execute()方法时，尽可能地避免获取全局锁（那将会是一个严重的可伸缩瓶颈）。在ThreadPoolExecutor完成预热之后（当前运行的线程数大于等于corePoolSize），几乎所有的execute()方法调用都是执行步骤2，而步骤2不需要获取全局锁。

总结：corePoolSize -> 任务队列 -> maximumPoolSize -> 拒绝策略

2.5 任务提交

2.5.1 Runnable和Callable

可以向线程池提交的任务有两种：Runnable和Callable，二者的区别如下：

方法签名不同，void Runnable.run(), V Callable.call() throws Exception
是否允许有返回值，Callable允许有返回值
是否允许抛出异常，Callable允许抛出异常。

总结：Callable是JDK1.5时加入的接口，作为Runnable的一种补充，允许有返回值，允许抛出异常。

2.5.2 三种提交任务的方式：

提交方式	是否关心返回结果
Future<T> submit(Callable<T> task)	是
void execute(Runnable command)	否
Future<?> submit(Runnable task)	否，虽然返回Future，但是其get()方法总是返回null

2.6 线程池的关闭

ThreadPoolExecutor 提供了两个方法，用于线程池的关闭，分别是 shutdown() 和 shutdownNow()。

shutdown()：不会立即的终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务。
shutdownNow()：立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务。

三、正确使用线程池

3.1 避免使用无界队列

不要使用Executors.newXXXThreadPool()快捷方法创建线程池，因为这种方式会使用无界的任务队列，为避免OOM，我们应该使用ThreadPoolExecutor的构造方法手动指定队列的最大长度：

ExecutorService executorService = new ThreadPoolExecutor(2, 2, 
                0, TimeUnit.SECONDS, 
                new ArrayBlockingQueue<>(512), // 使用有界队列，避免OOM
                new ThreadPoolExecutor.DiscardPolicy());

3.2 明确拒绝任务时的行为

任务队列总有占满的时候，这是再submit()提交新的任务会怎么样呢？RejectedExecutionHandler接口为我们提供了控制方式，接口定义如下：

public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

线程池给我们提供了几种常见的拒绝策略：

RejectedExcutionHandler.png

拒绝策略	拒绝行为
AbortPolicy	为Java线程池默认的阻塞策略，不执行此任务，而且直接抛出一个运行时异常，切记ThreadPoolExecutor.execute需要try catch，否则程序会直接退出。
DiscardPolicy	直接抛弃，任务不执行，空方法
DiscardOldestPolicy	丢弃执行队列中最老的任务，并再次execute 此task。
CallerRunsPolicy	在调用execute的线程里面执行此command，会阻塞入口

线程池默认的拒绝行为是AbortPolicy，也就是抛出RejectedExecution异常，该异常是非受检异常，很容易忘记捕获。如果不关心任务被拒绝的事件，可以将拒绝策略设置成DiscardPolicy，这样多余的任务会悄悄的被忽略。

ExecutorService executorService = new ThreadPoolExecutor(2, 2, 
                0, TimeUnit.SECONDS, 
                new ArrayBlockingQueue<>(512), 
                new ThreadPoolExecutor.DiscardPolicy());// 指定拒绝策略

获取处理结果和异常
线程池的处理结果、以及处理过程中的异常都被包装到Future中，并在调用Future.get()方法时获取，执行过程中的异常会被包装成ExecutionException，submit()方法本身不会传递结果和任务执行过程中的异常。获取执行结果的代码可以这样写：

ExecutorService executorService = Executors.newFixedThreadPool(4);
Future<Object> future = executorService.submit(new Callable<Object>() {
        @Override
        public Object call() throws Exception {
            throw new RuntimeException("exception in call~");// 该异常会在调用Future.get()时传递给调用者
        }
    });
    
try {
  Object result = future.get();
} catch (InterruptedException e) {
  // interrupt
} catch (ExecutionException e) {
  // exception in Callable.call()
  e.printStackTrace();
}

上述代码输出类似如下：

四、线程池的常用场景

4.1 正确构造线程池

int poolSize = Runtime.getRuntime().availableProcessors() * 2;
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(512);
RejectedExecutionHandler policy = new ThreadPoolExecutor.DiscardPolicy();
executorService = new ThreadPoolExecutor(poolSize, poolSize,
    0, TimeUnit.SECONDS,
            queue,
            policy);

4.2 获取结果

4.2.1 获取单个结果

过submit()向线程池提交任务后会返回一个Future，调用V Future.get()方法能够阻塞等待执行结果，V get(long timeout, TimeUnit unit)方法可以指定等待的超时时间。

4.2.2 获取多个结果

如果向线程池提交了多个任务，要获取这些任务的执行结果，可以依次调用Future.get()获得。但对于这种场景，我们更应该使用ExecutorCompletionService，该类的take()方法总是阻塞等待某一个任务完成，然后返回该任务的Future对象。向CompletionService批量提交任务后，只需调用相同次数的CompletionService.take()方法，就能获取所有任务的执行结果，获取顺序是任意的，取决于任务的完成顺序：

void solve(Executor executor, Collection<Callable<Result>> solvers)
   throws InterruptedException, ExecutionException {

   CompletionService<Result> ecs = new ExecutorCompletionService<Result>(executor);// 构造器

   for (Callable<Result> s : solvers)// 提交所有任务
       ecs.submit(s);

   int n = solvers.size();
   for (int i = 0; i < n; ++i) {// 获取每一个完成的任务
       Result r = ecs.take().get();
       if (r != null)
           use(r);
   }
}

4.3 设置超时

4.3.1 单个任务的超时时间

V Future.get(long timeout, TimeUnit unit)方法可以指定等待的超时时间，超时未完成会抛出TimeoutException。

4.3.2 多个任务的超时时间

等待多个任务完成，并设置最大等待时间，可以通过CountDownLatch完成：

public void testLatch(ExecutorService executorService, List<Runnable> tasks) 
    throws InterruptedException{

    CountDownLatch latch = new CountDownLatch(tasks.size());
      for(Runnable r : tasks){
          executorService.submit(new Runnable() {
              @Override
              public void run() {
                  try{
                      r.run();
                  }finally {
                      latch.countDown();// countDown
                  }
              }
          });
      }
      latch.await(10, TimeUnit.SECONDS); // 指定超时时间
  }

4.4 线程池和装修公司

以运营一家装修公司做个比喻。公司在办公地点等待客户来提交装修请求；公司有固定数量的正式工以维持运转；旺季业务较多时，新来的客户请求会被排期，比如接单后告诉用户一个月后才能开始装修；当排期太多时，为避免用户等太久，公司会通过某些渠道（比如人才市场、熟人介绍等）雇佣一些临时工（注意，招聘临时工是在排期排满之后）；如果临时工也忙不过来，公司将决定不再接收新的客户，直接拒单。

线程池就是程序中的“装修公司”，代劳各种脏活累活。上面的过程对应到线程池上：

// Java线程池的完整构造函数
public ThreadPoolExecutor(
  int corePoolSize, // 正式工数量
  int maximumPoolSize, // 工人数量上限，包括正式工和临时工
  long keepAliveTime, TimeUnit unit, // 临时工游手好闲的最长时间，超过这个时间将被解雇
  BlockingQueue<Runnable> workQueue, // 排期队列
  ThreadFactory threadFactory, // 招人渠道
  RejectedExecutionHandler handler) // 拒单方式