ExecutorService – 10个技巧和窍门

ExecutorService抽象从java5一直持续到现在。我们在这里讨论2004，简单提醒一下:java5和java6将不会被支持，java7won’t be in half a year。我提出这个问题的原因是因为很多java程序员仍然不能完全理解ExecutorService的工作原理。有很多地方需要了解，今天我想分享一些鲜为人知的特性和实践。然而这篇文章针对中级程序员，没有特别牛逼的。

1，命名线程池

我不能强调这一点。在dump一个运行的jvm的所有线程或者调试的时候,默认的线程池命名模式是pool-N-thread-M，其中N代表线程池序列号（每次你创建一个新的线程池，全局N计数增加），M是线程池里面线程的序列号。例如pool-2-thread-3表示在jvm进程生命周期里面创建的第二个线程池的第三个线程。参阅：Executors.defaultThreadFactory()。不是很具描述性。JDK使正确命名线程变得稍微有的复杂因为命名策略隐藏在ThreadFactory。幸运的是Guava（google开源的一组工具类集合）工具包有个帮助类来做这件事情：

import com.google.common.util.concurrent.ThreadFactoryBuilder;
//通过ThreadFactoryBuilder这个类来设置线程池的名称并返回一个ThreadFactory
final ThreadFactory threadFactory = new ThreadFactoryBuilder()
        .setNameFormat("Orders-%d")
        .setDaemon(true)
        .build();
final ExecutorService executorService = Executors.newFixedThreadPool(10, threadFactory);

默认情况下线程池创建非守护线程，取决于你是否需要这种类型的线程。

2，根据上下文切换名称

这是一个我从supercharged-jstack-how-to-debug-your-servers-at-100mph学到的技巧，一旦我们记住线程的名称，我们可以在运行随时修改他们！这是有意义的因为线程dump显示了类和方法名称，而不是参数和本地变量。通过调整线程名称来保留一些基本的事物标识符，我们可以容易跟踪哪个消息/记录/查询等很慢或者引起了死锁。例如：

private void process(String messageId) {
    executorService.submit(() -> {
        final Thread currentThread = Thread.currentThread();
        final String oldName = currentThread.getName();
        currentThread.setName("Processing-" + messageId);
        try {
            //这里是业务逻辑...
        } finally {
            currentThread.setName(oldName);
        }
    });
}

在try-finally代码块里面当前线程命名为Processing-WHATEVER-MESSAGE-ID-IS。当追踪经过这个系统的消息的时候这可能会带来便利。

3，明确和安全的关闭线程池

在客户线程（待提交运行的任务）和线程池（执行任务的线程）之间有个任务对列。当你的应用程序关闭的时候，你必须关心两间事情：排队的任务如何处理以及已经在线程池里面的任务怎么运行（稍后会详细介绍）。令人惊讶的是很多开发者并没有正确地或有意识地关闭线程池。有两种技术：让所有排队的任务执行（通过shutdown()这个方法）或者从队列删除他们（通过shutdownNow()这个方法）-完全取决于你的实际情况。例如如果我们想提交一堆任务并且想当它们都完成了才结束，这个情况可以使用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);

在这个场景，我们发送一堆邮件，每一个邮件的发送在线程池里面作为一个单独的任务。在提交这些任务之后我们关闭线程池以至于它不再接收新的任务。这时候我们等待最长一分钟，直到这些任务都完成。然而一些任务仍然未结束，awaitTermination()会返回false。此外，未结束的任务会继续执行。我知道赶时髦的人这样做：

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

称我为老式的，但是我喜欢控制并发线程的数量。不要紧，一个优雅替代shutdown()的是shutdownNow()：

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

这次所有排队任务会被丢弃返回。已经执行的任务任然可以继续执行。

4，谨慎处理中断

Future接口的鲜为人知的功能是取消。查看之前的文章InterruptedException and interrupting threads explained

5,监控队列长度并且让队列有界

