Java并发编程实践 目录

并发编程 01—— ThreadLocal

并发编程 02—— ConcurrentHashMap

并发编程 03—— 阻塞队列和生产者-消费者模式

并发编程 04—— 闭锁CountDownLatch 与 栅栏CyclicBarrier

并发编程 05—— Callable和Future

并发编程 06—— CompletionService : Executor 和 BlockingQueue

并发编程 07—— 任务取消

并发编程 08—— 任务取消 之 中断

并发编程 09—— 任务取消 之 停止基于线程的服务

并发编程 10—— 任务取消 之 关闭 ExecutorService

并发编程 11—— 任务取消 之 “毒丸”对象

并发编程 12—— 任务取消与关闭 之 shutdownNow 的局限性

并发编程 13—— 线程池的使用 之 配置ThreadPoolExecutor 和 饱和策略

并发编程 14—— 线程池 之 整体架构

并发编程 15—— 线程池 之 原理一

并发编程 16—— 线程池 之 原理二

并发编程 17—— Lock

并发编程 18—— 使用内置条件队列实现简单的有界缓存

并发编程 19—— 显式的Conditon 对象

并发编程 20—— AbstractQueuedSynchronizer 深入分析

并发编程 21—— 原子变量和非阻塞同步机制

 1 package com.concurrency.TaskExecution_6;
 2
 3 import java.util.concurrent.Callable;
 4 import java.util.concurrent.TimeUnit;
 5
 6 /**
 7  *
 8  * @ClassName: ReturnAfterSleepCallable
 9  * TODO
10  * @author Xingle
11  * @date 2014-9-16 上午9:20:34
12  */
13 public class ReturnAfterSleepCallable implements Callable<Integer>{
14
15     private int sleepSeconds;
16     private int returnValue;
17
18     public ReturnAfterSleepCallable(int sleepSeconds,int returnValue){
19         this.sleepSeconds = sleepSeconds;
20         this.returnValue = returnValue;
21     }
22
23
24     @Override
25     public Integer call() throws Exception {
26         System.out.println("begin to execute ");
27
28         TimeUnit.SECONDS.sleep(sleepSeconds);
29         return returnValue;
30     }
31
32 }

 1 package com.concurrency.TaskExecution_6;
 2
 3 import java.util.ArrayList;
 4 import java.util.List;
 5 import java.util.concurrent.ExecutionException;
 6 import java.util.concurrent.ExecutorService;
 7 import java.util.concurrent.Executors;
 8 import java.util.concurrent.Future;
 9
10 /**
11  * 传统的繁琐做法
12  * @ClassName: TraditionalTest
13  * TODO
14  * @author Xingle
15  * @date 2014-9-16 上午10:06:21
16  */
17 public class TraditionalTest {
18
19     public static void main(String[] args){
20         int taskSize = 5;
21         ExecutorService executor = Executors.newFixedThreadPool(taskSize);
22         List<Future<Integer>> futureList = new ArrayList<Future<Integer>>();
23
24         for(int i= 1; i<=taskSize; i++){
25             int sleep = taskSize -1;
26             int value = i;
27             //向线程池提交任务
28             Future<Integer> future = executor.submit(new ReturnAfterSleepCallable(sleep, value));
29             //保留每个任务的Future
30             futureList.add(future);
31         }
32         // 轮询,获取完成任务的返回结果
33         while(taskSize > 0){
34             for (Future<Integer> future : futureList){
35                 Integer result = null;
36                 try {
37                     result = future.get();
38                 } catch (InterruptedException e) {
39                     e.printStackTrace();
40                 } catch (ExecutionException e) {
41                     e.printStackTrace();
42                 }
43                 //任务已经完成
44                 if(result!=null){
45                     System.out.println("result = "+result);
46                     //从future列表中删除已经完成的任务
47                     futureList.remove(future);
48                     taskSize --;
49                     break;
50                 }
51             }
52         }
53         // 所有任务已经完成,关闭线程池
54         System.out.println("all over ");
55         executor.shutdown();
56     }
57
58 }

 1 package com.concurrency.TaskExecution_6;
 2
 3 import java.util.concurrent.CompletionService;
 4 import java.util.concurrent.ExecutionException;
 5 import java.util.concurrent.ExecutorCompletionService;
 6 import java.util.concurrent.ExecutorService;
 7 import java.util.concurrent.Executors;
 8
 9 /**
10  * 使用CompletionService
11  * @ClassName: CompletionServiceTest
12  * TODO
13  * @author Xingle
14  * @date 2014-9-16 上午11:32:45
15  */
16 public class CompletionServiceTest {
17
18     public static void main(String[] args){
19         int taskSize = 5;
20         ExecutorService executor = Executors.newFixedThreadPool(taskSize);
21         // 构建完成服务
22         CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(executor);
23
24         for (int i=1;i<= taskSize; i++){
25             // 睡眠时间
26             int sleep = taskSize - i;
27             // 返回结果
28             int value = i;
29             //向线程池提交任务
30             completionService.submit(new ReturnAfterSleepCallable(sleep, value));
31             try {
32                 System.out.println("result:"+completionService.take().get());
33             } catch (InterruptedException e) {
34                 e.printStackTrace();
35             } catch (ExecutionException e) {
36                 e.printStackTrace();
37             }
38         }
39
40         System.out.println("all over. ");
41         executor.shutdown();
42
43     }
44
45 }

/*** A service that decouples the production of new asynchronous tasks* from the consumption of the results of completed tasks.  Producers* <tt>submit</tt> tasks for execution. Consumers <tt>take</tt>* completed tasks and process their results in the order they* complete. */

也就是说，CompletionService实现了生产者提交任务和消费者获取结果的解耦，生产者和消费者都不用关心任务的完成顺序，由CompletionService来保证，消费者一定是按照任务完成的先后顺序来获取执行结果。

ExecutorCompletionService是CompletionService的实现，融合了线程池Executor和阻塞队列BlockingQueue的功能。

private class QueueingFuture extends FutureTask<Void> {QueueingFuture(RunnableFuture<V> task) {super(task, null);this.task = task;}protected void done() { completionQueue.add(task); }private final Future<V> task;}