ExecutorService-10个要诀和技巧-编程开发技术

1、exccutorscrvicc-10个要诀和技巧-编程 开发技术executorservice-10个要诀和技巧译文出处：simonwang丿京文出处：tomasz nurkiewiczexecutorservice抽象概念自jdva5就已经提出來了，现在是2014年。顺便提 醒一下：java5和jqvei6都已不被支持，jqvej在半年内也将会这样。我提出这 个的原因是许多java程序员仍然不能完全明白executorservice到底是怎样工 作的。还有很多地方要去学习，今天我会分享一些很少人知道的特性和实践。然 而这篇文章仍然是面向屮等程序员的，没什么特别高级的地方。1. name po

2、ol threads我想强调一点的是，当在运行jvm或调试期间创建线程时，默认的线程池命名规 则是pool-n-thread-m,这里n代表线程池的序列数(每一次你创建一个线程池 的吋候，全局计数n就加1),而m则是某一个线程池的线程序列数。例如， pool-2-thread-3就意味着jvm生命周期中第2线程池的第3线程。具体可以查 看：executors. dcfaultthrcadfactory () o这样不具备描述性，jdk使得线程命 名的过程冇些微的复杂，因为命名的方法隐藏在threadeactory内部。幸运地是 guava有一个很有用的类：import com. google,

3、 common. ut订.concurrent. threadfactorybuiider;final threadeactory thrcadfactory = new threadfactorybuiider().setnameformat (，z0rders-%d，z).setdaemon(true).build();f inal executorservice executorservi ce 二 executors. newfixedthreadpool (10, thrcadfactory);线程池默认创造的是非守护线程，由你来决定是否合适。2. switch names acco

4、rding to context冇一个我从?supercharged jstack: how to debug your servers at 1 oomph 学 到的小技巧。一旦我们记住了线程的名字，那么在任何时刻我们都能够改变它们！ 这是冇道理的，因为线程转储显示了类名和方法名，没冇参数和局部变量。通过 调整线程名保留一些必要的事务标识符，我们可以很容易追踪某一条运行缓慢或 者造成死锁的信息/记录/查询等。例如：private void process (string messageld)executorservice.submit() -> final thread current

5、thread = thread.currentthread (); final string oldname 二 currentthread. getnameo ; currentthread. setname(processing-，z + messagetd); try /real logic here. finally currentthread. setname(oldname);)；在try-finally块内部，当前线程被命名为process!ng-whatever-message-td-ts,当通过系统追踪信息流时这可能会派上 用场。3. explicit and safe sh

6、utdown在客户端线程和线程池z间冇一个任务队列，当你的应用关闭时，你必须关心两 件事：任务队列会发生什么；正在运行的任务会怎样(这个时候将详细介绍)。令 人感到吃惊的是许多程序员并不会适当地或有意识地关闭线程池。这有两个方 法：要么让所有的任务队列全都执行完(shutdowno),要么舍弃它们 (shutdownnowo),这依赖你使用的具体情况。例如如杲我们提交一连吊的任务 并月想要它们在完成后尽可能快的返回，可以使用shutdown()：private void sendallemails(lising> emails) throwsinierruptedexcept ion e

