（11）java5的Semaphere、CyclicBarrier同步工具

java.util.concurrent

类 Semaphore

java.lang.Object

java.util.concurrent.Semaphore

所有已实现的接口：

Serializable

public class Semaphoreextends Objectimplements Serializable一个计数信号量。

从概念上讲，信号量维护了一个许可集。

如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。

每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。

但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。

Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目。

例如，下面的类使用信号量控制对内容池的访问： 

 class Pool {
	 private static final int MAX_***AILABLE = 100;
	private final Semaphore available = new Semaphore(MAX_***AILABLE, true);

	public Object getItem() throws InterruptedException {
		  available.acquire();
		  return getNextAvailableItem();
  }

   public void putItem(Object x) {
	 if (markAsUnused(x))
		available.release();
   }

   // Not a particularly efficient data structure; just for demo

   protected Object[] items = ... whatever kinds of items being managed
   protected boolean[] used = new boolean[MAX_***AILABLE];

   protected synchronized Object getNextAvailableItem() {
     for (int i = 0; i < MAX_***AILABLE; ++i) {
       if (!used[i]) {
          used[i] = true;
          return items[i];
       }
     }
     return null; // not reached
   }

   protected synchronized boolean markAsUnused(Object item) {
     for (int i = 0; i < MAX_***AILABLE; ++i) {
       if (item == items[i]) {
          if (used[i]) {
            used[i] = false;
            return true;
          } else
            return false;
       }
     }
     return false;
   }

 }

获得一项前，每个线程必须从信号量获取许可，从而保证可以使用该项。



该线程结束后，将项返回到池中并将许可返回到该信号量，从而允许其他线程获取该项。



注意，调用 acquire() 时无法保持同步锁，因为这会阻止将项返回到池中。

信号量封装所需的同步，以限制对池的访问，这同维持该池本身一致性所需的同步是分开的。

将信号量初始化为 1，使得它在使用时最多只有一个可用的许可，从而可用作一个相互排斥的锁。

这通常也称为二进制信号量，因为它只能有两种状态：一个可用的许可，或零个可用的许可。

按此方式使用时，二进制信号量具有某种属性（与很多 Lock 实现不同），即可以由线程释放“锁”，而不是由所有者（因为信号量没有所有权的概念）。

在某些专门的上下文（如死锁恢复）中这会很有用。

此类的构造方法可选地接受一个公平 参数。当设置为 false 时，此类不对线程获取许可的顺序做任何保证。

特别地，闯入 是允许的，也就是说可以在已经等待的线程前为调用 acquire() 的线程分配一个许可，

从逻辑上说，就是新线程将自己置于等待线程队列的头部。

当公平设置为 true 时，信号量保证对于任何调用获取方法的线程而言，都按照处理它们调用这些方法的顺序（即先进先出；FIFO）来选择线程、获得许可。

注意，FIFO 排序必然应用到这些方法内的指定内部执行点。

所以，可能某个线程先于另一个线程调用了 acquire，但是却在该线程之后到达排序点，并且从方法返回时也类似。

还要注意，非同步的 tryAcquire 方法不使用公平设置，而是使用任意可用的许可。

通常，应该将用于控制资源访问的信号量初始化为公平的，以确保所有线程都可访问资源。

为其他的种类的同步控制使用信号量时，非公平排序的吞吐量优势通常要比公平考虑更为重要。

此类还提供便捷的方法来同时 acquire 和释放多个许可。

小心，在未将公平设置为 true 时使用这些方法会增加不确定延期的风险。

内存一致性效果：线程中调用“释放”方法（比如 release()）之前的操作 happen-before 另一线程中紧跟在成功的“获取”方法（比如 acquire()）之后的操作。

acquire

public void acquire()

throws InterruptedException从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。

获取一个许可（如果提供了一个）并立即返回，将可用的许可数减 1。

如果没有可用的许可，则在发生以下两种情况之一前，禁止将当前线程用于线程安排目的并使其处于休眠状态：

某些其他线程调用此信号量的 release() 方法，并且当前线程是下一个要被分配许可的线程；或者

其他某些线程中断当前线程。

如果当前线程：

被此方法将其已中断状态设置为 on ；或者

在等待许可时被中断。

则抛出 InterruptedException，并且清除当前线程的已中断状态。

抛出：

InterruptedException - 如果当前线程被中断

release

public void release()释放一个许可，将其返回给信号量。

释放一个许可，将可用的许可数增加 1。

如果任意线程试图获取许可，则选中一个线程并将刚刚释放的许可给予它。然后针对线程安排目的启用（或再启用）该线程。

