jdk1.5的多线程总结二

新的Synchronizer：
Java 5.0里新加了4个协调线程间进程的同步装置，它们分别是Semaphore, CountDownLatch, CyclicBarrier和Exchanger。

Semaphore:
用来管理一个资源池的工具，Semaphore可以看成是个通行证，线程要想从资源池拿到资源必须先拿到通行证，Semaphore提供的通行证数量和资源池的大小一致。如果线程暂时拿不到通行证，线程就会被阻断进入等待状态。

public class Pool {
	List<String> pool = null;

	Semaphore pass = null;

	public Pool(int size) {
		// 初始化资源池
		pool = new ArrayList<String>();
		for (int i = 0; i < size; i++) {
			pool.add("Resource " + i);
		}
		// Semaphore的大小和资源池的大小一致
		pass = new Semaphore(size);
	}

	public String get() throws InterruptedException {
		// 获取通行证,只有得到通行证后才能得到资源
		pass.acquire();
		return getResource();
	}

	public void put(String resource) {
		// 归还通行证，并归还资源
		pass.release();
		releaseResource(resource);
	}

	private synchronized String getResource() {
		String result = pool.get(0);
		pool.remove(0);
		System.out.println("Take " + result);
		return result;
	}

	private synchronized void releaseResource(String resource) {
		System.out.println("return " + resource);
		pool.add(resource);
	}
}

public class SemaphoreTest {
	public static void main(String[] args) {

		final Pool aPool = new Pool(2);
		Runnable worker = new Runnable() {
			public void run() {
				String resource = null;
				try {
					// 取得resource
					resource = aPool.get();
				} catch (InterruptedException ex) {
					ex.printStackTrace();
				}
				// 用resource做工作
				System.out.println("I worked on " + resource);
				// 归还resource
				aPool.put(resource);
			}
		};

		ExecutorService service = Executors.newCachedThreadPool();
		for (int i = 0; i < 20; i++) {
			service.submit(worker);
		}
		service.shutdown();
	}
}

CountDownLatch:
CountDownLatch是个计数器，它有一个初始数，等待这个计数器的线程必须等到计数器倒数到零时才可继续。比如说一个Server启动时需要初始化4个部件，Server可以同时启动4个线程去初始化这4个部件，然后调用CountDownLatch(4).await()阻断进入等待，每个线程完成任务后会调用一次CountDownLatch.countDown()来倒计数, 当4个线程都结束时CountDownLatch的计数就会降低为0，此时Server就会被唤醒继续下一步操作。CountDownLatch的方法主要有：
await()：使调用此方法的线程阻断进入等待
countDown(): 倒计数，将计数值减1
getCount(): 得到当前的计数值

public class Server {
	public static void main(String[] args) throws InterruptedException{
		//一个server调了三个ComponentThread分别去启动三个组件，然后server等到组件都启动了再继续
        System.out.println("Server is starting.");
        //初始化一个初始值为3的CountDownLatch
        CountDownLatch latch = new CountDownLatch(3);
        //起3个线程分别去启动3个组件
        ExecutorService service = Executors.newCachedThreadPool();
        service.submit(new ComponentThread(latch, 1));
        service.submit(new ComponentThread(latch, 2));
        service.submit(new ComponentThread(latch, 3));
        service.shutdown();
        //进入等待状态
        latch.await();
        //当所需的三个组件都完成时，Server就可继续了
        System.out.println("Server is up!");
  }
}

public class ComponentThread implements Runnable {

	CountDownLatch latch;

	int ID;

	public ComponentThread(CountDownLatch latch, int ID) {
		this.latch = latch;
		this.ID = ID;
	}

	public void run() {
		System.out.println("Component " + ID + " initialized!");
		// 将计数减一
		latch.countDown();
	}
}

CyclicBarrier:
CyclicBarrier类似于CountDownLatch也是个计数器，不同的是CyclicBarrier数的是调用了CyclicBarrier.await()进入等待的线程数，当线程数达到了CyclicBarrier初始时规定的数目时，所有进入等待状态的线程被唤醒并继续。CyclicBarrier就象它名字的意思一样，可看成是个障碍，所有的线程必须到齐后才能一起通过这个障碍。CyclicBarrier初始时还可带一个Runnable的参数，此Runnable任务在CyclicBarrier的数目达到后，所有其它线程被唤醒前被执行。
CyclicBarrier提供以下几个方法：
await()：进入等待
getParties()：返回此barrier需要的线程数
reset()：将此barrier重置

public class CyclicBarrierTest {
	public static void main(String[] args) throws InterruptedException,
			BrokenBarrierException, TimeoutException {
		// 两个线程分别在一个数组里放一个数，当这两个线程都结束后，主线程算出数组里的数的和
		final int[] array = new int[2];
		CyclicBarrier barrier = new CyclicBarrier(2, new Runnable() {
			public void run() {
				System.out.println("Total is:" + (array[0] + array[1]));
			}
		});

		// 启动线程
		new Thread(new ComponentThread(barrier, array, 0)).start();
		new Thread(new ComponentThread(barrier, array, 1)).start();
	}
}

