java并發(fā)編程之同步器代碼示例

package hb.java.thread;
import java.util.concurrent.CountDownLatch;
/**
 * 
 * @author hb
 *     CountDownLatch最重要的方法是countDown()和await()，前者主要是倒數(shù)一次，后者是等待倒數(shù)到0，如果沒有到達(dá)0
 *     ，就只有阻塞等待了。 *JAVA同步器之
 *     CountDownLatch（不能循環(huán)使用，如果需要循環(huán)使用可以考慮使用CyclicBarrier） 兩種比較常規(guī)用法： 1：new
 *     CountDownLatch(1);所有的線程在開始工作前需要做一些準(zhǔn)備工作，當(dāng)所有的線程都準(zhǔn)備到位后再統(tǒng)一執(zhí)行時(shí)有用 2：new
 *     CountDownLatch(THREAD_COUNT);當(dāng)所有的線程都執(zhí)行完畢后，等待這些線程的其他線程才開始繼續(xù)執(zhí)行時(shí)有用
 */
public class CountDownLatchTest {
	private static final int THREAD_COUNT = 10;
	// 在調(diào)用startSingal.countDown()之前調(diào)用了startSingal.await()的線程一律等待，直到startSingal.countDown()的調(diào)用
	private static final CountDownLatch startSingal = new CountDownLatch(1);
	// 在finishedSingal的初始化記數(shù)量通過調(diào)用finishedSingal.countDown()減少為0時(shí)調(diào)用了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();
		// 所有的線程被喚醒，同時(shí)開始工作.countDown 方法的線程等到計(jì)數(shù)到達(dá)零時(shí)才繼續(xù)
		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(能夠循環(huán)使用，當(dāng)計(jì)數(shù)器增加到Barrier的初始化計(jì)數(shù)器之后馬上會(huì)被置為0為下一次循環(huán)使用做準(zhǔn)備) 
 * Barrier能夠?yàn)橹付ǖ囊粋€(gè)或多個(gè)（一般為多個(gè)）線程設(shè)置一道屏障，只有當(dāng)所有的線程都到達(dá)該屏障后才能一起沖過該屏障繼續(xù)其他任務(wù) 一般可以new 
 * CyclicBarrier(ThreadCount)來進(jìn)行初始化，也可以new 
 * CyclicBarrier(ThreadCount,RunableAction)當(dāng)初始化數(shù)量的線程都調(diào)用 
 * 了await()方法后觸發(fā)RunableAction線程，也可以通過初始化一個(gè)new 
 * CyclicBarrier(ThreadCount+1)的Barrier在前置線程未執(zhí)行完成時(shí)一直阻塞一個(gè)或多個(gè) 
 * 后續(xù)線程，這一點(diǎn)類似于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();
		// --------------開始準(zhǔn)備循環(huán)使用-------------- 
		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<List<String>> exchanger = new Exchanger<List<String>>();
	public static void main(String[] args) {
		new Producer("Producer", exchanger).start();
		new Consumer("Consumer", exchanger).start();
	}
	static class Producer extends Thread {
		private Exchanger<List<String>> exchanger;
		/** 
     *  
     */
		public Producer(String threadName, Exchanger<List<String>> exchanger) {
			super(threadName);
			this.exchanger = exchanger;
		}
		/* 
     * (non-Javadoc) 
     * 
     * @see java.lang.Thread#run() 
     */
		@Override 
		    public void run() {
			List<String> products = new ArrayList<String>();
			for (int i = 0; i < 10; i++) {
				products.add("product " + i);
			}
			try {
				List<String> 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<List<String>> exchanger;
		/** 
     *  
     */
		public Consumer(String threadName, Exchanger<List<String>> exchanger) {
			super(threadName);
			this.exchanger = exchanger;
		}
		/* 
     * (non-Javadoc) 
     * 
     * @see java.lang.Thread#run() 
     */
		@Override 
		    public void run() {
			List<String> products = new ArrayList<String>();
			for (int i = 0; i < 10; i++) {
				products.add("consumed " + i);
			}
			try {
				List<String> 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();
			}
		}
	}
}