不要求释放许可的线程必须通过调用 acquire() 来获取许可。通过应用程序中的编程约定来建立信号量的正确用法。

应用实例：

package com.itm.thread;

/*
 * 
 
 Semaphore的作用类似Lock的功能，不同的是Semaphore的构造函数中可以传入一个int型的参数，用来确定创建一个多大的通道。
 Lock一次只允许一个线程进入，解锁后才允许别的线程进入。而Semaphore可以允许多个线程同时进入，一旦有线程释放就会空出一个位置让另外的线程进入。
 相当与Lock是一个单车道，而Semaphore是一个多车道。
 
 */
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class SemaphoreTest {
	public static void main(String[] args) {
		ExecutorService es = Executors.newCachedThreadPool();
		
		// final Semaphore sema=new Semaphore(3); //共有3个通道
		final Semaphore sema = new Semaphore(3, true); // 表示按排队的先后顺序进入
		for (int i = 0; i < 10; i++) {
			es.execute(new Runnable() {
				@Override
				public void run() {
					try {
						sema.acquire();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("线程" + Thread.currentThread().getName()
							+ "已进入，当前有" + (3 - sema.availablePermits())
							+ "个线程并发");
					try {
						Thread.sleep(new Random().nextInt(5000));
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("线程" + Thread.currentThread().getName()
							+ "即将离开");
					sema.release(); // 一个线程释放就会允许另外的线程进入
				}
			});
		}
		es.shutdown();
	}
}

运行结果：

线程pool-1-thread-2已进入，当前有2个线程并发

线程pool-1-thread-1已进入，当前有2个线程并发

线程pool-1-thread-3已进入，当前有3个线程并发

线程pool-1-thread-1即将离开

线程pool-1-thread-4已进入，当前有3个线程并发

线程pool-1-thread-4即将离开

线程pool-1-thread-5已进入，当前有3个线程并发

线程pool-1-thread-2即将离开

线程pool-1-thread-9已进入，当前有3个线程并发

线程pool-1-thread-5即将离开

线程pool-1-thread-7已进入，当前有3个线程并发

线程pool-1-thread-3即将离开

线程pool-1-thread-6已进入，当前有3个线程并发

线程pool-1-thread-6即将离开

线程pool-1-thread-8已进入，当前有3个线程并发

线程pool-1-thread-9即将离开

线程pool-1-thread-10已进入，当前有3个线程并发

线程pool-1-thread-7即将离开

线程pool-1-thread-8即将离开

线程pool-1-thread-10即将离开

CyclicBarrier

java.util.concurrent

类 CyclicBarrier

java.lang.Object

java.util.concurrent.CyclicBarrier

public class CyclicBarrierextends Object一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。

在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。

因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。

CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。

若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。

示例用法：下面是一个在并行分解设计中使用 barrier 的例子：

class Solver {
   final int N;
   final float[][] data;
   final CyclicBarrier barrier;
   
   class Worker implements Runnable {
     int myRow;
     Worker(int row) { myRow = row; }
     public void run() {
       while (!done()) {
         processRow(myRow);

         try {
           barrier.await(); 
         } catch (InterruptedException ex) { 
return; 
         } catch (BrokenBarrierException ex) { 
return; 
         }
       }
     }
   }

   public Solver(float[][] matrix) {
     data = matrix;
     N = matrix.length;
     barrier = new CyclicBarrier(N, 
                                 new Runnable() {
                                   public void run() { 
                                     mergeRows(...); 
                                   }
                                 });
     for (int i = 0; i < N; ++i) 
       new Thread(new Worker(i)).start();

     waitUntilDone();
   }
 }

在这个例子中，每个 worker 线程处理矩阵的一行，在处理完所有的行之前，该线程将一直在屏障处等待。

处理完所有的行之后，将执行所提供的 Runnable 屏障操作，并合并这些行。

如果合并者确定已经找到了一个解决方案，那么 done() 将返回 true，所有的 worker 线程都将终止。

如果屏障操作在执行时不依赖于正挂起的线程，则线程组中的任何线程在获得释放时都能执行该操作。

为方便此操作，每次调用 await() 都将返回能到达屏障处的线程的索引。然后，您可以选择哪个线程应该执行屏障操作，例如：

if (barrier.await() == 0) {

// log the completion of this iteration

}对于失败的同步尝试，CyclicBarrier 使用了一种要么全部要么全不 (all-or-none) 的破坏模式：如果因为中断、失败或者超时等原因，导致线程过早地离开了屏障点，那么在该屏障点等待的其他所有线程也将通过 BrokenBarrierException（如果它们几乎同时被中断，则用 InterruptedException）以反常的方式离开。

内存一致性效果：线程中调用 await() 之前的操作 happen-before 那些是屏障操作的一部份的操作，后者依次 happen-before 紧跟在从另一个线程中对应 await() 成功返回的操作。

(1)await

public int await()

throws InterruptedException,

BrokenBarrierException在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。

如果当前线程不是将到达的最后一个线程，出于调度目的，将禁用它，且在发生以下情况之一前，该线程将一直处于休眠状态：

最后一个线程到达；或者

其他某个线程中断当前线程；或者

其他某个线程中断另一个等待线程；或者

其他某个线程在等待 barrier 时超时；或者

其他某个线程在此 barrier 上调用 reset()。

如果当前线程：

在进入此方法时已经设置了该线程的中断状态；或者

在等待时被中断

则抛出 InterruptedException，并且清除当前线程的已中断状态。

如果在线程处于等待状态时 barrier 被 reset()，或者在调用 await 时 barrier 被损坏，抑或任意一个线程正处于等待状态，则抛出 BrokenBarrierException 异常。

如果任何线程在等待时被 中断，则其他所有等待线程都将抛出 BrokenBarrierException 异常，并将 barrier 置于损坏状态。

如果当前线程是最后一个将要到达的线程，并且构造方法中提供了一个非空的屏障操作，则在允许其他线程继续运行之前，当前线程将运行该操作。

如果在执行屏障操作过程中发生异常，则该异常将传播到当前线程中，并将 barrier 置于损坏状态。

返回：

到达的当前线程的索引，其中，索引 getParties() - 1 指示将到达的第一个线程，零指示最后一个到达的线程

抛出：

InterruptedException - 如果当前线程在等待时被中断

BrokenBarrierException - 如果另一个 线程在当前线程等待时被中断或超时，或者重置了 barrier，或者在调用 await 时 barrier 被损坏，抑或由于异常而导致屏障操作（如果存在）失败。

(2)getNumberWaiting

public int getNumberWaiting()返回当前在屏障处等待的参与者数目。此方法主要用于调试和断言。

返回：

当前阻塞在 await() 中的参与者数目。

应用实例：

package com.itm.thread;

import java.util.Random;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CyclicBarrierTest {
	public static void main(String[] args) {
		final CyclicBarrier cb = new CyclicBarrier(3); // 这个团队中共有3个队员,即需要3个线程
		ExecutorService es = Executors.newFixedThreadPool(3); // 在线程池中放入三个线程
		for (int i = 0; i < 3; i++) { // 开启三个任务
			es.execute(new Runnable() {

				@Override
				public void run() {
					for (int i = 0; i < 5; i++) {
						try {
							Thread.sleep(new Random().nextInt(5000));
							System.out.print(Thread.currentThread().getName()
									+ "已到达集合点" + (i + 1) + "，现在共有"
									+ (cb.getNumberWaiting() + 1) + "个线程到达");
							// 如果有2个线程已经在等待，那么最后一个线程到达后就可以一起开始后面操作
							if (cb.getNumberWaiting() + 1 == 3) {
								System.out.println("，全部到齐，出发去下一个目标");
							} else {
								System.out.println("，正在等待");
							}
							cb.await();
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				}
			});
		}
		es.shutdown();
	}
}

运行结果：

pool-1-thread-3已到达集合点1，现在共有1个线程到达，正在等待

pool-1-thread-2已到达集合点1，现在共有2个线程到达，正在等待

pool-1-thread-1已到达集合点1，现在共有3个线程到达，全部到齐，出发去下一个目标

pool-1-thread-3已到达集合点2，现在共有1个线程到达，正在等待

pool-1-thread-1已到达集合点2，现在共有2个线程到达，正在等待

pool-1-thread-2已到达集合点2，现在共有3个线程到达，全部到齐，出发去下一个目标

pool-1-thread-1已到达集合点3，现在共有1个线程到达，正在等待

pool-1-thread-2已到达集合点3，现在共有2个线程到达，正在等待

pool-1-thread-3已到达集合点3，现在共有3个线程到达，全部到齐，出发去下一个目标

pool-1-thread-1已到达集合点4，现在共有1个线程到达，正在等待

pool-1-thread-2已到达集合点4，现在共有2个线程到达，正在等待

pool-1-thread-3已到达集合点4，现在共有3个线程到达，全部到齐，出发去下一个目标

pool-1-thread-2已到达集合点5，现在共有1个线程到达，正在等待

pool-1-thread-3已到达集合点5，现在共有2个线程到达，正在等待

pool-1-thread-1已到达集合点5，现在共有3个线程到达，全部到齐，出发去下一个目标