java并发编程之同步器代码示例

package hb.java.thread;
import java.util.concurrent.CountDownLatch;
/**
 * 
 * @author hb
 *     CountDownLatch最重要的方法是countDown()和await()，前者主要是倒数一次，后者是等待倒数到0，如果没有到达0
 *     ，就只有阻塞等待了。 *JAVA同步器之
 *     CountDownLatch（不能循环使用，如果需要循环使用可以考虑使用CyclicBarrier） 两种比较常规用法： 1：new
 *     CountDownLatch(1);所有的线程在开始工作前需要做一些准备工作，当所有的线程都准备到位后再统一执行时有用 2：new
 *     CountDownLatch(THREAD_COUNT);当所有的线程都执行完毕后，等待这些线程的其他线程才开始继续执行时有用
 */
public class CountDownLatchTest {
	private static final int THREAD_COUNT = 10;
	// 在调用startSingal.countDown()之前调用了startSingal.await()的线程一律等待，直到startSingal.countDown()的调用
	private static final CountDownLatch startSingal = new CountDownLatch(1);
	// 在finishedSingal的初始化记数量通过调用finishedSingal.countDown()减少为0时调用了finishedSingal.await()的线程一直阻塞
	private static final CountDownLatch finishedSingal = new CountDownLatch(
				THREAD_COUNT);
	public static void main(String[] args) throws InterruptedException {
		for (int i = 0; i < THREAD_COUNT; i++) {
			new Thread("Task " + i) {
				public void run() {
					System.out.println(Thread.currentThread().getName()
												+ " prepared!!");
					try {
						startSingal.await();
					}
					catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println(Thread.currentThread().getName()
												+ " finished!!");
					finishedSingal.countDown();
				}
				;
			}
			.start();
		}
		Thread.sleep(1000);
		startSingal.countDown();
		// 所有的线程被唤醒，同时开始工作.countDown 方法的线程等到计数到达零时才继续
		finishedSingal.await();
		// 等待所有的线程完成!!
		System.out.println("All task are finished!!");
	}
}

package hb.java.thread;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
/** 
 * 
 * JAVA同步器之Barrier(能够循环使用，当计数器增加到Barrier的初始化计数器之后马上会被置为0为下一次循环使用做准备) 
 * Barrier能够为指定的一个或多个（一般为多个）线程设置一道屏障，只有当所有的线程都到达该屏障后才能一起冲过该屏障继续其他任务 一般可以new 
 * CyclicBarrier(ThreadCount)来进行初始化，也可以new 
 * CyclicBarrier(ThreadCount,RunableAction)当初始化数量的线程都调用 
 * 了await()方法后触发RunableAction线程，也可以通过初始化一个new 
 * CyclicBarrier(ThreadCount+1)的Barrier在前置线程未执行完成时一直阻塞一个或多个 
 * 后续线程，这一点类似于CountDownLatch 
 */
public class BarrierTest {
	private static final int THREAD_COUNT = 10;
	private static final CyclicBarrier barrier = new CyclicBarrier( 
	      THREAD_COUNT + 1, new Runnable() {
		@Override 
		        public void run() {
			System.out.println("All task are prepared or finished!!");
		}
	}
	);
	public static void main(String[] args) throws InterruptedException, 
	      BrokenBarrierException {
		for (int i = 0; i < THREAD_COUNT; i++) {
			new Thread("Task " + i) {
				public void run() {
					try {
						System.out.println(Thread.currentThread().getName() 
						                + " prepared!!");
						barrier.await();
					}
					catch (InterruptedException e) {
						// TODO Auto-generated catch block 
						e.printStackTrace();
					}
					catch (BrokenBarrierException e) {
						// TODO Auto-generated catch block 
						e.printStackTrace();
					}
					// do something 
					System.out.println(Thread.currentThread().getName() 
					              + " finished!!");
				}
				;
			}
			.start();
		}
		barrier.await();
		// --------------开始准备循环使用-------------- 
		for (int i = 0; i < THREAD_COUNT; i++) {
			new Thread("Task " + i) {
				public void run() {
					// do something 
					System.out.println(Thread.currentThread().getName() 
					              + " finished!!");
					try {
						barrier.await();
					}
					catch (InterruptedException e) {
						// TODO Auto-generated catch block 
						e.printStackTrace();
					}
					catch (BrokenBarrierException e) {
						// TODO Auto-generated catch block 
						e.printStackTrace();
					}
				}
				;
			}
			.start();
		}
		barrier.await();
	}
}

package hb.java.thread;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Exchanger;
public class ExchangerTest {
	final static Exchanger> exchanger = new Exchanger>();
	public static void main(String[] args) {
		new Producer("Producer", exchanger).start();
		new Consumer("Consumer", exchanger).start();
	}
	static class Producer extends Thread {
		private Exchanger> exchanger;
		/** 
     *  
     */
		public Producer(String threadName, Exchanger> exchanger) {
			super(threadName);
			this.exchanger = exchanger;
		}
		/* 
     * (non-Javadoc) 
     * 
     * @see java.lang.Thread#run() 
     */
		@Override 
		    public void run() {
			List products = new ArrayList();
			for (int i = 0; i < 10; i++) {
				products.add("product " + i);
			}
			try {
				List results = exchanger.exchange(products);
				System.out.println("get results from consumer");
				for (String s : results) {
					System.out.println(s);
				}
			}
			catch (InterruptedException e) {
				// TODO Auto-generated catch block 
				e.printStackTrace();
			}
		}
	}
	static class Consumer extends Thread {
		private Exchanger> exchanger;
		/** 
     *  
     */
		public Consumer(String threadName, Exchanger> exchanger) {
			super(threadName);
			this.exchanger = exchanger;
		}
		/* 
     * (non-Javadoc) 
     * 
     * @see java.lang.Thread#run() 
     */
		@Override 
		    public void run() {
			List products = new ArrayList();
			for (int i = 0; i < 10; i++) {
				products.add("consumed " + i);
			}
			try {
				List results = exchanger.exchange(products);
				System.out.println("got products from produces");
				for (String s : results) {
					System.out.println(s);
				}
			}
			catch (InterruptedException e) {
				// TODO Auto-generated catch block 
				e.printStackTrace();
			}
		}
	}
}