53、java.util.concurrent.其它同步工具类

一、java.util.concurrent.Semaphore

/**
 * Semaphore
 * 一个计数信号量。
 * 从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。
 * 每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。
 * Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目。
 * 获得一项前，每个线程必须从信号量获取许可，从而保证可以使用该项。
 * 该线程结束后，将项返回到池中并将许可返回到该信号量，从而允许其他线程获取该项。
 * 注意，调用 acquire() 时无法保持同步锁，因为这会阻止将项返回到池中。信号量封装所需的同步，以限制对池的访问，这同维持该池本身一致性所需的同步是分开的。
 * 将信号量初始化为 1，使得它在使用时最多只有一个可用的许可，从而可用作一个相互排斥的锁。
 * 这通常也称为二进制信号量，因为它只能有两种状态：一个可用的许可，或零个可用的许可。
 * 按此方式使用时，二进制信号量具有某种属性（与很多 Lock 实现不同），即可以由线程释放“锁”，而不是由所有者（因为信号量没有所有权的概念）。在某些专门的上下文（如死锁恢复）中这会很有用
 * 此类的构造方法可选地接受一个公平 参数。当设置为 false 时，此类不对线程获取许可的顺序做任何保证
 * 当公平设置为 true 时，信号量保证对于任何调用获取方法的线程而言，都按照处理它们调用这些方法的顺序（即先进先出；FIFO）来选择线程、获得许可。
 * 非同步的 tryAcquire 方法不使用公平设置，而是使用任意可用的许可。 
 * 通常，应该将用于控制资源访问的信号量初始化为公平的，以确保所有线程都可访问资源。
 * 为其他的种类的同步控制使用信号量时，非公平排序的吞吐量优势通常要比公平考虑更为重要。 
 * 此类还提供便捷的方法来同时 acquire 和释放多个许可。小心，在未将公平设置为 true 时使用这些方法会增加不确定延期的风险
 */
public class Semaphore implements Serializable
{
	//构造方法

	//创建具有给定的许可数和非公平的公平设置的 Semaphore
	public Semaphore(int permits){}

	//创建具有给定的许可数和给定的公平设置的 Semaphore
	public Semaphore(int permits,
                 boolean fair){}

	//常用方法

	//从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。
	public void acquire()
             throws InterruptedException{}

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

	//返回此信号量中当前可用的许可数。 此方法通常用于调试和测试目的。
	public int availablePermits(){}

	//返回标识此信号量的字符串，以及信号量的状态。括号中的状态包括 String 类型的 "Permits ="，后跟许可数。
	public String toString(){}

	//返回正在等待获取的线程的估计数目
	public final int getQueueLength(){}

	//返回一个 collection，包含可能等待获取的线程。
	protected Collection<Thread> getQueuedThreads()
}

 

二、示例

import java.util.concurrent.*;

public class SemaphoreTest
{
	public static void main(String[] args)
	{
		ExecutorService threadPool = Executors.newCachedThreadPool();
		final Semaphore semaphore = new Semaphore(3);
		for(int i=0;i<10;i++)
		{
			Runnable run = new Runnable(){
				public void run(){
					try{
						semaphore.acquire();
					}catch ( InterruptedException e ){}
					System.out.println("线程"+Thread.currentThread().getName()+"进入，当前已经有 "+(3-semaphore.availablePermits())+" 个线程进入");
					try{
						Thread.sleep(1000);
					}catch ( InterruptedException e ){}
					System.out.println("线程"+Thread.currentThread().getName()+"即将离开");
					semaphore.release();
					System.out.println("线程"+Thread.currentThread().getName()+"已经离开，当前剩余 "+semaphore.availablePermits()+" 个信号量");
				}
			};
			threadPool.execute(run);
		}
		threadPool.shutdown();
	}
}

 

• 由上例可以看出，其实Semaphore也相当于一个锁，只不过可以指定并发的线程数，而不是一次只能进一个线程

三、java.util.concurrent.CyclicBarrier

/**
 * CyclicBarrier
 * 一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。
 * 在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。
 * 因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。 
 * CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。
 * 若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。
 */
public class CyclicBarrier
{
	//创建一个新的 CyclicBarrier，它将在给定数量的参与者（线程）处于等待状态时启动，但它不会在启动 barrier 时执行预定义的操作
	public CyclicBarrier(int parties){}

	//创建一个新的 CyclicBarrier，它将在给定数量的参与者（线程）处于等待状态时启动，并在启动 barrier 时执行给定的屏障操作，该操作由最后一个进入 barrier 的线程执行。
	public CyclicBarrier(int parties,
                     Runnable barrierAction){}

	//在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。
	public int await()
          throws InterruptedException,
                 BrokenBarrierException{}

	//返回当前在屏障处等待的参与者数目。
	public int getNumberWaiting(){}

	//返回要求启动此 barrier 的参与者数目。 
	public int getParties(){}
}

 

四、示例

import java.util.concurrent.*;
import java.util.*;

