Java并行编程–从并行任务集获取反馈

柔光的暖阳◎ 2023-10-16 23:18 17阅读 0赞

在并行任务启动后，强制性地从并行任务得到反馈。  
  
假想有一个程序，可以发送批邮件，还使用了多线程机制。你想知道有多少邮件成功发送吗？你想知道在实际发送过程期间，这个批处理工作的实时进展吗？  
  
要实现多线程的这种反馈，我们可以使用Callable接口。此接口的工作方式基本上与Runnable相同，但是执行方法（call()）会返回一个值，该值反映了执行计算的结果。

**\[java\]** [view plain][] [copy][view plain]

[print][view plain] [?][view plain]

1.  **package** com.ricardozuasti; 
2.  
3.  **import** java.util.concurrent.Callable; 
4.  
5.  **public****class** FictionalEmailSender **implements** Callable<Boolean>\{ 
6.  **private** String to; 
7.  **private** String subject; 
8.  **private** String body; 
9.  **public** FictionalEmailSender(String to, String subject, String body)\{ 
10. **this**.to = to; 
11. **this**.subject = subject; 
12. **this**.body = body; 
13.  \} 
14. 
15. @Override
16. **public** Boolean call() **throws** InterruptedException \{ 
17. // 在0~0.5秒间模拟发送邮件
18.  Thread.sleep(Math.round(Math.random()\*0.5\*1000)); 
19. // 假设我们有80%的几率成功发送邮件
20. **if**(Math.random()>0.2)\{ 
21. **return****true**; 
22.  \}**else**\{ 
23. **return****false**; 
24.  \} 
25.  \} 
26. 
27. \}

注意：Callable接口可用于返回任意数据类型，因此我们的任务可以返回我们需要的任何信息。

现在，我们使用一个线程池ExecutorService来发送邮件，由于我们的任务是以Callable接口实现的，我们提交执行的每个新任务，都会得到一个Future引用。注意我们要使用直接的构造器创建ExecutorService，而不是使用来自Executors的工具方法创建。这是因为使用指定类ThreadPoolExecutor提供了一些方法可以派上用场。

**\[java\]** [view plain][] [copy][view plain]

[print][view plain] [?][view plain]

1.  **package** com.ricardozuasti; 
2.  
3.  **import** java.util.concurrent.Future; 
4.  **import** java.util.concurrent.LinkedBlockingQueue; 
5.  **import** java.util.concurrent.ThreadPoolExecutor; 
6.  **import** java.util.concurrent.TimeUnit; 
7.  **import** java.util.ArrayList; 
8.  **import** java.util.List; 
9.  
10. **public****class** Concurrency2 \{ 
11. **public****static****void** main(String\[\] args)\{ 
12. **try**\{ 
13.  ThreadPoolExecutor executor = **new** ThreadPoolExecutor(30, 30, 1, 
14.  TimeUnit.SECONDS, **new** LinkedBlockingQueue()); 
15.  List<Future<Boolean>> futures = **new** ArrayList<Future<Boolean>>(9000); 
16. // 发送垃圾邮件, 用户名假设为4位数字
17. **for**(**int** i=1000; i<10000; i++)\{ 
18.  futures.add(executor.submit(**new** FictionalEmailSender(i+"@sina.com", 
19. "Knock, knock, Neo", "The Matrix has you..."))); 
20.  \} 
21. // 提交所有的任务后,关闭executor
22.  System.out.println("Starting shutdown..."); 
23.  executor.shutdown(); 
24. 
25. // 每秒钟打印执行进度
26. **while**(!executor.isTerminated())\{ 
27.  executor.awaitTermination(1, TimeUnit.SECONDS); 
28. **int** progress = Math.round((executor.getCompletedTaskCount() 
29. \*100)/executor.getTaskCount()); 
30.  System.out.println(progress + "% done (" \+ 
31. executor.getCompletedTaskCount() + " emails have been sent)."); 
32.  \} 
33. // 现在所有邮件已发送完, 检查futures, 看成功发送的邮件有多少
34. **int** errorCount = 0; 
35. **int** successCount = 0; 
36. **for**(Future<Boolean> future : futures)\{ 
37. **if**(future.get())\{ 
38.  successCount++; 
39.  \}**else**\{ 
40.  errorCount++; 
41.  \} 
42.  \} 
43.  System.out.println(successCount + " emails were successfully sent, but " \+ 
44.  errorCount + " failed."); 
45.  \}**catch**(Exception ex)\{ 
46.  ex.printStackTrace(); 
47.  \} 
48.  \} 
49. \}

执行这个类，输出结果如下：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain] [?][view plain]

1.  Starting shutdown... 
2.  1% done (118 emails have been sent). 
3.  2% done (232 emails have been sent). 
4.  3% done (358 emails have been sent). 
5.  5% done (478 emails have been sent). 
6.  6% done (587 emails have been sent). 
7.  7% done (718 emails have been sent). 
8.  9% done (850 emails have been sent). 
9.  10% done (969 emails have been sent). 
10. ……

所有的任务都由ExecutorService提交，我们开始它的关闭（防止提交新任务）并使用一个循环（实时场景，可能你会继续做其它的事情）来等待，直至所有任务都被执行完成、计算和打印当前每次迭代的进度。

注意，你可以存储executor引用，也可以在任意时间从其它线程查询它的计算结果和报告进程进度。  
  
最后，使用Future集合引用，我们得到ExecutorService提交的每个Callable接口，通知成功发送的邮件数量和发送失败的邮件数量。  
  
此结构不但易于使用，还使得相关性得到清晰的隔离，在调度程序和实际任务之间提供了一个预定义的通信机制。

转自：http://blog.csdn.net/chszs/article/details/7418880

[view plain]: http://blog.csdn.net/chszs/article/details/7418880#