ExecutorService的十个使用技巧

2022-08-19 17:26:32

ExecutorService] (https://docs.oracle.com/javase/8/docs/api/java/util/concurrent /ExecutorService.html)这个接口从Java 5开始就已经存在了。这得追溯到2004年了。这里小小地提醒一下，官方已经不再支持Java 5, Java 6了，Java 7[在半年后也将停止支持。我之所以会提起ExecutorService这么旧的一个接口是因为，大多数Java程序员并没有搞清楚它的工作原理。关于它可以介绍的有很多，这里我只想分享它的一些较少为人所知的特性以及实践技巧。本文主要是面向初级程序员的，并没有过于高深的东西。

1. 线程命名

这点得反复强调。对正在运行的JVM进行线程转储（thread dump）或者调试时，线程池默认的命名机制是pool-N-thread-M，这里N是线程池的序号（每新创建一个线程池，这个N都会加一），而M是池里线程的序号。比方说，pool-2-thread-3指的是JVM生命周期中第二个线程池里的第三个线程。参考这里 Executors.defaultThreadFactory()] (https://docs.oracle.com/javase/8/docs/api/java/util/concurrent /Executors.html#defaultThreadFactory--)。这样的名字表述性不佳。由于JDK将命名机制都隐藏在 [ThreadFactory 里面，这使得要正确地命名线程得稍微费点工夫。所幸的是Guava提供了这么一个工具类：

import com.google.common.util.concurrent.ThreadFactoryBuilder;

final ThreadFactory threadFactory = new ThreadFactoryBuilder()

        .setNameFormat("Orders-%d")

        .setDaemon(true)

        .build();

final ExecutorService executorService = Executors.newFixedThreadPool(10, threadFactory);

2. 根据上下文切换名字

这是我从高效的jstack:如何对高速运行的服务器进行调试一文中学到的一个技巧。线程名可以随时进行修改，只要你想这么做的话。这是有一定的意义的，因为线程转储只能看到类名和方法名，而没有参数及本地变量。通过调整线程名可以保留一些比较关键的上下文信息，这样排查消息/记录/查询等变慢或者出现死锁的问题时就容易多了。示例：

private void process(String messageId) {

  executorService.submit(() -> {

    final Thread currentThread = Thread.currentThread();

    final String oldName = currentThread.getName();

    currentThread.setName("Processing-" + messageId);

    try {

      //real logic here...

    } finally {

      currentThread.setName(oldName);

    }

  });

}

在try-finally块中当前线程的名字是Processing-某个消息ID。这对跟踪系统内的消息流会比较有用。

3. 显式地安全地关闭线程

客户端线程和线程池之间会有一个任务队列。当程序要关闭时，你需要注意两件事情：入队的这些任务的情况怎么样了以及正在运行的这个任务执行得如何了。令人惊讶的是很多开发人员并没能正确地或者有意识地去关闭线程池。正确的方法有两种：一个是让所有的入队任务都执行完毕（shutdown()），再就是舍弃这些任务（shutdownNow())——这完全取决于你。比如说如果我们提交了N多任务并且希望等它们都执行完后才返回的话，那么就使用 shutdown()：

private void sendAllEmails(List<String> emails) throws InterruptedException {

    emails.forEach(email ->

      executorService.submit(() ->

        sendEmail(email)));

    executorService.shutdown();

    final boolean done = executorService.awaitTermination(1, TimeUnit.MINUTES);

    log.debug("All e-mails were sent so far? {}", done);

}

本例中我们发送了许多电子邮件，每一封邮件都对应着线程池中的一个任务。提交完这些任务后我们会关闭线程池，这样就不会再有新的任务进来了。然后我们会至少等待一分钟，直到这些任务执行完。如果1分钟后还是有的任务没执行到的话，awaitTermination()便会返回false。但是剩下的任务还会继续执行。我知道有些赶时髦的人会这么写：

emails.parallelStream().forEach(this::sendEmail);

他们觉得我那样很老套，不过我个人比较喜欢能控制并发线程的数量。还有一个优雅地关闭掉线程池的方法就是shutdownNow()：

final List<Runnable> rejected = executorService.shutdownNow();

log.debug("Rejected tasks: {}", rejected.size());

这么做的话队列中的所有任务都会被舍弃并返回。已执行的任务仍会继续执行。

4. 谨慎地处理中断

Future的一个较少提及的特性便是cancelling。这里我就不重复多说了，可以看下我之前的一篇文章： InterruptedException及线程中断。

5. 监控队列长度，确保队列有界