7、mailsforeach(ema订 ->executorservice submit() -sendemai1 (email);executorservice. shutdown();final boolean done = executorservice.awaittermination(l, timeunit. minutes);logdcbug("all e-mails were sent so far? ", done);在这个例子屮我们发送了一堆e-ma订，每一个都作为一个独立的任务交给线程 池。在提交了所有的任务之后我们执行shutdown使线程池不再接收

8、新的任务。 然后最多等待lminute直到所冇的任务都完成。然而如果冇些任务仍然处于挂起 状态，awaittcrmination()将返回false,而那些在等待的任务会继续执彳亍。我 知道一些人会使用新潮的用法：emails, parallelstream(). foreach(this:sendemai1); 你可能会觉得我太保守，但我喜欢去控制并行线程的数量。不用介意，述冇一种 优雅的 shutdown()方法 shutdownnowo :final list<rurmable> rejected = executorservice.shutdownnow();log. de

9、bug ("rejected tasks: ", rejected, size ();这样一来队列屮述在等待的任务将会被舍弃并被返回，但已经在运行的任务将会 继续。4. handle interruption with care很少人知道future接口的cancel,这里我不想重复说明，你可以去看我以前的 文章：intcrruptcdexccption emd intcrrupting threads cxplaincd5. monitor queue length and keep it bounded不合适的线程池大小可能会造成运行缓慢、不稳定以及内存泄漏。如果你配置

10、太 少的线程，那么任务队列就会变大，消耗太多内存。另一方面太多的线程又会由 于过度频繁的上下文切换而造成整个系统运行缓慢。所以观察队列的长度并将其 限定在一定范围内是很重要的，这样的话过载的线程池会短暂拒绝新任务的提 交：final blockingqueue<runnable> queue 二 new arrayblockingqueueo(100); executorservice 二 new threadpoolexecutor(n, n,0l, timeunit. milliseconds,queue);上面的代码和executors. newfixedthreadpool

11、 (n)是等价的，然而不同的是默认 情况下固定线程池使用的是无限制的linkedblockingqueue ,我们使用的是固 定容量100的arrayblockingqueueo这就意味着如果已经冇100个任务在排队 (其中有n个任务止在执行)，那么新的任务就将被驳回并抛出 rejectedexecutionexceptiono 一旦在外部可以访问queue ,那么我们就可以 周期性地调用sizeo,并把它提交到logs/jmx或其他任何你使用的监视器中。6. remember about exception handling卜面代码段的结果是什么？executorservice.submit

12、() -> system, out. printin (1 / 0););我深受其苦：它不会打卬任何东西。不会抛出java. lang. arithmeticexception: / by zero,什么也没有。线程池将忽略这个异常，就像它从来没发生过。如果 上面的代码是用java. lang. thread偶然创造的，那么 uncaughtexccptionhandlcr可能会起作用。但在线程池里你就要多加小心了。 如果你正在提交runnable (没有返回结果，就像上面)，那么你必须将整个代码 块用try-catch包起來，至少要log下。如果你提交的是callable,确保你 总是

13、使用阻塞的get ()方法来重抛异常：final future<tnteger> division 二 executorservice. submit() -> 1 / 0); /below will throw executionexception caused by arithmeticexccption division. get ();有趣的是就算是spring框架在处理这个bug的时候会使用©async,详 细：7spr-8995 和 spr-12090。7. monitor waiting time in a queue监控工作队列深度又是一个层面，在排

14、除单个事务或任务的故障时，有必要了解 从任务的捉交到实际执行耗时多长。这种等待时间最好趋近于零(当线程池屮冇 空闲的线程时)，但任务乂不得不在队列中排队导致等待时间变长。而且如果池 内没有-定数量的线程，在运行新任务吋可能需要创造新的线程，而这个过程也 是要消耗少量时间的。为了能够清楚地监测这个吋间，我们使用类似下面的代码 包装原始的 executorservice :public class waittimemonitoringexecutorservice implements executorservice private final executorservice target;pub

15、lic waittimemonitoringexecutorservice(executorservice target) this.target = target;©overridepublic <t> future<t> submit(callable<t> task) final long starttime = system. currenttimemillis();return target, submit () -> final long queueduration 二systemcurrcnttimcmillis() - sta

16、rttime;log .debug ("task spent ms in queue", task,queueduration);return task, call ();©overridepublic <t> future<t> submit(runnable task, t result) return submit() -> task, run ();return result;)；©overridepublic future<?> submit(runnable task) rcturn submit (

17、new callablc<void>() ©overridepublic void call () throws exception task, run (); return nul1;)；/.这并不是完整的实现，但你得知道这个基本概念。当我们向线程池提交任务的那 一刻，就立马开始测量时间，而任务一开始被执行就停止测量。不要被上面源码 中很接近的starttime和queueduration所迷惑了，事实上这两行是在不同的 线程中执行的，可能有数毫秒甚至数秒的差别，例如：task com.nurkicwiczmyteisk7c7f3894 spent 9883ms in

