JAVA 并发编程-同步工具类

 

CountDownLatch 闭锁

闭锁是一种同步工具类，可以延迟线程的进度知道其到达终止状态。闭锁的作用相当于一扇门（await）：在闭锁到达结束状态之前，这扇门一直是关闭的，不允许任何线程通过，当到达结束状态时(所有线程均到达countDown)，这扇门会打开并且允许所有的线程通过。而且，当门打开了，就永远保持打开状态。
作用：

1、确保某些活动直到其他活动都完成后才继续执行。
2、确保某个服务在其依赖的所有其他服务都已经启动之后才启动。
3、等待直到某个操作的所有参与者都就绪再继续执行。

public class CountDownLatchTest {
	public static void main(String[] args) {
		CountDownLatch latch = new CountDownLatch(3);
		new Thread(new Father(latch)).start();
		new Thread(new Mother(latch)).start();
		new Thread(new GrandmaAndLizzy(latch)).start();
		try {
			latch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("Start lauch!!");
	}

	public static class Father implements Runnable {
		CountDownLatch latch;

		public Father(CountDownLatch latch) {
			this.latch = latch;
		}

		@Override
		public void run() {
			System.out.println("Father ready!");
			latch.countDown();
		}
	}

	public static class Mother implements Runnable {
		CountDownLatch latch;

		public Mother(CountDownLatch latch) {
			this.latch = latch;
		}

		@Override
		public void run() {
			System.out.println("Mother ready!");
			latch.countDown();
		}

	}

	public static class GrandmaAndLizzy implements Runnable {
		CountDownLatch latch;

		public GrandmaAndLizzy(CountDownLatch latch) {
			this.latch = latch;
		}

		@Override
		public void run() {
			System.out.println("GrandmaAndLizzy ready!");
			latch.countDown();
		}
	}
}

 

FutureTask

FutureTask 实现了Future接口，表示一种抽象的可生成结果（可撤销、超时get）的计算。
FuturTask 表示的计算是通过Callable 来实现的，相当于一种可生成结果的Runnable，可以处于以下三种状态：等外运行、正在运行和运行完成。“运行完成”表示计算的所有可能结束方式，包括正常结束、由于取消而结束和由于异常而结束等。当FutureTask进入完成状态后，它会永远停止在这个状态上。
FutureTask.get()的行为取决于任务的状态，如果任务已经完成，那么get会立即返回结果，否则get将阻塞知道任务进入完成状态，然后返回结果或者抛出异常。FutureTask将计算结果从执行计算的线程传递到获取这个结果的线程，而且FutureTask的规范确保了这种传递过程能实现结果的安全发布。

FuturTask在Executor框架中表示异步任务。

public class MyFutureTaskDemo {
	public static void main(String[] args) {
		final FutureTask<String> task = new FutureTask<String>(
				new Callable<String>() {
					@Override
					public String call() throws Exception {
						System.out.println("let's go!");
						Thread.sleep(3000);
						return "Hello Beautiful World!";
					}
				});

		try {
			System.out.println("start get!");
			new Thread(task).start();
			String result = task.get();
			System.out.println("finish  get=" + result);
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}
	}
}

 

Semaphore 信号量

可以用来控制同时访问某个特定资源的操作数量，或者同时执行某个指定操作的数量。计数信号量还可以用来实现某种资源池，或者对容器施加边界。

    acquire 从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。获取一个许可（如果提供了一个）并立即返回，将可用的许可数减 1。

    release 释放一个许可，将其返回给信号量。释放一个许可，将可用的许可数增加 1。如果任意线程试图获取许可，则选中一个线程并将刚刚释放的许可给予它。然后针对线程安排目的启用（或再启用）该线程。

public class MySemaphoreDemo {
	private final Semaphore sem;

	public MySemaphoreDemo(int count) {
		sem = new Semaphore(count);
	}

	// 申请资源
	public void getResource() throws InterruptedException {
		sem.acquire();
		System.out.println("资源已下发");
	}

	// 释放资源
	public void releaseResource() {
		sem.release();
		System.out.println("资源已回收");
	}