不当的线程池大小会使得处理速度变慢，稳定性下降，并且导致内存泄露。如果配置的线程过少，则队列会持续变大，消耗过多内存。而过多的线程又会由于频繁的上下文切换导致整个系统的速度变缓——殊途而同归。队列的长度至关重要，它必须得是有界的，这样如果线程池不堪重负了它可以暂时拒绝掉新的请求：

final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);

executorService = new ThreadPoolExecutor(n, n,

        0L, TimeUnit.MILLISECONDS,

        queue);

上面的代码等价于Executors.newFixedThreadPool(n)，然而不同的是默认的实现是一个*的 LinkedBlockingQueue。这里我们用的是一个固定100大小的ArrayBlockingQueue。也就是说如果已经有100个任务在队列中了（还有N个在执行中），新的任务就会被拒绝掉，并抛出RejectedExecutionException异常。由于这里的队列是在外部声明的，我们还可以时不时地调用下它的size()方法来将队列大小记录在到日志/JMX/或者你所使用的监控系统中。

6. 别忘了异常处理

下面这段代码执行的结果是什么？

executorService.submit(() -> {

    System.out.println(1 / 0);

});

我被它坑过无数回了：它什么也不会输出。没有任何的java.lang.ArithmeticException: / by zero的征兆，啥也没有。线程池会把这个异常吞掉，就像什么也没发生过一样。如果是你自己创建的java.lang.Thread还好，这样 UncaughtExceptionHandler 还能起作用。不过如果是线程池的话你就得小心了。如果你提交的是Runnable对象的话（就像上面那个一样，没有返回值），你得将整个方法体用try- catch包起来，至少打印一下异常。如果你提交的是Callable<Integer>的话，得确保你在用get()方法取值的时候重新抛出异常：

final Future<Integer> division = executorService.submit(() -> 1 / 0);

//below will throw ExecutionException caused by ArithmeticException

division.get();

有趣的是Spring框架的@Async为此还弄出了个BUG，参见： SPR-8995](https://jira.spring.io/browse/SPR-8995)以及 [SPR-12090 。

7. 监控队列中的等待时间

监控工作队列的长度只是一个方面。然而排除故障时查看从提交任务到实际执行之间的时间差就显得非常重要了。这个时间差越接近0就越好（说明正好线程池中有空闲的线程），否则任务要入队的话这个时间就会增加了。再进一步说，如果线程池不是固定线程数的话，执行新的任务还得新创建一个线程，这个同样也会消耗一定的时间。为了能更好地监控这项指标，可以对ExecutorService做一下封装：

public class WaitTimeMonitoringExecutorService implements ExecutorService {

  private final ExecutorService target;

  public WaitTimeMonitoringExecutorService(ExecutorService target) {

    this.target = target;

  }

  @Override

  public <T> Future<T> submit(Callable<T> task) {

    final long startTime = System.currentTimeMillis();

    return target.submit(() -> {

          final long queueDuration = System.currentTimeMillis() - startTime;

          log.debug("Task {} spent {}ms in queue", task, queueDuration);

          return task.call();

        }

    );

  }

  @Override

  public <T> Future<T> submit(Runnable task, T result) {

    return submit(() -> {

      task.run();

      return result;

    });

  }

  @Override

  public Future<?> submit(Runnable task) {

    return submit(new Callable<Void>() {

      @Override

      public Void call() throws Exception {

        task.run();

        return null;

      }

    });

  }

  //...

}

这个实现并不完整，不过也能说明大概的意思了。当我们将任务提交给线程池的时候，便立即开始记录它的时间。一旦这个任务被取出并开始执行时便停止计时。不要被代码中的startTime和queueDuration这两个变量搞混了。事实上它们是在两个不同的线程中进行求值的，通常都会差个毫秒级或者秒级：

Task com.nurkiewicz.MyTask@7c7f3894 spent 9883ms in queue

8. 保留客户端的栈跟踪信息

近来响应式编程受到了不少关注。 Reactive manifesto](http://www.reactivemanifesto.org/), [reactive streams](http://www.reactive-streams.org/), [RxJava](https://github.com/ReactiveX/RxJava)（仅发布了1.0版本!）,[Clojure agents](http://clojure.org/agents), [scala.rx 等等。它们都非常不错，但栈跟踪信息就完蛋了，它们几乎是毫无价值的。假设提交到线程池中的一个任务出现了异常：

java.lang.NullPointerException: null

  at com.nurkiewicz.MyTask.call(Main.java:76) ~[classes/:na]

  at com.nurkiewicz.MyTask.call(Main.java:72) ~[classes/:na]

  at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0]

  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[na:1.8.0]

  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[na:1.8.0]

  at java.lang.Thread.run(Thread.java:744) ~[na:1.8.0]