public class ComponentThread implements Runnable{
	CyclicBarrier barrier;

    int ID;

    int[] array;

    public ComponentThread(CyclicBarrier barrier, int[] array, int ID) {
          this.barrier = barrier;
          this.ID = ID;
          this.array = array;
    }

    public void run() {
          try {
                array[ID] = new Random().nextInt();
                System.out.println(ID+ " generates:"+array[ID]);
                //该线程完成了任务等在Barrier处
                barrier.await();
          } catch (BrokenBarrierException ex) {
                ex.printStackTrace();
          } catch (InterruptedException ex) {
               ex.printStackTrace();
          }
    }
}

Exchanger:
让两个线程可以互换信息。这个跟生产消费模式有点不同，生产消费模式是生产者和消费者对同一个数据争用,这里是对两个同类型的数据争用，用完就互换。用一个例子来解释比较容易。例子中服务生线程往空的杯子里倒水，顾客线程从装满水的杯子里喝水，然后通过Exchanger双方互换杯子，服务生接着往空杯子里倒水，顾客接着喝水，然后交换，如此周而复始。两个线程同时只对自己的那个数据操作。


BlockingQueue接口
BlockingQueue是一种特殊的Queue，若BlockingQueue是空的，从BlockingQueue取东西的操作将会被阻断进入等待状态直到BlocingkQueue进了新货才会被唤醒。同样，如果BlockingQueue是满的任何试图往里存东西的操作也会被阻断进入等待状态，直到BlockingQueue里有新的空间才会被唤醒继续操作。BlockingQueue提供的方法主要有：
add(anObject): 把anObject加到BlockingQueue里，如果BlockingQueue可以容纳返回true，否则抛出IllegalStateException异常。
offer(anObject)：把anObject加到BlockingQueue里，如果BlockingQueue可以容纳返回true，否则返回false。
put(anObject)：把anObject加到BlockingQueue里，如果BlockingQueue没有空间，调用此方法的线程被阻断直到BlockingQueue里有新的空间再继续。
poll(time)：取出BlockingQueue里排在首位的对象，若不能立即取出可等time参数规定的时间。取不到时返回null。
take()：取出BlockingQueue里排在首位的对象，若BlockingQueue为空，阻断进入等待状态直到BlockingQueue有新的对象被加入为止。

根据不同的需要BlockingQueue有4种具体实现：
ArrayBlockingQueue：规定大小的BlockingQueue，其构造函数必须带一个int参数来指明其大小。其所含的对象是以FIFO（先入先出）顺序排序的。
LinkedBlockingQueue：大小不定的BlockingQueue，若其构造函数带一个规定大小的参数，生成的BlockingQueue有大小限制，若不带大小参数，所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定。其所含的对象是以FIFO（先入先出）顺序排序的。LinkedBlockingQueue和ArrayBlockingQueue比较起来，它们背后所用的数据结构不一样，导致LinkedBlockingQueue的数据吞吐量要大于ArrayBlockingQueue，但在线程数量很大时其性能的可预见性低于ArrayBlockingQueue。
PriorityBlockingQueue：类似于LinkedBlockingQueue，但其所含对象的排序不是FIFO，而是依据对象的自然排序顺序或者是构造函数所带的Comparator决定的顺序。
SynchronousQueue：特殊的BlockingQueue，对其的操作必须是放和取交替完成的。
下面是用BlockingQueue来实现经典的Producer和Consumer模式

public class BlockingQueueTest {
	static BlockingQueue<String> basket;

	public BlockingQueueTest() {
		// 定义了一个大小为2的BlockingQueue，也可根据需要用其他的具体类
		basket = new ArrayBlockingQueue<String>(2);
	}

	class Producor implements Runnable {

		public void run() {
			while (true) {
				try {
					// 放入一个对象，若basket满了，等到basket有位置
					basket.put("An apple");
				} catch (InterruptedException ex) {
					ex.printStackTrace();
				}
			}
		}
	}

	class Consumer implements Runnable {
		public void run() {
			while (true) {
				try {
					// 取出一个对象，若basket为空，等到basket有东西为止
					String result = basket.take();
				} catch (InterruptedException ex) {
					ex.printStackTrace();
				}
			}
		}
	}

	public void execute() {
		for (int i = 0; i < 10; i++) {
			new Thread(new Producor()).start();
			new Thread(new Consumer()).start();
		}
	}

	public static void main(String[] args) {
		BlockingQueueTest test = new BlockingQueueTest();
		test.execute();
	}
}

Atomics 原子级变量
主要的类有AtomicBoolean, AtomicInteger, AotmicIntegerArray, AtomicLong, AtomicLongArray, AtomicReference ……。这些原子量级的变量主要提供两个方法：
compareAndSet(expectedValue, newValue): 比较当前的值是否等于expectedValue,若等于把当前值改成newValue，并返回true。若不等，返回false。
getAndSet(newValue): 把当前值改为newValue，并返回改变前的值。
这些原子级变量利用了CPU的硬件支持可把两步操作合为一步的功能，避免了不必要的锁定，提高了程序的运行效率。