上一章介绍了线程池如何复用工作线程。线程池中的Worker会不断取得任务并调用任务的run()while((taskgetTask())!null){task.run();}这解决了线程反复创建和销毁的问题但也带来一个新问题Runnable.run()没有返回值。如果一个计算任务在线程池的工作线程中执行提交任务的线程怎样取得计算结果如果任务执行时抛出异常异常又怎样传回提交任务的线程Java 使用Callable、Future和FutureTask共同解决这个问题。Callable表示有返回值的任务Future表示异步结果的凭证FutureTask则把“可以被工作线程执行”和“可以被提交线程查询结果”这两件事合在同一个对象里。一、Runnable 为什么不够用在线程池中工作线程最终执行的是task.run();所以不论原始任务是什么进入线程池执行流程后都需要表现为一个Runnable。对于不需要返回结果的任务这没有问题executor.execute(()-{handleRequest();});但如果任务需要计算并返回结果Runnable就不够用了。它的核心方法是publicvoidrun();返回类型是void不能通过return把计算结果返回给提交任务的线程。即使在run()内部定义局部变量那个变量也只存在于工作线程的调用栈中不会自动出现在提交线程的栈里。更关键的是方法返回值只能返回给当前方法的直接调用者。run()是由工作线程调用的返回路径也只回到工作线程的执行流程不能直接跨线程回到提交任务的线程。因此要在线程池中表达“有返回值的任务”需要引入Callable。二、Callable 和 Future 分别解决什么问题CallableV用来表示有返回值的任务publicinterfaceCallableV{Vcall()throwsException;}例如CallableIntegertask()-{return1020;};Callable和Runnable的区别可以概括为接口方法特点Runnablevoid run()没有返回值不能直接抛出受检异常CallableVV call()有返回值可以直接抛出受检异常Callable解决的是“任务如何产生结果”。但它还没有解决“提交线程以后如何拿到结果”。使用submit()提交Callable后线程池会返回一个FutureFutureIntegerfutureexecutor.submit(()-{return1020;});可以先把Future理解为异步任务结果的凭证。提交线程拿到Future时任务可能还没开始也可能正在执行甚至可能已经完成。Future本身不表示任务已经结束它只提供了后续查询、等待、取消和获取结果的入口。常用方法包括方法含义get()等待并获取任务结果get(timeout, unit)最多等待指定时间cancel(boolean)尝试取消任务isDone()判断任务是否已经结束isCancelled()判断任务是否已经被取消所以这一层可以先建立两个分工Callable负责定义有结果的任务Future负责让另一个线程以后取得这个异步结果。三、get()、超时和取消分别表示什么提交任务后工作线程可能还没有执行完成FutureIntegerfutureexecutor.submit(()-{Thread.sleep(3000);return100;});Integerresultfuture.get();调用get()的线程会检查任务是否已经完成。如果任务已经完成get()立即处理结果如果任务还没完成调用线程进入等待。阻塞的是调用get()的线程不是整个线程池其他工作线程仍然可以继续执行任务。普通get()没有等待时间上限。如果任务一直没有结束调用线程就可能一直等待。为了避免无限等待可以使用带超时的方法Integerresultfuture.get(3,TimeUnit.SECONDS);如果三秒内任务完成get()返回结果如果三秒后任务仍未完成get()抛出TimeoutException。但超时只表示调用线程不再等待并不表示后台任务已经停止。如果任务超时后已经没有继续执行的价值可以尝试取消future.cancel(true);参数mayInterruptIfRunning表示如果任务已经开始执行是否允许通过中断请求尝试停止任务。cancel(false)不会中断正在运行的任务cancel(true)会尝试中断执行任务的工作线程。这里要注意两点。第一如果任务还在队列中没有开始执行取消成功后它不会再被正常执行。第二如果任务已经开始执行cancel(true)也不是强制杀死线程只是发出中断请求。任务是否真正停止取决于任务代码是否响应中断。例如sleep()、wait()、阻塞队列等阻塞方法通常能通过InterruptedException响应中断普通计算循环则需要主动检查中断状态。while(!Thread.currentThread().isInterrupted()){calculate();}如果取消成功之后调用future.get()会抛出CancellationException。四、isDone() 不等于任务成功isDone()只能说明任务已经进入结束状态但不能说明任务是正常成功、执行失败还是被取消。isDone() true may mean: success, failure, or cancellationisCancelled()用来判断任务是否被取消。只要isCancelled()为trueisDone()一定也为true但isDone()为true时不一定是取消也可能是正常完成或执行失败。因此要判断任务最终结果仍然需要结合get()get()的结果任务状态正常返回任务执行成功抛出ExecutionException任务执行失败抛出CancellationException任务已经被取消抛出TimeoutException当前调用线程等待超时任务未必结束如果任务已经正常完成再调用cancel(true)会返回false因为任务已经进入最终状态不能再改成取消状态。任务执行失败后也是同理。五、为什么需要 FutureTaskFuture规定了如何查询结果、等待和取消任务但工作线程需要的是能执行的Runnable。线程池需要一个对象同时具有两种能力身份给谁使用作用Runnable工作线程调用run()执行任务Future提交线程调用get()获取结果、异常或取消状态这个对象就是FutureTask。FutureTask实现了RunnableFuturepublicclassFutureTaskVimplementsRunnableFutureV{}而RunnableFuture同时继承Runnable和FuturepublicinterfaceRunnableFutureVextendsRunnable,FutureV{}因此FutureTask可以近似理解为FutureTask Runnable task Future result holder它既包装了需要执行的原始任务又负责保存任务的结果、异常和状态。六、submit() 内部如何把 Callable 变成 FutureTask假设提交一个CallableFutureIntegerfutureexecutor.submit(()-{return1020;});在线程池的默认实现中可以把内部过程简化成CallableIntegercallable()-1020;FutureTaskIntegerfutureTasknewFutureTask(callable);execute(futureTask);returnfutureTask;这里发生了三件事原始Callable被包装成FutureTaskFutureTask作为Runnable提交给线程池同一个FutureTask又作为Future返回给提交线程。也就是说工作线程执行的对象和提交线程拿到的Future通常是同一个堆对象。提交线程和工作线程各自有独立的线程栈但它们都持有同一个FutureTask引用。工作线程通过这个对象执行任务并写入结果提交线程通过这个对象等待并读取结果。七、FutureTask.run() 如何执行原始任务原始的Callable在创建FutureTask时已经保存在对象内部CallableIntegercallable()-1020;FutureTaskIntegerfutureTasknewFutureTask(callable);工作线程后来只需要调用futureTask.run();FutureTask.run()会从自身内部取出之前保存的Callable然后调用callable.call();如果任务正常完成call()的返回值先返回给当前工作线程中的FutureTask.run()再由FutureTask保存到自己的字段中。这个返回值不会直接跨线程回到提交线程。可以简化为publicvoidrun(){Vvaluecallable.call();outcomevalue;stateNORMAL;}真实实现还要处理并发状态、取消、异常和等待线程唤醒。当前先抓住主线Callable.call()在工作线程中执行结果或异常被写入FutureTask提交线程之后通过同一个FutureTask.get()读取这些信息。八、state 和 outcome 分别负责什么FutureTask至少要保存两类信息任务结束到了什么状态以及任务结束后产生了什么内容。state负责说明任务状态简化状态含义Not Completed任务还没完成Success任务正常完成Failed任务执行失败Cancelled任务已被取消outcome负责保存任务结束后的内容任务结果outcome 保存什么正常完成返回值执行失败异常对象被取消没有正常结果二者必须分开因为outcome null不能说明任务处于什么状态。任务可以正常返回null任务还没完成时outcome也可能是null任务取消后也没有正常结果。因此必须由state决定怎样解释outcome。例如任务正常返回一个异常对象也是合法的CallableThrowabletask()-{returnnewIOException(这是任务的正常返回值);};此时state是正常完成outcome是返回值不能因为outcome instanceof Throwable就把它当作任务失败。真正的失败必须由state EXCEPTIONAL这类状态来表达。九、真实 FutureTask 的状态如何理解真实FutureTask的状态比“成功、失败、取消”更细。它内部定义了这些状态状态含义NEW任务未进入最终状态COMPLETING正在写入结果或异常NORMAL任务正常完成EXCEPTIONAL任务执行时抛出异常CANCELLED任务取消不中断运行线程INTERRUPTING正在向运行线程发送中断请求INTERRUPTED中断请求已经发出正常完成时状态从NEW经过短暂的COMPLETING最终到达NORMAL。执行失败时同样先进入COMPLETING再到达EXCEPTIONAL。调用cancel(false)成功时状态进入CANCELLED调用cancel(true)成功时状态会经历INTERRUPTING再进入INTERRUPTED。COMPLETING和INTERRUPTING都是短暂过渡状态。它们的作用不是给业务代码判断而是让结果写入、状态发布和中断请求之间保持正确顺序。理解Future的主要行为时可以先把最终状态归纳为三类正常完成、执行失败、被取消。十、为什么要先写 outcome再发布 statestate相当于“任务是否已经完成”的发布标志。提交线程调用get()时会先读取state如果发现任务已经完成再读取outcome。因此工作线程完成任务时必须先准备结果再发布完成状态write outcome publish state如果顺序反过来就可能出现提交线程先看到完成状态随后立即读取outcome但工作线程还没有把结果写进去。FutureTask对状态读写提供了相应的内存可见性保证。提交线程看到完成状态后也能看到完成状态发布之前写入的结果或异常。这个模式和前面讲过的 volatile 发布思想相似先写普通数据再写发布标志另一个线程先看到发布标志再读取普通数据。十一、get() 如何处理结果、异常和取消调用Vresultfuture.get();时FutureTask会先检查state。如果任务还没完成调用线程等待如果任务正常完成返回outcome如果任务执行失败抛出ExecutionException如果任务被取消抛出CancellationException。任务执行失败时FutureTask保存的是原始异常对象CallableIntegertask()-{thrownewIOException(文件读取失败);};工作线程执行失败后可以理解为state EXCEPTIONAL outcome IOException object提交线程调用get()时FutureTask会把这个原始异常包装成ExecutionExceptiontry{Integerresultfuture.get();}catch(ExecutionExceptione){Throwablecausee.getCause();System.out.println(cause.getClass());System.out.println(cause.getMessage());}为什么要包装一层因为get()需要用统一方式告诉提交线程任务本身失败了。原始异常仍然保存在ExecutionException.getCause()中。这样既能区分“任务失败”和“任务正常返回一个异常对象”又能保留异常类型、错误信息和调用栈。十二、Future.get() 的可见性边界Future还提供内存一致性保证异步任务中的操作 happens-before 另一个线程从对应的Future.get()成功返回后的操作。例如classData{intvalue;}DatadatanewData();FutureIntegerfutureexecutor.submit(()-{data.value100;return200;});intresultfuture.get();System.out.println(result);System.out.println(data.value);工作线程在任务中写入data.value 100随后任务完成并发布状态。主线程从future.get()返回后不仅能拿到返回值200也能看到任务完成前对共享对象的写入。这里的同步边界是“对应的Future.get()”。不能把它扩大成只要任务提交给线程池所有线程在任何时刻都一定能看到任务中的修改。如果另一个线程没有通过这个Future建立同步关系仍然需要使用正确的共享数据同步机制。十三、submit(Runnable) 为什么也能返回 FutureRunnable.run()没有返回值但线程池仍然允许Future?futureexecutor.submit(()-{handleRequest();});这是因为线程池可以把Runnable转换成一种没有业务返回值的CallableCallableObjectcallable()-{runnable.run();returnnull;};然后再创建FutureTask。因此submit(Runnable)返回的Future仍然可以用于等待任务完成、判断任务状态、取消任务和获取任务异常只是任务正常完成时get()返回null。还有一个重载方法TFutureTsubmit(Runnabletask,Tresult);例如FutureStringfutureexecutor.submit(()-handleRequest(),完成);Stringresultfuture.get();如果Runnable正常完成get()会返回提交时指定的固定结果。这不是Runnable自己计算出的返回值而是调用submit()时提前提供的结果。三种提交方式可以对比为提交方式get()结果submit(CallableV)call()返回的结果submit(Runnable)nullsubmit(Runnable, V result)提交时指定的固定结果十四、execute() 和 submit() 如何处理异常execute()和submit()都能把任务交给线程池但二者对任务异常的处理方式不同。使用execute()时原始Runnable直接交给线程池。工作线程取得任务后调用task.run()。如果任务中的运行时异常没有在任务内部被捕获异常会从run()中逃出导致当前工作线程异常结束。executor.execute(()-{thrownewIllegalStateException(执行失败);});结束的是执行该任务的工作线程不是整个线程池。线程池随后会清理已经结束的Worker并根据线程池状态、队列状态和工作线程数量决定是否补充新的Worker。失败的任务不会被线程池自动重试异常发生前已经完成的业务操作也不会自动回滚是否需要补偿要由业务代码负责。如果任务自己捕获了异常异常没有逃出run()工作线程就不会因此结束仍然可以继续从队列取得后续任务。使用submit()时原始任务会先被包装成FutureTask。任务抛出的异常会被FutureTask捕获并保存状态变为异常完成异常不会逃出FutureTask.run()工作线程通常可以继续执行后续任务。Future?futureexecutor.submit(()-{thrownewIllegalStateException(执行失败);});提交线程需要调用future.get()才能发现任务失败try{future.get();}catch(ExecutionExceptione){Throwablecausee.getCause();}二者的差异可以概括为提交方式执行对象异常处理execute()原始Runnable异常可能逃出run()导致当前工作线程结束submit()FutureTask异常被保存到FutureTask通过get()取出这也解释了为什么使用submit()后如果从不调用get()任务异常可能很容易被忽略。异常并没有消失只是被保存在FutureTask中。十五、立即调用 get() 可能让并行任务重新串行化下面的代码虽然使用了线程池但每提交一个任务就立即等待for(inti0;i10;i){FutureIntegerfutureexecutor.submit(()-calculate());Integerresultfuture.get();}执行过程会变成提交第一个任务等待第一个任务完成再提交第二个任务等待第二个任务完成。这样任务接近串行执行无法充分利用线程池的并行能力。更合理的方式通常是先提交多个任务ListFutureIntegerfuturesnewArrayList();for(inti0;i10;i){futures.add(executor.submit(()-calculate()));}然后再统一获取结果for(FutureIntegerfuture:futures){Integerresultfuture.get();System.out.println(result);}这样多个工作线程可以并发执行任务。不过依次调用get()时如果第一个Future对应的任务很慢即使后面的任务已经完成当前线程仍然会先等待第一个任务。因此Future适合表示单个异步任务的结果但不擅长处理复杂的任务组合、完成顺序和异步回调。十六、FutureTask 的完整执行过程把前面的内容合在一起一次submit()调用可以概括为工作线程一侧提交线程一侧FutureTask的核心价值就在于它连接了两个线程工作线程负责执行任务并写入结果提交线程负责等待并读取结果。任务的返回值和异常不能直接跨线程返回但可以通过同一个堆上的FutureTask对象完成传递。本章总结FutureTask的因果链条可以从Runnable.run()的限制开始理解线程池中的工作线程只会调用run()而run()没有返回值也不能把工作线程栈上的局部变量直接交给提交线程。于是 Java 引入Callable表达有返回值的任务再引入Future作为提交线程获取异步结果的凭证。但工作线程需要的是Runnable提交线程需要的是Future所以线程池在submit()时会创建FutureTask。同一个FutureTask一边作为Runnable进入线程池被工作线程调用run()一边作为Future返回给提交线程被提交线程调用get()、cancel()、isDone()等方法。工作线程在FutureTask.run()中调用原始任务。任务正常完成时返回值写入outcome状态发布为NORMAL任务失败时异常对象写入outcome状态发布为EXCEPTIONAL任务被取消时状态进入取消相关的最终状态。state决定如何解释outcome因此不能只看outcome是否为null也不能只看outcome是否是异常对象。最终FutureTask不只是一个结果容器。它把任务执行、结果保存、异常传递、取消状态、等待唤醒和内存可见性组织到同一个对象中使提交线程能够在另一个工作线程完成任务后可靠地拿到结果、异常或取消状态。