可以很容易发现NPE异常出现在MyTask的76行。但是我们并不知道是谁提交的这个任务，因为栈信息只能看到Thread以及 ThreadPoolExecutor。技术上来讲我们当然是可以看下代码，看看是何处创建的MyTask。不过如果没有线程在这中间的话，我们马上便能知道是谁提交的任务。那么如果我们可以保留客户端代码（提交任务的那段代码）的栈信息呢？这个想法并非我首创的， Hazelcast](http://hazelcast.com/)就将[异常从所有者节点传播到了客户端中。下面是一个非常简单的将客户端栈信息保留下来以便失败时查看的例子：

public class ExecutorServiceWithClientTrace implements ExecutorService {

  protected final ExecutorService target;

  public ExecutorServiceWithClientTrace(ExecutorService target) {

    this.target = target;

  }

  @Override

  public <T> Future<T> submit(Callable<T> task) {

    return target.submit(wrap(task, clientTrace(), Thread.currentThread().getName()));

  }

  private <T> Callable<T> wrap(final Callable<T> task, final Exception clientStack, String clientThreadName) {

    return () -> {

      try {

        return task.call();

      } catch (Exception e) {

        log.error("Exception {} in task submitted from thrad {} here:", e, clientThreadName, clientStack);

        throw e;

      }

    };

  }

  private Exception clientTrace() {

    return new Exception("Client stack trace");

  }

  @Override

  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {

    return tasks.stream().map(this::submit).collect(toList());

  }

  //...

}

这样一旦失败的话我们便可以取到完整的栈信息以及提交任务时所在的线程的名字。跟之前相比我们有了一些更有价值的信息：

Exception java.lang.NullPointerException in task submitted from thrad main here:

java.lang.Exception: Client stack trace

  at com.nurkiewicz.ExecutorServiceWithClientTrace.clientTrace(ExecutorServiceWithClientTrace.java:43) ~[classes/:na]

  at com.nurkiewicz.ExecutorServiceWithClientTrace.submit(ExecutorServiceWithClientTrace.java:28) ~[classes/:na]

  at com.nurkiewicz.Main.main(Main.java:31) ~[classes/:na]

  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0]

  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0]

  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0]

  at java.lang.reflect.Method.invoke(Method.java:483) ~[na:1.8.0]

  at com.intellij.rt.execution.application.AppMain.main(AppMain.java:134) ~[idea_rt.jar:na]

9. 优先使用CompletableFuture

Java 8中引入了更为强大的 CompletableFuture 。有可能的话尽量使用下它。ExecutorService并没有扩展以支持这个增强型的接口，因此你得自己动手了。这么写是不行的了：

final Future<BigDecimal> future =

    executorService.submit(this::calculate);

你得这样：

final CompletableFuture<BigDecimal> future =

    CompletableFuture.supplyAsync(this::calculate, executorService);

CompletableFuture 继承自Future，因此跟之前的用法一样。但是使用你接口的人一定会感谢CompletableFuture所提供的这些额外的功能的。

10. 同步队列

SynchronousQueue 是一个非常有意思的BlockingQueue。它本身甚至都算不上是一个数据结构。最好的解释就是它是一个容量为0的队列。这里引用下Java文档中的一段话：

每一个insert操作都需要等待另一个线程的一个对应的remove操作，反之亦然。同步队列内部不会有任何空间，甚至连一个位置也没有。你无法对同步队列执行peek操作，因为仅当你要移除一个元素的时候才存在这么个元素;如果没有别的线程在尝试移除一个元素你也无法往里面插入元素;你也无法对它进行遍历，因为它什么都没有。。。

同步队列与CSP和Ada中所用到的集结管道（rendezvous channel）有异曲同工之妙。

它和线程池有什么关系？你可以试试在ThreadPoolExecutor中用下SynchronousQueue：

BlockingQueue<Runnable> queue = new SynchronousQueue<>();

ExecutorService executorService = new ThreadPoolExecutor(n, n,

        0L, TimeUnit.MILLISECONDS,

        queue);

我们创建了一个拥有两个线程的线程池，以及一个SynchronousQueue。由于SynchronousQueue本质上是一个容量为0 的队列，因此这个ExecutorService只有当有空闲线程的时候才能接受新的任务。如果所有的线程都在忙，新的任务便会马上被拒绝掉，不会进行等待。这在要么立即执行，要么马上丢弃的后台执行的场景中会非常有用。

终于讲完了，希望你能找到一个自己感兴趣的特性！

原创文章转载请注明出处： http://it.deepinmind.com

码农公寓

相关文章