java并发中的协同工具类介绍-CountDownLatch-CyclicBarrier-Semphone-Exchanger

CountDownLatch:闭锁

这个工具类为多个线程在同一个地方进行等待，或者等待某些事情完成，才能继续执行接下来的任务，看代码示例:

public class TestHarness
{
public long timeTasks(int nThread,final Runnable task) throws InterruptedException{

//开始开关
final CountDownLatch startGate = new CountDownLatch(1);
final CountDownLatch endGate = new CountDownLatch(nThread);

for (int i = 0; i < nThread; i++)
{
Thread t = new Thread(){
public void run(){
try
{
//等待开关打开
startGate.await();
task.run();
}
catch (Exception e)
{
// TODO: handle exception
}finally{
//任务完成，endGate-1
endGate.countDown();
}
}
};
t.start();
}
long start = System.nanoTime();
//开关打开，所有线程同一起跑线开始执行
startGate.countDown();
//等待所有线程执行完成
endGate.await();
long end = System.nanoTime();
//返回时间
return end-start;
}
}

Exchanger---两个线程间数据交换

从名字上就可以看出，这个工具类用于线程间数据的交换，线程会阻塞在Exchanger的exchange方法上，直到另外一个线程也执行到同一个Exchanger的exchange方法，二者进行数据的交换，然后彼此各自执行各自的任务，看如下代码:

public static void main(String[] args)
{
final Exchanger<List<Integer>> exchanger = new Exchanger<List<Integer>>();
new Thread()
{
@Override
public void run()
{
List<Integer> l = new ArrayList<Integer>(2);
l.add(1);
l.add(2);
try
{
l=exchanger.exchange(l);
}
catch (InterruptedException e)
{

e.printStackTrace();
}
System.out.println("thread1:"+Thread.currentThread().getName()+":"+l);
}
}.start();

new Thread()
{
@Override
public void run()
{
List<Integer> l = new ArrayList<Integer>(2);
l.add(4);
l.add(5);
try
{
l = exchanger.exchange(l);
}
catch (InterruptedException e)
{

e.printStackTrace();
}
System.out.println("thread2:"+Thread.currentThread().getName()+":"+l);
}
}.start();

}

Semaphore:信号量

计数信号量是用来控制资源访问的操作数量，或者对执行的线程数进行控制，可以用来实现某种资源池，或者进行容器边界控制，即实现有界容器。代码实例如下:

public class BounderHashSet<T>
{
private final Set<T> set;

private final Semaphore sem;

public BounderHashSet(int bound){
//构造函数中，初始化set为同步容器类
this.set = Collections.synchronizedSet(new HashSet<T>());

//初始化容器的大小，用sem来控制容器
sem = new Semaphore(bound);
}

public boolean add(T o) throws InterruptedException{
//获取许可，其实就是表明，容器的可用空间减1
sem.acquire();
boolean wasAdded = false;
try
{
wasAdded = set.add(o);
return wasAdded;
}finally{
if (!wasAdded)
{
//如果添加不成功，释放许可
sem.release();
}
}
}

public boolean remove(Object o){
boolean wasRemoved = set.remove(o);
if (wasRemoved)
{
//释放许可，可用空间+1
sem.release();
}
return wasRemoved;
}

}

实现一个有界队列：

public class BounderBuffer<E>
{
//可用的元素
private final Semaphore availableItems;

//可用的空闲空间
private final Semaphore availableSpaces;

//容器底层为数组
private final E[] items;

private int putPosition = 0;

private int takePosition = 0;

public BounderBuffer(int capacity){
//初始可用数据为0
availableItems = new Semaphore(0);

//空闲空间数量
availableSpaces = new Semaphore(capacity);

//初始化数组
items = (E[])new Object[capacity];
}

public boolean isEmpty(){
return availableItems.availablePermits()==0;
}

public boolean isFull(){
return availableSpaces.availablePermits()==0;
}

public void put(E x) throws InterruptedException{
availableSpaces.acquire();
//doInsert
doInsert(x);
availableItems.release();
}

public E take() throws InterruptedException{

availableItems.acquire();
//doExtract
E x = doExtract();
availableSpaces.release();
return x;
}

private synchronized void doInsert(E x){
int i = putPosition;
items[i]=x;
putPosition = ++i%items.length;
}

private synchronized E doExtract(){
int i = takePosition;
E x = items[i];
items[i]=null;
takePosition = ++i%items.length;
return  x;
}

}