class  CyclicBarrierTest
{
	public static void main(String[] args) 
	{
		ExecutorService threadPool = Executors.newCachedThreadPool();
		final CyclicBarrier barrier = new CyclicBarrier(3,new Runnable(){
			public void run(){
				System.out.println(Thread.currentThread().getName()+" 所有人都到齐了，继续。。。");
			}
		});
		for(int i=0;i<3;i++)
		{
			Runnable run = new Runnable(){
				public void run(){
					try{
						Thread.sleep(1000*(new Random().nextInt(10)));
						System.out.println("线程 "+Thread.currentThread().getName()+" 已到达集合点1，当前已有 "+(barrier.getNumberWaiting()+1)+" 个人在等待");
						barrier.await();
						Thread.sleep(1000*(new Random().nextInt(10)));
						System.out.println("线程 "+Thread.currentThread().getName()+" 已到达集合点2，当前已有 "+(barrier.getNumberWaiting()+1)+" 个人在等待");
						barrier.await();
						Thread.sleep(1000*(new Random().nextInt(10)));
						System.out.println("线程 "+Thread.currentThread().getName()+" 已到达集合点3，当前已有 "+(barrier.getNumberWaiting()+1)+" 个人在等待");
						barrier.await();
						Thread.sleep(1000*(new Random().nextInt(10)));
						System.out.println("线程 "+Thread.currentThread().getName()+" 已到达集合点4，当前已有 "+(barrier.getNumberWaiting()+1)+" 个人在等待");
						barrier.await();
					}catch(Exception e){}
				}
			};
			threadPool.execute(run);
		}
		threadPool.shutdown();
	}
}

 

五、java.util.concurrent.CountDownLatch

/**
 * CountDownLatch
 * 一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。 
 * 用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。
 * 之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。
 * 这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。
 * CountDownLatch 是一个通用同步工具，它有很多用途。
 * 将计数 1 初始化的 CountDownLatch 用作一个简单的开/关锁存器，或入口：
 * 在通过调用 countDown() 的线程打开入口前，所有调用 await 的线程都一直在入口处等待。
 * 用 N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待，或者使其在某项操作完成 N 次之前一直等待。
 */
public class CountDownLatch
{
	//构造一个用给定计数初始化的 CountDownLatch
	public CountDownLatch(int count){}

	//使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断。 
	public void await()
           throws InterruptedException{}

	//递减锁存器的计数，如果计数到达零，则释放所有等待的线程
	public void countDown(){}

	//返回当前计数。此方法通常用于调试和测试
	public long getCount(){}
}

 

六、示例

import java.util.concurrent.*;
import java.util.*;

class  CountDownLatchTest
{
	public static void main(String[] args) 
	{
		ExecutorService threadPool = Executors.newCachedThreadPool();
		final CountDownLatch cdHandler = new CountDownLatch(1);
		final CountDownLatch cdSportsman = new CountDownLatch(3);
		final List<String> result = new ArrayList<String>();
		for(int i=0;i<3;i++)
		{
			Runnable run = new Runnable(){
				public void run(){
					try{
						System.out.println("运动员："+Thread.currentThread().getName()+"正在准备冲刺。。。");
						cdHandler.await();
						System.out.println("运动员："+Thread.currentThread().getName()+"开始冲刺--->>>>");
						Thread.sleep((long)(Math.random()*10000));
						System.out.println("运动员："+Thread.currentThread().getName()+"已达到终点！");
						result.add(Thread.currentThread().getName());
						cdSportsman.countDown();
					}catch (Exception e ){}
				}
			};
			threadPool.execute(run);
		}
		try{
			Thread.sleep((long)(Math.random()*10000));
			System.out.println("裁判员："+Thread.currentThread().getName()+"扣动扳机！ 啪！！！！！！！！");
			cdHandler.countDown();
			cdSportsman.await();
			System.out.println("比赛结束，裁判员："+Thread.currentThread().getName()+"公布比赛结果：");
			for(int i=1;i<4;i++)
				System.out.println("第"+i+"名: "+result.get(i-1));
		}catch(Exception e){}
		threadPool.shutdown();
	}
}

 

七、java.util.concurrent.Exchanger<V>

/**
 * Exchanger<V>
 * 可以在对中对元素进行配对和交换的线程的同步点。
 * 每个线程将条目上的某个方法呈现给 exchange 方法，与伙伴线程进行匹配，并且在返回时接收其伙伴的对象。
 * Exchanger 可能被视为 SynchronousQueue 的双向形式。Exchanger 可能在应用程序（比如遗传算法和管道设计）中很有用。
 */
public class Exchanger<V>
{
	//创建一个新的 Exchanger。
	public Exchanger(){}

	//等待另一个线程到达此交换点（除非当前线程被中断），然后将给定的对象传送给该线程，并接收该线程的对象。 
	public V exchange(V x)
           throws InterruptedException{}
}

 

八、示例

import java.util.concurrent.*;

class  ExchangerTest
{
	private static final Exchanger<String> changer = new Exchanger<String>();

	static class Solder implements Runnable
	{
		public void run(){
			try{
				Thread.sleep((long)(Math.random()*10000));
				System.out.println("卖东西的: "+Thread.currentThread().getName()+"达到指定地点");
				String data = "物品";
				String get = changer.exchange(data);
				System.out.println("卖东西的: "+Thread.currentThread().getName()+"收到"+get+"，马上撤退....");
			}catch(Exception e){}
		}
	}
	static class Buyer implements Runnable
	{
		public void run(){
			try{
				Thread.sleep((long)(Math.random()*10000));
				System.out.println("----买东西的: "+Thread.currentThread().getName()+"达到指定地点");
				String data = "money";
				String get = changer.exchange(data);
				System.out.println("----买东西的: "+Thread.currentThread().getName()+"收到"+get+"，马上撤退....");
			}catch(Exception e){}
		}
	}
	public static void main(String[] args) 
	{
		new Thread(new Solder()).start();
		new Thread(new Buyer()).start();
	}
}