Java并发性-任务反馈

从我上一篇关于java.util.concurrent包的文章的结尾处摘下来，这很有趣，有时甚至必须在并发任务启动后从它们那里获得反馈。

例如，假设一个应用程序必须发送电子邮件批处理，除了使用多线程机制外，您还想知道成功发送了多少预期的电子邮件，以及在实际发送过程中，整个邮件的实时进度批量。

为了实现带有反馈的这种多线程，我们可以使用 Callable 接口。该接口的工作方式与 Runnable 大致相同，但是执行方法（call（））返回的值应反映所执行计算的结果。

让我们首先定义将执行实际任务的类：

package com.ricardozuasti;import java.util.concurrent.Callable;public class FictionalEmailSender implements Callable<Boolean> {public FictionalEmailSender (String to, String subject, String body){this.to = to;this.subject = subject;this.body = body;}@Overridepublic Boolean call() throws InterruptedException {// Simulate that sending the email takes between 0 and 0.5 secondsThread.sleep(Math.round(Math.random()* 0.5 * 1000));// Lets say we have an 80% chance of successfully sending our emailif (Math.random()>0.2){return true;} else {return false;}}private String to;private String subject;private String body;
}

请注意， Callable可以使用任何返回类型，因此您的任务可以返回所需的任何信息。

现在，我们可以使用线程池ExecutorService发送电子邮件，并且由于我们的任务是作为Callable实现的，因此对于提交给执行的每个新任务，我们都会获得Future参考。请注意，我们将使用直接构造函数而不是Executors中的实用程序方法来创建ExecutorService ，这是因为使用特定类（ ThreadPoolExecutor ）提供了一些方便的方法（在ExecutorService接口中不存在）。

package com.ricardozuasti;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;public class Concurrency2 {public static void main(String[] args) {try {ThreadPoolExecutor executor = new ThreadPoolExecutor(30, 30, 1, TimeUnit.SECONDS,new LinkedBlockingQueue());List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>(9000);// Lets spam every 4 digit numeric user on that silly domainfor (int i = 1000; i < 10000; i++) {futures.add(executor.submit(new FictionalEmailSender(i + '@wesellnumericusers.com','Knock, knock, Neo', 'The Matrix has you...')));}// All tasks have been submitted, wen can begin the shutdown of our executorSystem.out.println('Starting shutdown...');executor.shutdown();// Every second we print our progresswhile (!executor.isTerminated()) {executor.awaitTermination(1, TimeUnit.SECONDS);int progress = Math.round((executor.getCompletedTaskCount() * 100) /executor.getTaskCount());System.out.println(progress + '% done (' + executor.getCompletedTaskCount() +' emails have been sent).');}// Now that we are finished sending all the emails, we can review the futures// and see how many were successfully sentint errorCount = 0;int successCount = 0;for (Future<Boolean> future : futures) {if (future.get()) {successCount++;} else {errorCount++;}}System.out.println(successCount + ' emails were successfully sent, but '+ errorCount + ' failed.');} catch (Exception ex) {ex.printStackTrace();}}
}

在将所有任务提交给ExecutorService之后，我们开始关闭它（防止提交新任务），并使用循环（在实际情况下，如果可能，您应该继续执行其他操作）等待所有任务完成，然后进行计算并打印到目前为止每次迭代的进度。请注意，您可以随时存储执行程序引用并从其他线程查询它，以计算和报告流程进度。

最后，使用对每个提交给ExecutorService的 Callable的Future引用的集合，我们可以告知成功发送的电子邮件数量和失败的电子邮件数量。

这种基础结构不仅易于使用，而且还可以促进关注点的清晰分离，从而在调度程序和实际任务之间提供了预定义的通信机制。

参考： Java并发示例–在Ricardo Zuasti的博客博客中，从我们的JCG合作伙伴 Ricardo Zuasti 获得并发任务的反馈。