ExecutorService的十个使用技巧

时间  2019-11-06
标签 executorservice 十个 使用 技巧 繁體版
原文   原文链接

ExecutorService的十个使用技巧

ExecutorService这个接口从Java 5开始就已经存在了。这得追溯到2004年了。这里小小地提醒一下，官方已经再也不支持Java 5, Java 6了，Java 7在半年后也将中止支持。我之因此会提起ExecutorService这么旧的一个接口是由于，大多数Java程序员并无搞清楚它的工做原理。关于它能够介绍的有不少，这里我只想分享它的一些较少为人所知的特性以及实践技巧。本文主要是面向初级程序员的，并无过于高深的东西。html

  1. 线程命名

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

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);
  1. 根据上下文切换名字

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

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。这对跟踪系统内的消息流会比较有用。git

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

客户端线程和线程池之间会有一个任务队列。当程序要关闭时，你须要注意两件事情：入队的这些任务的状况怎么样了以及正在运行的这个任务执行得如何了。使人惊讶的是不少开发人员并没能正确地或者有意识地去关闭线程池。正确的方法有两种：一个是让全部的入队任务都执行完毕（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。可是剩下的任务还会继续执行。我知道有些赶时髦的人会这么写：github

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

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

final List<Runnable> rejected = executorService.shutdownNow(); log.debug("Rejected tasks: {}", rejected.size());

这么作的话队列中的全部任务都会被舍弃并返回。已执行的任务仍会继续执行。编程

  1. 谨慎地处理中断

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

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

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

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/或者你所使用的监控系统中。

  1. 别忘了异常处理

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

executorService.submit(() -> {     System.out.println(1 / 0); });

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

final Future<Integer> division = executorService.submit(() -> 1 / 0);
//below will throw ExecutionException caused by ArithmeticException
division.get();

有趣的是Spring框架的@Async为此还弄出了个BUG，参见：SPR-8995以及 SPR-12090。

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

监控工做队列的长度只是一个方面。然而排除故障时查看从提交任务到实际执行之间的时间差就显得很是重要了。这个时间差越接近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
  1. 保留客户端的栈跟踪信息

近来响应式编程受到了很多关注。 Reactive manifesto, reactive streams, RxJava（仅发布了1.0版本!）,Clojure 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就将异常从全部者节点传播到了客户端中。下面是一个很是简单的将客户端栈信息保留下来以便失败时查看的例子：

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]
  1. 优先使用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所提供的这些额外的功能的。

  1. 同步队列

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只有当有空闲线程的时候才能接受新的任务。若是全部的线程都在忙，新的任务便会立刻被拒绝掉，不会进行等待。这在要么当即执行，要么立刻丢弃的后台执行的场景中会很是有用。

终于讲完了，但愿你能找到一个本身感兴趣的特性！

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程