	public static void main(String[] args) {
		final MySemaphoreDemo pool = new MySemaphoreDemo(3);
		for (int i = 0; i < 10; i++) {
			new Thread(new Runnable() {
				@Override
				public void run() {
					try {
						pool.getResource();
						Thread.sleep(5000);
						pool.releaseResource();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}).start();
		}
	}
}

 

CyclicBarrier 栅栏

 

CyclicBarrier 是一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。
CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。

CyclicBarrier 类似于闭锁，它能阻塞一组线程直到某个事件发生。栅栏与闭锁的关键区别在于，所有线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他先。

public class MyBarrierDemo {
	public static void main(String[] args) {
		CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {
			@Override
			public void run() {
				System.out.println("当前线程" + Thread.currentThread().getName());
			}
		});
		new Thread(new Father(barrier)).start();
		new Thread(new Mother(barrier)).start();
		new Thread(new GrandmaAndLizzy(barrier)).start();
	}

	public static class Father implements Runnable {
		CyclicBarrier barrier;

		public Father(CyclicBarrier barrier) {
			this.barrier = barrier;
		}

		@Override
		public void run() {
			System.out.println("Father ready!");
			try {
				barrier.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
			}
			System.out.println("Every arrive ,Start lauch!!");
		}
	}

	public static class Mother implements Runnable {
		CyclicBarrier barrier;

		public Mother(CyclicBarrier barrier) {
			this.barrier = barrier;
		}

		@Override
		public void run() {
			System.out.println("Mother ready!");
			try {
				barrier.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
			}
			System.out.println("Every arrive ,Start lauch!!");
		}

	}

	public static class GrandmaAndLizzy implements Runnable {
		CyclicBarrier barrier;

		public GrandmaAndLizzy(CyclicBarrier barrier) {
			this.barrier = barrier;
		}

		@Override
		public void run() {
			try {
				System.out.println("GrandmaAndLizzy ready!");
				Thread.sleep(3000);
				barrier.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
			}
			System.out.println("Every arrive ,Start lauch!!");
		}
	}
}

 

Exchanger 双边栅栏

Exchanger 是一种两方栅栏：各方栅栏在阻塞位置上交换数据（双方都达到阻塞位置时）。当两方执行不对称的操作时，会非常有用。当两个线程通过Exchanger交换对象时，这种交换就把这两个对象安全的发布给另一方。

public class ExchangeTest {

	public static void main(String[] args) {
		Exchanger<List<Integer>> exchanger = new Exchanger<List<Integer>>();
		new Consumer(exchanger).start();
		new Producer(exchanger).start();
	}

	static class Producer extends Thread {
		List<Integer> list = new ArrayList<Integer>();
		Exchanger<List<Integer>> exchanger = null;

		public Producer(Exchanger<List<Integer>> exchanger) {
			this.exchanger = exchanger;
		}

		@Override
		public void run() {
			Random random = new Random();
			for (int i = 0; i < 10; i++) {
				list.clear();
				list.add(random.nextInt(100));
				list.add(random.nextInt(100));
				list.add(random.nextInt(100));
				list.add(random.nextInt(100));
				list.add(random.nextInt(100));
				if (list.size() == 5) {
					System.out.println("Producer :before exchange:"
							+ list.get(0) + "," + list.get(1) + ","
							+ list.get(2) + "," + list.get(3) + ","
							+ list.get(4));
				} else {
					System.out.println("Producer :before exchange: is null");
				}

				try {
					list = exchanger.exchange(list);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				if (list.size() == 5) {
					System.out.println("Producer :after exchange:"
							+ list.get(0) + "," + list.get(1) + ","
							+ list.get(2) + "," + list.get(3) + ","
							+ list.get(4));
				} else {
					System.out.println("Producer :after exchange: is null");
				}

				System.out
						.println("==================Producer======================");
			}
		}
	}

	static class Consumer extends Thread {
		List<Integer> list = new ArrayList<Integer>();
		Exchanger<List<Integer>> exchanger = null;

		public Consumer(Exchanger<List<Integer>> exchanger) {
			this.exchanger = exchanger;
		}

		public void run() {
			for (int i = 0; i < 10; i++) {
				if (list.size() == 5) {
					System.out.println("Consumer :before exchange:"
							+ list.get(0) + "," + list.get(1) + ","
							+ list.get(2) + "," + list.get(3) + ","
							+ list.get(4));
				} else {
					System.out.println("Consumer :before exchange: is null");
				}
				try {
					list = exchanger.exchange(list);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				if (list.size() == 5) {
					System.out.println("Consumer :after exchange:"
							+ list.get(0) + "," + list.get(1) + ","
							+ list.get(2) + "," + list.get(3) + ","
							+ list.get(4));
				} else {
					System.out.println("Consumer :after exchange: is null");
				}
				System.out
						.println("==================Consumer======================");

			}
		}
	}
}