ExecutorService的十个使用技巧

标签: executorservice 技巧 | 发表时间:2014-11-26 17:13 | 作者:
出处:http://it.deepinmind.com

ExecutorService这个接口从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()。这样的名字表述性不佳。由于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);
  1. 根据上下文切换名字

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

  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。但是剩下的任务还会继续执行。我知道有些赶时髦的人会这么写:

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

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

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

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

  1. 谨慎地处理中断

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

  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 [email protected] 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只有当有空闲线程的时候才能接受新的任务。如果所有的线程都在忙,新的任务便会马上被拒绝掉,不会进行等待。这在要么立即执行,要么马上丢弃的后台执行的场景中会非常有用。

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

原创文章转载请注明出处: ExecutorService的十个使用技巧

英文原文链接

相关 [executorservice 技巧] 推荐:

ExecutorService的十个使用技巧

- - Java译站
ExecutorService这个接口从Java 5开始就已经存在了. 这里小小地提醒一下,官方已经不再支持Java 5, Java 6了,Java 7 在半年后也将停止支持. 我之所以会提起ExecutorService这么旧的一个接口是因为,大多数Java程序员并没有搞清楚它的工作原理. 关于它可以介绍的有很多,这里我只想分享它的一些较少为人所知的特性以及实践技巧.

ExecutorService中submit和execute的区别

- - Java - 编程语言 - ITeye博客
ExecutorService中submit和execute的区别. 在Java5之后,并发线程这块发生了根本的变化,最重要的莫过于新的启动、调度、管理线程的一大堆API了. 在Java5以后,通过Executor来启动线程比用Thread的start()更好. 在新特征中,可以很容易控制线程的启动、执行和关闭过程,还可以很容易使用线程池的特性.

Java定时任务调度:用ExecutorService取代Timer

- - ITeye博客
《Java并发编程》一书提到,用ExecutorService取代Java Timer有几个理由,我认为其中最重要的理由是:. 如果TimerTask抛出未检查的异常,Timer将会产生无法预料的行为. Timer线程并不捕获异常,所以 TimerTask抛出的未检查的异常会终止timer线程. 这种情况下,Timer也不会再重新恢复线程的执行了;它错误的认为整个Timer都被取消了.

Hadoop MapReduce技巧

- - 简单文本
我在使用Hadoop编写MapReduce程序时,遇到了一些问题,通过在Google上查询资料,并结合自己对Hadoop的理解,逐一解决了这些问题. Hadoop对MapReduce中Key与Value的类型是有要求的,简单说来,这些类型必须支持Hadoop的序列化. 为了提高序列化的性能,Hadoop还为Java中常见的基本类型提供了相应地支持序列化的类型,如IntWritable,LongWritable,并为String类型提供了Text类型.

WordPress 技巧

- - CSDN博客互联网推荐文章
WordPress字体设置方法详解.          WordPress开源程序功能越来越强大,未来我们不仅仅可以使用wordpress制作个人博客,还可以使用wordpress程序制作CMS内容管理系统. 很多 Wordpress主题SEO优化的非常好,而且还附带了一些adsense广告位置,让不懂SEO以及代码修改的朋友轻松解决博客优化以及广告位放置问题.

javascript技巧

- - ITeye博客
oncontextmenu="window.event.returnValue=false"  将彻底屏蔽鼠标右键. < table border oncontextmenu=return(false)>< td>no< /table>  可用于Ta bl e. < body onselectstart="return false">  取消选取、防止复制.

linux 小技巧

- - DBA Blog
2:如何限制用户的最小密码长度. 修改/etc/login.defs里面的PASS_MIN_LEN的值. 比如限制用户最小密码长度是8:. 3:如何使新用户首次登陆后强制修改密码. 4:更改Linux启动时用图形界面还是字符界面. 将id:5:initdefault: 其中5表示默认图形界面. 改id:3: initdefault: 3表示字符界面.

面试技巧

- - 非技术 - ITeye博客
问题一:“请你自我介绍一下” .   1、这是面试的必考题目.   2、介绍内容要与个人简历相一致.   3、表述方式上尽量口语化.   4、要切中要害,不谈无关、无用的内容.   5、条理要清晰,层次要分明.   6、事先最好以文字的形式写好背熟. 问题二:“谈谈你的家庭情况” .   1、 况对于了解应聘者的性格、观念、心态等有一定的作用,这是招聘单位问该问题的主要原因.

101 个 Google+ 技巧

- Nicholas - 精品博客
Google+ 虽然诞生不到一个月,并且仍然属于测试版本,但是已经有超过一千万的用户,每天分享十亿条信息. 以下是 101 条让你玩转 Google+ 的技巧,不管你目前是否注册了 Google+. 你只要在这篇文章后面留言就可以获得 Google+ 邀请. 你可以把任何人(包括非 Google+用户)添加进你的圈子里.

Google+ 技巧四则

- Digitalboy(张扬) - 望月的博客
玩Google+也有一段时间了,尽管需要一些特殊的手段(比如修改hosts)才能访问,尽管存在不实名可能会被删除账户的风险,但不得不说,Google+ 的确“有点意思”. 同时,看了很多关于Google+的文章,比如如何修改Hosts顺利访问Google+,比如如何搜索Google+的内容(目前官方只提供用户搜索),比如如何在wordpress的博客中显示Google+的最新内容,再比如15个让你获得更佳用户体验的Google Chrome 扩展等,罗列于此,以馈读者.