18、queue8. preserve client stack trace响应式编程这段日子似乎比较火，reactive manifesto, reactive streams, rxjava(刚冈lj发布 1. 0), clojure agents, scala. rx，这些东西都挺好 的，但它们的堆栈跟踪将不再友好，大多数堆栈跟踪没有什么卵用。举个例子， 当线程池中的任务抛岀了一个异常：java.lang.nullpointerexception: nullat com. nurkiewicz. mytask. call (main. java:76) 'classes/:naat c

19、om. nurkiewicz. mytask. cal 1(main. java:72) classes/:naat java uti1. concurrent. futuretask run(futuretask java:266)na: 1. 8. 0atjava. util, concurrent. threadpoolexecutor. runworker(threadpoolexecutor, java: 1142)、na: 1. 8. 0atjava. util, concurrent. threadpoolexecutor$worker. run(threadpoolexecut

20、or .java:617) na: 1. 8. 0at java. lang. thread, run (thread, java: 744) na: 1. 8. 0我们很容易就发现mytask在76行抛出了空指针异常，但我们并不知道是谁提交 了这个任务，因为堆栈跟踪仅仅只是告诉你thread和threadpoolexecutor的 信息。我们能通过源码从技术上定位mytask被创造的位置，不需要线程(更不必 说事件驱动、响应式编程)我们就能够马上看到全面信息。如果我们保留客户端 代码(提交任务的代码)的堆栈跟踪并在岀现故障的时候将其打印岀來会怎么 样？这不是什么新想法，例如hazclcast

21、会将当前点发生的异常传送回客户端代 码，下面就看看保持客户端堆栈跟踪是怎样实现的：public class executorservicewithc1ienttrace implements executorservice protected final executorservice target;public executorservicewithclienttrace(executorservice target) this.target 二 target;©overridepublic <t> future<t> submit(callable<

22、t> task) return target, submit (wrap(task, clienttraceo,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 (，zexc

23、eption in task submitted from thrad here:", e, dientthreadmame, dientstack);throw e;； private exception dienttrace() return new exception (z/c1 i e nt st ack t race");©overridepublic <t> listfuturet>> invokeall(collection？ extends cal 1able<t>> tasks) tbrows tnter

24、ruptedexception return tasks, stream(). map(this:submit). collect(tolist(); /.这次一旦岀现异常我们将检索任务被提交地方的所有堆栈跟踪和线程名，和标准 异常相比下面的异常信息更有价值：exception java. lang. nullpointerexception in task submitted from thrad main here:java, lang. exccption: client stack traceatcom. nurkiewicz. executorservicewithclienttra

25、ce. clienttrace (executorser vicewithclienttrace. java:43) "classes/:naatcom. nurkicwicz. exccutorscrviccwithclienttrace, submit (exccutorscrviccw ithclienttrace. java:28) classes/:naat com nurkiewicz. main, main(main. java:31) classes/:naat sun. reflect. nativemethodaccessorlmpl. invokeo(nativ

26、e method) na: 1. 8. 0atsun. reflect. nativemethodaccessorlmpl. invoke(nativemethodaccessorlmpl. java:62) na:l. 8. 0atsun. reflect. delegatingmethodaccessortmpl invoke(delegatingmethodacces sorlmpl. java:43) na: 1. 8. 0at java. lemg. reflect. method, invokc(mcthod. java:483) na: 1. 8. 0 atcom. intell

27、ij. rt. execution, application. appmain. main(appmain. java:134) "idea_rt. jar:na9. prefer completablefuturejava 8 提出了强大的 complctablcfuturc,请尽可能的使用它。exccutorscrvicc 并没冇扩展支持这个强大的抽彖，所以你要小心使用它。用：final completablefuture<bigdecimal> future 二completablefuture, supplyasync(this:calculate, executorservice);代替：final futurc<bigdccimal> future 二executorservice. submit (this:calculate);completablefuture 继承了 future 及其所有功能，而且 completablefuture 所 提供的扩展功能极大地丰富了我们的apto10. synchronous queuesynchronousqueue是一种冇趣的blockingqueue但真正意义上并不是queue,事 实上它连数据