大小不正确的线程池可能导致缓慢，不稳定以及内存溢出。如果你配置太少的线程，队列会堆积，耗费很多内存。另一方面太多的线程会减慢整个系统，因为过多的线程上下文切换 - 并导致和之前相当的症状。查看队列的深度并保持有界很重要，因此超载的线程暂时拒绝新的任务。

final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(100);
executorService = new ThreadPoolExecutor(n, n,
        0L, TimeUnit.MILLISECONDS,
        queue);

上面的代码等同于Executors.newFixedThreadPool(n)。而不是使用默认的无界LinkedBlockingQueue我们使用容量大小为100的ArrayBlockingQueue。这意味着如果一百个任务已经在队列里面了（并且n个正在执行）新的任务会被拒绝返回RejectedExecutionException。此外，由于队列现在可以在外部使用，我们可以定期调用size（）并将其放在logs / JMX /任何您使用的监视机制中。

6，记得处理异常

下面代码片段的结果是？

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

我被上面的代码坑过很多次了，它不会打印任何东西。没有 java.lang.ArithmeticException: / by zero 整个的标注，什么都没有。线程池只是吞掉整个异常，好像它从未发生过一样。如果它是一个从头开始创建的线程，UncaughtExceptionHandler 可以工作。但是使用线程池必须更加小心。如果你提交一个Runnable(像上面没有任何结果)你需要用try catche包括代码主体，并且打印日志。如果你提交一个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。

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, reactive streams, RxJava (just released 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]

我们容易地发下在76行MyTask threw NPE，但是我们不知道谁提交了这个任务，因为堆栈跟踪只显示Thread和ThreadPoolExecutor。我们可以在技术上浏览源代码，希望找到一个创建MyTask的地方。但是没有线程（更不用说事件驱动，响应式， actor-ninja-programming）我们可以立即看到全貌。如果我们可以保留客户端代码的堆栈并且显示它，例如如果失败了？这个想法并不新鲜，例如Hazelcast from owner node to client code。这就是在出现故障时保持客户端堆栈跟踪的天真支持：

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，所以一切都像以前一样工作。 但是，API的更高级消费者将真正欣赏CompletableFuture提供的扩展功能。

10，同步队列

SynchronousQueue是一个有趣的BlockingQueue，它不是真正的队列。 它本身甚至都不是数据结构。 最好将其解释为容量为0的队列。引用JavaDoc：
”each insert operation must wait for a corresponding remove operation by another thread, and vice versa. A synchronous queue does not have any internal capacity, not even a capacity of one. You cannot peek at a synchronous queue because an element is only present when you try to remove it; you cannot insert an element (using any method) unless another thread is trying to remove it; you cannot iterate as there is nothing to iterate. […]

Synchronous queues are similar to rendezvous channels used in CSP and Ada.“

这个怎么和线程池关联上呢？尝试将SynchronousQueue与ThreadPoolExecutor一起使用：

BlockingQueue<Runnable> queue = new SynchronousQueue<>();
ExecutorService executorService = new ThreadPoolExecutor(n, n,
        0L, TimeUnit.MILLISECONDS,
        queue);

我们创建了一个带有两个线程的线程池，并在它前面有一个SynchronousQueue。因为SynchronousQueue本质上是一个容量为0的队列，所以如果有可用的空闲线程，这样的ExecutorService将只接受新任务。 如果所有线程都忙，新任务将立即被拒绝，永远不会等待。 当在后台处理必须立即开始或被丢弃时，这个执行方式是被期待的。

就是这样，我希望你找到至少一个有趣的功能！

最后英文原文

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,378评论 6赞 481
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,356评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 152,702评论 0赞 342
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,259评论 1赞 279
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,263评论 5赞 371
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,036评论 1赞 285
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,349评论 3赞 400
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,979评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,469评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,938评论 2赞 323
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,059评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,703评论 4赞 323
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,257评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,262评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,485评论 1赞 262
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,501评论 2赞 354
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,792评论 2赞 345