并发编程学习笔记6

同步器
在集合类中阻塞队列是对一无二的：不仅仅因为他是对象的容器，而且还因为他能协调生产者和消费者线程的控制流程，这得归功于take和put方法的阻塞。

同步器可以是任何能基于状态协调线程控制流程的对象。阻塞线程可以当作同步器；其他类型的同步器包括semaphores,barriers,latches.这些同步器类在jdk中就有提供。

所有的同步器都具有某些基础的属性：他们封装了一些状态，用来决定已经到达了同步器的线程是允许通过还是强迫他等待，而且还提供了一些方法来操作这些状态，另外还提供了方法来等待同步器进入想要的状态。

Latches(门闩)
该同步器可以延迟线程的执行，直到同步器进入终结状态。Latch扮演了门的角色：在门闩进入终结状态之前，门是关闭的，没有任何线程可以通过。而当进入终结状态，即门开的时候，则允许任何线程通过。一旦门闩成为终结状态，那么他将不能在修改他的状态了，所以他将永远打开。Latches还可以被用来保证某些活动在其他活动结束之前不被执行，比如：

1.保证在资源初始化完毕之前，任何计算线程不能执行。一个简单的二进制latch就可以用来标记资源已经初始化完毕，在这之前，任何活动都将等待这个latch
2.保证一个服务在依赖的服务启动之后再启动.每个服务都和二进制的latch有关联;启动服务S，那么依赖于S的服务都将会等待，直到S启动，然后S的latch将被释放，这样，任何依赖于S的服务都可以继续执行。
3.在一个活动的各个部分都准备好之前，一直等待，比如多人游戏，在这种情况，latch在所有的player都准备好之后，将释放，进入最终状态。

CountDownLatch
这个比较有意思，当计数倒数到0的时候，所有线程执行

package com.company.jncz;

import java.util.concurrent.CountDownLatch;

public class LatchTest {
	class Worker implements Runnable{
		private CountDownLatch start;
		private CountDownLatch end;
		private int i;
		Worker(CountDownLatch start,CountDownLatch end, int i){
			this.start = start;
			this.end = end;
			this.i = i;
		}
		public void run() {
			try {
				start.await();//如果注释这句，则任务被创建出来之后即执行，不会等待startSignal.countDown()这个启动信号
				dowork();
				System.out.println("第"+i+"个任务执行");
			} catch (Exception e) {
				e.printStackTrace();
				Thread.currentThread().interrupt();
			}
			end.countDown();
		}
		private void dowork() {
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();//TODO
			}
		}
		
	}
	
	public static void main(String args[]) throws InterruptedException{
		int num = 50;
		LatchTest lt = new LatchTest();
		CountDownLatch startSignal = new CountDownLatch(1);
		CountDownLatch endSignal = new CountDownLatch(num);
		for(int i=0;i<num;i++){
			new Thread(lt.new Worker(startSignal,endSignal, i)).start();
		}
		System.out.println("任务开始执行");
		startSignal.countDown();//开始执行任务
		try {
			endSignal.await();//等待所有线程结束执行
		} catch (InterruptedException e) {
			e.printStackTrace();//TODO
		}
		System.out.println("执行完毕");
	}
}

在这段代码中startSignal表示了一个信号量，如同跑步的时候，发令员的枪声，由于startSignal初始值是1，所以一旦执行了startSignal.countDown()，那么立即就成了0，那么所有任务立即开始执行，而在Worker内部，endSignal不断的在countDown，所以随着任务的执行endSignal也慢慢接近于0，一旦所有任务执行完毕，则endSignal为0，则endSignal.await()方法通过，那么后续的工作可以继续进行。

FutureTask
FutureTask也可以当作一个latch。(FutureTask实现了Future接口，他描述了一种抽象的因果关系)。FutureTask代表的运算，最终是由一个实现Callable接口的类实现的，他可以处于以下三种状态:waiting to run,running,completed。所谓完成状态代表了正常完成，取消和异常。一旦一个FutureTask进入完成状态，那么他将永远处于完成状态。

