ExecutorService的十个使用技巧

Stella981
• 阅读 569

ExecutorService的十个使用技巧

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 com.nurkiewicz.MyTask@7c7f3894 spent 9883ms in queue
  1. 保留客户端的栈跟踪信息

近来响应式编程受到了不少关注。 Reactive manifestoreactive streamsRxJava(仅发布了1.0版本!),Clojure agentsscala.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只有当有空闲线程的时候才能接受新的任务。如果所有的线程都在忙,新的任务便会马上被拒绝掉,不会进行等待。这在要么立即执行,要么马上丢弃的后台执行的场景中会非常有用。

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

点赞
收藏
评论区
推荐文章
blmius blmius
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
待兔 待兔
4个月前
手写Java HashMap源码
HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Stella981 Stella981
3年前
Opencv中Mat矩阵相乘——点乘、dot、mul运算详解
Opencv中Mat矩阵相乘——点乘、dot、mul运算详解2016年09月02日00:00:36 \牧野(https://www.oschina.net/action/GoToLink?urlhttps%3A%2F%2Fme.csdn.net%2Fdcrmg) 阅读数:59593
Stella981 Stella981
3年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签,将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
Easter79 Easter79
3年前
Twitter的分布式自增ID算法snowflake (Java版)
概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移
Wesley13 Wesley13
3年前
00:Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork),斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟,已经成为Web应用程序的首选开发语言。Java是简单易学,完全面向对象,安全可靠,与平台无关的编程语言。
Wesley13 Wesley13
3年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
10个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这