Java线程同步方法

04 Sep 2013

CountDownLatch

CountDownLatch主要用来等待一系列线程都完成后，再继续下一步的执行，类似于C++中的join。

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CountDownLatchExample {

	public static void main(String[] args) {
		ExecutorService pool = Executors.newCachedThreadPool();
		final CountDownLatch doneLatch = new CountDownLatch(10);

		for(int i = 0; i < 10; i++) {
			Runnable runnable = new Runnable() {

				@Override
				public void run() {
					try {
						Thread.sleep((long)(Math.random() * 1000));
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("Thread: " + Thread.currentThread().toString() + " done");
					doneLatch.countDown();
				}
			};
			pool.execute(runnable);
		}
		try {
			doneLatch.await();
			System.out.println("All Thread Done!");
		} catch(InterruptedException e) {
			e.printStackTrace();
		}
		pool.shutdown();
	}
}

CyclicBarrier

CyclicBarrier跟CountDownLatch很相似，它主要用来等待一系列线程都运行到某一步后，再同时继续往下运行。相当于是一个栅栏，只有当所有的线程都到达这个栅栏的时候，栅栏才会打开，各个线程继续往下执行，所以CyclicBarrier有栅栏函数的作用。 CyclicBarrier的构造函数除了指定需要等待的线程数目外，还可以指定一个Runnable对象，由最后一个到达的线程调用。

public CyclicBarrier(int parties, Runnable barrierAction)

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

public class CyclicBarrierExample {
	
	public static void main(String[] args) {
		ExecutorService pool = Executors.newCachedThreadPool();
		final CyclicBarrier barrier = new CyclicBarrier(5);
		
		for(int i = 0; i < 5; i++) {
			Runnable runnable = new Runnable() {
				
				@Override
				public void run() {
					try {
						Thread.sleep((long)(Math.random() * 1000));
						System.out.println("Thread: " + Thread.currentThread().toString() + "step 1 done. "
								+ (barrier.getNumberWaiting() +1) + " thread finished!");
						
						barrier.await();
						System.out.println("Thread: " + Thread.currentThread().toString() + "start step 2 ...");
						
					} catch(InterruptedException e) {
						e.printStackTrace();
					} catch(BrokenBarrierException e) {
						e.printStackTrace();
					}
				}
			};
			pool.execute(runnable);
		}
		pool.shutdown();
	}
}

Semaphore

信号量，这个跟C++中的semaphore用法是一致的，在构造函数中指定资源数目，每个线程在访问临界区的时候都需要首先使用acquire()获得一个资源，完成后使用release()释放资源。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class SemaphoreExample {
	
	public static void main(String[] args) {
		ExecutorService pool = Executors.newCachedThreadPool();
		final Semaphore sem = new Semaphore(3);
		
		for(int i = 0; i < 10; i++) {
			Runnable runnable = new Runnable() {
				
				@Override
				public void run() {
					try {
						sem.acquire();
						System.out.println("Thread: " + Thread.currentThread().toString() + "running ...");
						Thread.sleep((long)(Math.random() * 1000));
						System.out.println("Thread: " + Thread.currentThread().toString() + "done");
						sem.release();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			};
			pool.execute(runnable);
		}
		pool.shutdown();
	}
}