Future.get的行为取决于任务的状态。如果是完成状态，则get立即返回结果,否则将阻塞，直到任务状态为完成才返回结果，或者抛出异常。FutureTask将结果从计算线程转移到接收线程;FutureTask的规范保证了这种转移始终安全的结果发布。

FutureTask被Executor框架用来表示异步任务，或者用来表示任意的耗时的计算(而该计算又不是很迫切需要返回结果).

Semaphore
计数信号量通常用来控制活动的数目，比如对资源的访问或者在某个时刻运行某个动作。计数信号量可以被用来实现资源池或者给集合加上边界限制。
Semaphore管理了一些虚拟的许可，任何活动可以获取许可，当他们做完之后，可以释放许可，如果没有许可了，则acquire会阻塞，直到有可用的许可(或被中断或者超时)。而release方法则将许可返回给Semaphore。如果许可数目初始化成1，则相当于一个非可重入的互斥锁。谁获取了许可谁就获得了这个互斥锁。

这里所说的许可，并不跟任何线程挂钩，所以许可可以在A线程被获得，而在B线程被返回给Semaphore。可以认为acquire是消费许可，而release是创建许可。

public class BoundedHashSet<T> {
    private final Set<T> set;
    private final Semaphore sem;

    public BoundedHashSet(int bound) {
        this.set = Collections.synchronizedSet(new HashSet<T>());
        sem = new Semaphore(bound);
    }

    public boolean add(T o) throws InterruptedException {
        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)
            sem.release();
        return wasRemoved;
    }
}

代码中的bound就是这个集合的边界，只能有这么长。否则获取许可的时候，已经没有许可可以给了，只能阻塞，只有在有一个元素被remove之后，许可才会有。
Barriers

package com.company.jncz.concurrent;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class BarrierTest {

	class Soldier implements Runnable{
		private CyclicBarrier cb;
		private int i;
		Soldier(CyclicBarrier cb,int i){
			this.i = i;
			this.cb = cb;
		}
		public void run() {
			try {
				System.out.println("士兵"+i+"号到达目的地，并等待其他士兵");
				cb.await();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
	}
	
	class AfterWork implements Runnable{

		public void run() {
			System.out.println("士兵到齐，出发");
		}
		
	}
	public static void main(String args[]){
		BarrierTest bt = new BarrierTest();
		int num = 50;
		CyclicBarrier cb = new CyclicBarrier(num,bt.new AfterWork());
		for (int i = 0; i < num; i++) {
			new Thread(bt.new Soldier(cb,i)).start();
		}
	}
}

这个类模拟了Barrier的作用，他会等待所有线程都完毕之后，才会执行其指定的Runnable接口。
如上代码，当每个线程运行完毕之后，会打印出“士兵到位”的信息，并会await，直到其他士兵全部到齐，然后会运行AfterWork这个类。

另外一种Barrier是Exchanger，他由两个barrier组成，每个部分在边界点交换数据。当在运行不对成的活动时很有用，比如当一个线程填充了buffer，另一个线程从buffer中消费数据；这些线程可以用Exchanger来交换数据。当两个线程通过Exchanger交互了对象，这个交换对于两个线程来说都是安全的。

package com.company.jncz.concurrent;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Exchanger;

public class ExchangerTest {
	private static final Exchanger ex = new Exchanger();
	class DataProducer implements Runnable{
		private List list = new ArrayList();
		public void run() {
			for (int i = 0; i < 10; i++) {
				System.out.println("生产了一个数据，耗时1秒");
				list.add(new Date());
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			try {
				ex.exchange(list);//将数据准备用于交换
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
	}
	
	class DataConsumer implements Runnable{
		private List list = new ArrayList();
		public void run() {
			try {
				list = (List) ex.exchange(list);//进行交换数据
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			for (Iterator iterator = list.iterator(); iterator.hasNext();) {
				Date d = (Date) iterator.next();
				System.out.println(d);
			}
		}
		
	}
	
	public static void main(String args[]){
		ExchangerTest et = new ExchangerTest();
		new Thread(et.new DataProducer()).start();
		new Thread(et.new DataConsumer()).start();
	}
}