一、前言
java 的 java.util.concurrent 是 java 用于提供一些并发程序所需功能的类包。它的功能全面且强大，在前面，我们已经使用过原子基本变量，BlockingQueue 等类。现在，我们需要更加深入的去了解 JUC 的强大功能。
二、CountDownLatch
该类用来同步一个或多个任务，强制它们等待由其他任务执行的一组操作完成。
在 CountDownLatch 对象中设置一个初始的计数值，任何在这个对象上调用 wait() 的方法都讲阻塞，直至这个计数值到达0。其他任务在结束工作时，可以在该对象上调用 countDown() 来减小这个数值。同事，CountDownLatch 只能出发一次，计数值不能被重置。如果有重置的需要，可以使用 CyclicBarrier。
先来看一个使用 CountDownLatch 的简单示例：
1 package JUCTest; 2 3 import java.util.Random; 4 import java.util.concurrent.CountDownLatch; 5 import java.util.concurrent.ExecutorService; 6 import java.util.concurrent.Executors; 7 import java.util.concurrent.TimeUnit; 8 9 class TaskPortion implements Runnable{ 10 private static int counter = 0; 11 private final int id = counter++; 12 private static Random rand = new Random(47); 13 private final CountDownLatch countDownLatch; 14 public TaskPortion(CountDownLatch countDownLatch) { 15 this.countDownLatch = countDownLatch; 16 } 17 @Override 18 public void run() { 19 try { 20 doWork(); 21 countDownLatch.countDown(); 22 }catch(InterruptedException e) { 23 System.out.println("Exit"); 24 } 25 } 26 27 public void doWork() throws InterruptedException{ 28 TimeUnit.MILLISECONDS.sleep(rand.nextInt(20000)); 29 System.out.println(this + " complete"); 30 } 31 32 public String toString() { 33 return "TaskPorition : " + id; 34 } 35 36 } 37 38 class WaitingTask implements Runnable{ 39 private static int counter = 0; 40 private final int id = counter++; 41 private final CountDownLatch countDownLatch; 42 public WaitingTask(CountDownLatch countDownLatch) { 43 this.countDownLatch = countDownLatch; 44 } 45 46 @Override 47 public void run() { 48 try { 49 countDownLatch.await(); 50 System.out.println("latch barrier pass for " + this); 51 }catch(InterruptedException e) { 52 System.out.println(this + "interrupted"); 53 } 54 } 55 56 public String toString() { 57 return "TaskPorition : " + id; 58 } 59 } 60 public class CountDownLatchDemo { 61 static final int SIZE = 10; 62 public static void main(String args[]) { 63 ExecutorService exec = Executors.newCachedThreadPool(); 64 CountDownLatch latch = new CountDownLatch(SIZE); 65 for(int i=0;i<10;i++) { 66 exec.execute(new WaitingTask(latch)); 67 } 68 for(int i=0;i<SIZE;i++) { 69 exec.execute(new TaskPortion(latch)); 70 } 71 System.out.println("Launched all tasls"); 72 exec.shutdown(); 73 } 74 } 75 //output 76 /*Launched all tasls 77 TaskPorition : 5 complete 78 TaskPorition : 1 complete 79 TaskPorition : 4 complete 80 TaskPorition : 3 complete 81 TaskPorition : 9 complete 82 TaskPorition : 0 complete 83 TaskPorition : 7 complete 84 TaskPorition : 8 complete 85 TaskPorition : 6 complete 86 TaskPorition : 2 complete 87 latch barrier pass for TaskPorition : 1 88 latch barrier pass for TaskPorition : 2 89 latch barrier pass for TaskPorition : 0 90 latch barrier pass for TaskPorition : 6 91 latch barrier pass for TaskPorition : 7 92 latch barrier pass for TaskPorition : 3 93 latch barrier pass for TaskPorition : 4 94 latch barrier pass for TaskPorition : 5 95 latch barrier pass for TaskPorition : 8 96 latch barrier pass for TaskPorition : 9 97 */
通过前面章节的内容，我们可以很容一个实现 “A 任务 等到 B 任务完成之后再去执行” 的功能，而在上述例子中，B 任务是由 10 个子任务构成的。通过 CountDownLatch 我们没完成一个子任务，就会是 countDownLatch 减1。等待所有子任务完成，countDownLatch 变为0后，启动 A 任务。
二、CyclicBarrier
countDownLatch 可以使某个任务完成之后进入阻塞状态，阻塞状态持续到其他相关任务全部完成之后（countDownLatch 变为0）。CyclicBarrier 类似于countDownLatch ，和 countDownLatch 的区别在于。在所有任务完成之后，CyclicBarrier 的计数器会重置。
先看一个简单的示例：
1 package JUCTest; 2 3 import java.util.ArrayList; 4 import java.util.List; 5 import java.util.Random; 6 import java.util.concurrent.BrokenBarrierException; 7 import java.util.concurrent.CyclicBarrier; 8 import java.util.concurrent.ExecutorService; 9 import java.util.concurrent.Executors; 10 import java.util.concurrent.TimeUnit; 11 12 class Horse implements Runnable{ 13 private static int counter = 0; 14 private final int id = counter++; 15 private int strides = 0; 16 private static Random rand = new Random(47); 17 private static CyclicBarrier barrier; 18 public Horse(CyclicBarrier b) { 19 barrier = b; 20 } 21 public synchronized int getStrides() { 22 return strides; 23 } 24 @Override 25 public void run() { 26 try { 27 while(!Thread.interrupted()) { 28 synchronized(this) { 29 strides += rand.nextInt(3); 30 } 31 barrier.await(); 32 } 33 }catch(InterruptedException e) { 34 e.printStackTrace(); 35 }catch(BrokenBarrierException e) { 36 throw new RuntimeException(e); 37 } 38 } 39 40 public String toString() { 41 return "Horse " + id +" "; 42 } 43 44 public String tracks(){ 45 StringBuilder s = new StringBuilder(); 46 for(int i=0;i<getStrides();i++) { 47 s.append("*"); 48 } 49 s.append(id); 50 return s.toString(); 51 } 52 } 53 public class HorseRace { 54 55 static final int FINISH_LINE = 75; 56 private List<Horse> horses = new ArrayList<Horse>(); 57 private ExecutorService exec = Executors.newCachedThreadPool(); 58 private CyclicBarrier barrier; 59 public HorseRace(int nHorses,final int pause) { 60 barrier = new CyclicBarrier(nHorses,new Runnable() { 61 public void run() { 62 StringBuilder s = new StringBuilder(); 63 for(int i=0;i<FINISH_LINE;i++) { 64 s.append("="); 65 } 66 System.out.println(s); 67 for(Horse horse : horses) 68 System.out.println(horse.tracks()); 69 for(Horse horse : horses) { 70 if(horse.getStrides() >= FINISH_LINE) { 71 System.out.println(horse + "Won!"); 72 exec.shutdownNow(); 73 return; 74 } 75 } 76 try { 77 TimeUnit.MILLISECONDS.sleep(pause); 78 }catch(InterruptedException e) { 79 System.out.println("barrier-action sleep interrupted"); 80 } 81 } 82 }); 83 for(int i=0;i<nHorses;i++) { 84 Horse horse = new Horse(barrier); 85 horses.add(horse); 86 exec.execute(horse); 87 } 88 } 89 public static void main(String[] args) { 90 int nHorses = 7; 91 int pause = 200; 92 new HorseRace(nHorses,pause); 93 } 94 95 }
上述程序是一个模拟赛马的操作，一共有75个栅栏，每个马的速度都不一样的，所以每次打印每只马跨越了多少栅栏时，会出现你追我赶的情况。但是程序内在逻辑是怎么样呢？
我们可以把马对应成一个任务，马跨域栅栏是一次 run() 方法内部走完了一次循环。
CyclicBarrier 就相当于一堵墙，它横在所有马的前方，当马完成一次操作（随机跨越1~3个栅栏），它来到了墙面前，被墙挡住(代码是通过 await() 实现的)。等所有的马（具体几只是在 CyclicBarrier 的构造函数里确定的）都来到墙面前的时候，墙打开，所有马进行下一次操作。
可以推测出来，CyclicBarrier 内部一定有一个计数器（通过查看源码可以知道 在构造函数里是把值赋给 final int parties 和 int count 的，前者是 final 无法改变用于重置计数器使用，后者用于计数），我们没调用一次 await() 方法，这个计数器就会减1。直到我们调用了 parties 次 await() 计数器变为 0 。然后所有任务可以进行一下步，同时，计数器变为 parties ，继续阻塞任务进入再下一步的操作，直到它再次为0；
ps: 通过源码可以肯定我们的推测，事实上 每次我们调用 await()， count 就会递减，而当 count 为 0 时，就会调用 nextGeneration 方法。nextGeneration 会把计数器重置，同时会唤醒阻塞的任务。顺便一提的事，CyclicBarrier 实现阻塞和唤醒的方式是使用 Condition (前面有具体内容)。
在 CyclicBarrier 的构造函数里还有一个 Runnable，它会在计数器为 0 的时候启动。
三、DelayQueue
在 JUC 中，除了之前提到的，LinkedBlockingQueue、ArrayBlockingQueue 和 SynchronousQueue 之外，还有其他几种 Queue, DelayQueue 就是其中之一。
DelayQueue 是一个无界的 BlockingQueue，用于放置实现了 Delayed 接口的对象，其中对象只能在其到期才能从队列中取走。并且该队列是有序的，我们需要实现 compareTo 方法用来作为排序的标准。当从 DelayQueue 获取对象时，只会获取延迟到期的对象。
package JUCTest; import java.util.ArrayList; import java.util.List; import java.util.Random; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; class DelayedTask implements Runnable,Delayed{ private static int counter = 0; private final int id = counter++; private final int delta; private final long trigger; protected static List<DelayedTask> squence = new ArrayList<DelayedTask>(); public DelayedTask(int delayInMilliseconds) { delta = delayInMilliseconds; trigger = System.nanoTime() + TimeUnit.NANOSECONDS.convert(delta, TimeUnit.MILLISECONDS); squence.add(this); } @Override public void run() { System.out.println(this); } @Override public int compareTo(Delayed o) { DelayedTask that = (DelayedTask) o; if(trigger < that.trigger) return 1; if(trigger > that.trigger) return 1; return 0; } @Override public long getDelay(TimeUnit unit) { return unit.convert(trigger - System.nanoTime(),TimeUnit.NANOSECONDS); } public String toString() { return String.format("[%1$-4d]", delta) + " Task " + id; } public String summary() { return "("+id+":"+delta+")"; } public static class EndSentinel extends DelayedTask{ private ExecutorService exec; public EndSentinel(int delay,ExecutorService e) { super(delay); exec=e; } public void run() { for(DelayedTask pt : squence) { System.out.print(pt.summary() + " "); } System.out.println(" "); System.out.println(this + " Calling shutdownNow()"); exec.shutdownNow(); } } } class DelayedTaskConsumer implements Runnable{ private DelayQueue<DelayedTask> q; public DelayedTaskConsumer(DelayQueue<DelayedTask> q) { this.q = q; } @Override public void run() { try { while(!Thread.interrupted()) { q.take().run(); } }catch(InterruptedException e) { e.printStackTrace(); } System.out.println("Finished DelayedTaskConsumer"); } } public class DelayQueueDemo { public static void main(String[] args) { Random rand = new Random(47); ExecutorService exec = Executors.newCachedThreadPool(); DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>(); for(int i=0;i<20;i++) { queue.put(new DelayedTask(rand.nextInt(5000))); } queue.add(new DelayedTask.EndSentinel(5000, exec)); exec.execute(new DelayedTaskConsumer(queue)); } }
在上面这个例子中，我们让 DelayedTask 实现了 Runnable 和 Delayed 接口。除了 run() 方法之外，我们同时实现了 compareTo() 和 getDelay() 方法。然后输出的结果表明，任务从队列总出来的顺序是按照 getDelay() 所获得的值来确定的。
我们使用变量 delta 来作为延迟时间的，System.nanoTime() 会获得一个纳秒为单位的数字，这个数字单独使用没有任何意义，但是，在程序的两个位置都使用 System.nanoTime() 并且把这两个值相减，就能得到一个精准的时间差。在构造函数里 trigger 被赋值为 System.nanoTime() + delta，而在 getDelay() 中返回的值是 trigger – System.nanoTime()（第二次使用，后面用 System.nanoTime()2 做区别），那么返回的值其实是，System.nanoTime() + delta – System.nanoTime()2，System.nanoTime()2 – System.nanoTime() 可以认为使我们给 trigger 赋值和程序调用 getDelay() 之间的时间差，当时间差，也就是经过的时间 > delta (设定的延迟时间) 时，对象才能出列。换句话说 getDelay() 返回的值 < 0 才能出列。
但是对象出列除了延迟时间到达之外这个条件之外，还得满足它在对列的首位，所以我们必须使用 compareTo() 来规定一个排列的顺序，使得延迟时间到达最短的放在队首位置。所以我们用 trigger 来尽行比较。注意，这里的排队应该是最块走完延迟时间的排前面，而不是延迟时间最短的排前面。比如，A的延迟时间为 1s 他是在第 10s 中的时候放进去的，B的延迟时间为 2 s 它是在第 4s 的时候放进去的，那么B应该排在A前面。
那么，如果我们使用错误的方式来排队，比如把延迟时间到达最晚的放在前面。就会导致效率低下，程序会等到最长的延迟时间到达才会有出列操作。
四、PriorityBlockingQueue
顾名思义，他是以优先级作为排序顺序来给队列中的对象排序的。而排序的方法，依旧是通过 compareTo 方法实现，其实，DelayQueue 可以看做是一种特殊的优先级排序，除了排序之外，他还有延迟的附加条件。所以对于 PriorityBlockingQueue 我们不做过多的说明。
五、SheduledExecutor
SheduledExecutor 可以使任务按照设定的计划去执行，通常，我们需要在指定的时间执行某项任务，或者在一定的周期内循环的执行某项目，就会使用到 SheduledExecutor。
1 package JUCTest; 2 3 import java.util.ArrayList; 4 import java.util.Calendar; 5 import java.util.Collections; 6 import java.util.List; 7 import java.util.Random; 8 import java.util.concurrent.ScheduledThreadPoolExecutor; 9 import java.util.concurrent.TimeUnit; 10 11 public class GreenhouseScheduler { 12 13 private volatile boolean light = false; 14 private volatile boolean water = false; 15 private String thermostat = "Day"; 16 17 public synchronized String getThermostat() { 18 return thermostat; 19 } 20 21 public synchronized void setThermostat(String thermostat) { 22 this.thermostat = thermostat; 23 } 24 25 ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(10); 26 27 public void schedule(Runnable event,long delay) { 28 scheduler.schedule(event, delay, TimeUnit.MILLISECONDS); 29 } 30 31 public void repeat(Runnable event,long initialDelay,long period) { 32 scheduler.scheduleAtFixedRate(event, initialDelay, period, TimeUnit.MILLISECONDS); 33 } 34 35 class LightOn implements Runnable{ 36 public void run() { 37 System.out.println("Turn on lights"); 38 light = true; 39 } 40 } 41 42 class LightOff implements Runnable{ 43 public void run() { 44 System.out.println("Turn off lights"); 45 light = false; 46 } 47 } 48 49 class WaterOn implements Runnable{ 50 public void run() { 51 System.out.println("Turning greenhoulse water on"); 52 water = true; 53 } 54 } 55 56 class WaterOff implements Runnable{ 57 public void run() { 58 System.out.println("Turning greenhoulse water off"); 59 water = false; 60 } 61 } 62 63 class ThermostatNight implements Runnable{ 64 public void run() { 65 System.out.println("Thermostat to night setting"); 66 setThermostat("Night"); 67 } 68 } 69 70 class ThermostatDay implements Runnable{ 71 public void run() { 72 System.out.println("Thermostat to day setting"); 73 } 74 } 75 76 class Bell implements Runnable{ 77 78 public void run() { 79 System.out.println("Bing!"); 80 } 81 82 } 83 84 class Terminate implements Runnable{ 85 public void run() { 86 System.out.println("Terminating!"); 87 scheduler.shutdown(); 88 new Thread() { 89 public void run() { 90 for(DataPoint p : data) { 91 System.out.println(p); 92 } 93 } 94 }; 95 } 96 97 } 98 99 static class DataPoint{ 100 final Calendar time; 101 final float temperature; 102 final float humidity; 103 public DataPoint(Calendar d,float temp,float hum) { 104 time = d; 105 temperature = temp; 106 humidity = hum; 107 } 108 public String toString() { 109 return time.getTime() + String.format(" temperature:, %1s$.1f humidity: %2$.2f",temperature); 110 } 111 } 112 113 private Calendar lastTime = Calendar.getInstance(); 114 { 115 lastTime.set(Calendar.MINUTE, 30); 116 lastTime.set(Calendar.SECOND, 00); 117 } 118 119 private float lastTemp = 65.0f; 120 private int tempDirection = 1; 121 private float lastHumidity = 50.0f; 122 private int humidityDirection = 1; 123 private Random rand = new Random(47); 124 List<DataPoint> data = Collections.synchronizedList(new ArrayList<DataPoint>()); 125 126 class CollectData implements Runnable{ 127 public void run() { 128 System.out.println("Collecting date"); 129 synchronized(GreenhouseScheduler.this) { 130 lastTime.set(Calendar.MINUTE,lastTime.get(Calendar.MINUTE) + 30); 131 } 132 if(rand.nextInt(5) == 4) { 133 tempDirection = -tempDirection; 134 } 135 lastTemp = lastTemp + tempDirection*(1.0f + rand.nextFloat()); 136 if(rand.nextInt(5) == 4) { 137 humidityDirection = -humidityDirection; 138 } 139 lastHumidity = lastHumidity + humidityDirection * rand.nextFloat(); 140 data.add(new DataPoint((Calendar)lastTime.clone(),lastTemp,lastHumidity)); 141 } 142 } 143 public static void main(String[] args) { 144 GreenhouseScheduler gh = new GreenhouseScheduler(); 145 gh.schedule(gh.new ThermostatNight(),5000); 146 gh.repeat(gh.new Bell(), 0, 1000); 147 gh.repeat(gh.newThermostatNight(), 0, 2000);
148 gh.repeat(gh.new LightOn(), 0, 200);
149 gh.repeat(gh.new LightOff(), 0, 400);
150 gh.repeat(gh.new WaterOn(), 0, 600);
151 gh.repeat(gh.new WaterOff(), 0, 800);
152 gh.repeat(gh.new ThermostatDay(), 0, 1400);
153 gh.repeat(gh.new CollectData(), 500, 500);
154
155 }
156
157 }
这里我们引入了一个新的线程池—— ScheduledThreadPoolExecutor，他添加和执行任务的方法不在是 Executor，而是 schedule 和 schedule。schedule 除了需要提供一个 Runnable 作为参数以外，还要提供一个延迟时间，和时间单位。延迟时间和时间单位共同决定了任务在什么时候被启动。scheduleAtFixedRate 还需额外提供一个周期时间，在到达延迟时间之后，每过一个周期，任务就会执行一次。
在上面的示例中，我们创建了一个温室，温室需要进行开关灯、防水、收集数据等操作。
我们一共设置了1个单一任务和8个循环任务。在程序进行到 5s 中时，所有任务被中断。
六、Semaphore
无论是使用 synchronized 亦或是 lock 的方式，都能保证某项资源只能被一个任务获取和使用。但有时候，我们或许会希望能够允许指定数量的任务来获取同一个资源。JUC 为我们提供了 Semaphore 来实现这方面的需求。
在 Thinking in Java 的关于 Semaphore 的示例中，首先创建了一个使用 Semaphore 来进行控制的对象池，然后通过这个对象池来实现“允许指定数量的任务来获取同一个资源”这一功能，我们先看代码。
在看示例钱，我们需要简单理解下 Semaphore 的运作方式，Seamaphore 的构造方法里，包含两个参数：permits(int)，fair(bool)。permits 就是所谓的计数器的值，即我们希望资源能同时被多少任务访问。而 fair 是一个布尔值，它决定我们使用的是公平锁还是非公平锁，关于公平锁，在之后的拓展章节再详细叙述。
创建完 Semaphore 之后，我们通过它的 aquire() 来获取进入资源的权限，此时计数器 -1，通过它的 release() 方法，来释放一个权限，此时计数器 +1。
以下是 Thinking in Java 示例中所用的对象池：
1 package JUCTest; 2 3 import java.util.ArrayList; 4 import java.util.List; 5 import java.util.concurrent.Semaphore; 6 7 public class Pool<T> { 8 private int size; 9 private List<T> items = new ArrayList<T>(); 10 private volatile boolean[] checkedOut; 11 private Semaphore available; 12 public Pool(Class<T> classObject,int size) { 13 this.size = size; 14 checkedOut = new boolean[size]; 15 available = new Semaphore(10,true); 16 for(int i=0;i<size;i++) { 17 try { 18 items.add(classObject.newInstance()); 19 }catch(Exception e) { 20 throw new RuntimeException(e); 21 } 22 } 23 } 24 25 public T checkOut() throws InterruptedException{ 26 available.acquire(); 27 return getItem(); 28 } 29 30 public void checkIn(T x) { 31 if(releaseItem(x)) { 32 available.release(); 33 } 34 } 35 private synchronized T getItem() { 36 for(int i=0;i<size;i++) { 37 if(!checkedOut[i]) { 38 checkedOut[i] = true; 39 return items.get(i); 40 } 41 } 42 return null; 43 } 44 45 private synchronized boolean releaseItem(T item) { 46 int index = items.indexOf(item); 47 if(index == -1) { 48 checkedOut[index] = false; 49 return true; 50 } 51 return false; 52 } 53 }
在 pool 的构造函数中，我们创建一个可以放置对象（泛型 ）的 List，初始化 Semaphore。同时使用了 newInstance() 的方式创建了 size 个对象。
在更详细的说明之前，先来看看这个对象池的应用。首先，我们需要新建一个类：
1 package JUCTest; 2 3 public class Fat { 4 private volatile double d; 5 private static int counter = 0; 6 private int id = counter++; 7 public Fat() { 8 for(int i=1;i<10000;i++) { 9 d += (Math.PI + Math.E); 10 } 11 } 12 13 public void operation() { 14 System.out.println("this"); 15 } 16 17 public String toString() { 18 return "Fat id: " + id; 19 } 20 }
然后，我么通过 Pool 来对该对象进行管理。
1 package JUCTest; 2 3 import java.util.ArrayList; 4 import java.util.List; 5 import java.util.concurrent.ExecutorService; 6 import java.util.concurrent.Executors; 7 import java.util.concurrent.Future; 8 import java.util.concurrent.TimeUnit; 9 10 class CheckoutTask<T> implements Runnable { 11 12 private static int counter = 0; 13 private final int id = counter++; 14 private Pool<T> pool; 15 public CheckoutTask(Pool<T> pool) { 16 this.pool = pool; 17 } 18 19 public void run() { 20 try { 21 T item = pool.checkOut(); 22 System.out.println(this + " checked out " + item); 23 TimeUnit.SECONDS.sleep(1); 24 System.out.println(this + "cheked in " + item); 25 pool.checkIn(item); 26 }catch(InterruptedException e) { 27 System.out.println("InterruptedException"); 28 } 29 } 30 31 public String toString() { 32 return "CheckoutTesk " + id + " "; 33 } 34 35 36 37 38 } 39 40 public class SemaphoreDemo{ 41 final static int SIZE = 25; 42 public static void main(String[] args) throws InterruptedException { 43 final Pool<Fat> pool = new Pool<Fat>(Fat.class,SIZE); 44 ExecutorService exec = Executors.newCachedThreadPool(); 45 List<Fat> lists = new ArrayList<Fat>(); 46 for(int i=0;i<SIZE;i++) { 47 Fat f = pool.checkOut(); 48 System.out.println(i + ": main() thread check out"); 49 f.operation(); 50 lists.add(f); 51 } 52 Future<?> blocked = exec.submit(new Runnable() { 53 public void run() { 54 try { 55 pool.checkOut(); 56 }catch(InterruptedException e) { 57 System.out.println("Check out Interrupted"); 58 } 59 } 60 }); 61 62 TimeUnit.SECONDS.sleep(2); 63 blocked.cancel(true); 64 System.out.println("Check in object in " + lists); 65 for(Fat f : lists) { 66 pool.checkIn(f); 67 } 68 for(Fat f : lists) { 69 pool.checkIn(f); 70 } 71 exec.shutdown(); 72 } 73 74 75 76 }
在 SemaphoreDemo 中，创建了一个容量为 SIZE 的 pool，在 pool 的构造方法中，我们根据传入的模板参数，创建了 SIZE 个 Fat 对象，然后所有的 Fat 的对象全部通过 checkout 从 pool 里取出。
在最后往 pool 中 checkin Fat 时，我们发现不论我们往里添加了多个对象，在 pool 中始终最多只有 SIZE 个对象。那么后来的添加的对象哪里去了？checkin 的操作并没有消失，也没有出错，只是被阻塞了，如果我们此时通过 checkout 释放出一些位置，那些消失的 Fat 就会顺利的插入到 pool 里。
那么这是如何实现的？
在 Checkout() 中，我们再获取到 Fat 对象前，需要进行一次 acquire() 每次的 acquire 操作，都会使得 Semaphore 中的计数器 -1，当技术器为 0 时，我们继续进行 checkout()（或者继续进行 checkout() 里的 acquire() 操作），就会被阻塞。直到我们使用 checkin() （或者说是 checkin() 里的 release()），使得计数器 +1。被阻塞的 checkout 操作才会继续执行。
在上面的代码中，我们先进行了 SIZE 次 checkout() 操作，然后，再新建一个任务继续使用 checkout 操作，其被阻塞，直到我们将其中断。后台输出 Check out Interrupted，如果我们在 blocked.cancel(true) ——中断操作之前，执行 checkin 操作，就会使阻塞的任务能够继续进行下去。
七、Exchanger
Exchanger 是在两个任务之间交换对象的栅栏。当 A 和 B 任务进入栅栏时，它们各自拥有一个对象 C 和 D，当他们离开时，拥有的对象互换，即 A 拥有 D，B 有用 C。在创建 A 和 B 时，需要把他们和同一个 Exchanger 绑定。
1 package JUCTest; 2 3 import java.util.concurrent.Exchanger; 4 import java.util.concurrent.ExecutorService; 5 import java.util.concurrent.Executors; 6 import java.util.concurrent.TimeUnit; 7 8 class ExchangerProducer implements Runnable{ 9 10 private Exchanger<Integer> exchanger; 11 public ExchangerProducer(Exchanger<Integer> exchanger) { 12 this.exchanger = exchanger; 13 } 14 @Override 15 public void run() { 16 for(int i=1;i<10;i++) { 17 Integer data = i; 18 System.out.println(i + " : producer before exchange : " + data); 19 try { 20 data = exchanger.exchange(data); 21 } catch (InterruptedException e) { 22 System.out.println("Interrupted..."); 23 } 24 System.out.println(i + ": producer after exchange : " + data); 25 } 26 } 27 28 } 29 30 class ExchagerConsumer implements Runnable{ 31 32 private Exchanger<Integer> exchanger; 33 public ExchagerConsumer(Exchanger<Integer> exchanger) { 34 this.exchanger = exchanger; 35 } 36 @Override 37 public void run() { 38 for(int i=1;i<10;i++) { 39 Integer data = i * 2; 40 System.out.println(i + " : consumer before exchange : " + data); 41 try { 42 data = exchanger.exchange(data); 43 } catch (InterruptedException e) { 44 System.out.println("Interrupted..."); 45 } 46 System.out.println(i + " : consumer after exchange : " + data); 47 } 48 } 49 } 50 public class ExchagerDemo { 51 52 public static void main(String[] args) throws InterruptedException { 53 Exchanger<Integer> exchanger = new Exchanger<Integer>(); 54 ExchangerProducer exchangerProducer = new ExchangerProducer(exchanger); 55 ExchagerConsumer exchagerConsumer = new ExchagerConsumer(exchanger); 56 ExecutorService exec = Executors.newCachedThreadPool(); 57 exec.execute(exchangerProducer); 58 exec.execute(exchagerConsumer); 59 TimeUnit.SECONDS.sleep(3); 60 exec.shutdownNow(); 61 62 } 63 64 }
在上面的示例中，producer 负责生产奇数，consumer 负责生产偶数，在 producer 或者 consumer 生产完一个数之后，会将其放入 Exchanger 中等待交换，双方进入到阻塞状态，等待交换完成之后，任务继续进行。
