Curator Barrier

分布式Barrier，直接看代码好了：

package curator.barrier;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class BarrierDemo {

	public static void main(String[] args) throws Exception {
		final List<DistributedBarrier> barriers = new ArrayList<DistributedBarrier>(3);
		ExecutorService service = Executors.newCachedThreadPool();

		for (int i = 0; i < 3; i++) {
			Runnable run = new Runnable() {
				public void run() {
					try {
						RetryPolicy retryPolicy = new ExponentialBackoffRetry(
								1000, 3);
						final CuratorFramework client = CuratorFrameworkFactory
								.builder().connectString("127.0.0.1:2181")
								.sessionTimeoutMs(5000)
								.connectionTimeoutMs(10000)
								.retryPolicy(retryPolicy).namespace("text")
								.build();
						client.start();

						DistributedBarrier barrier = new DistributedBarrier(client, "/barrier");
						barriers.add(barrier);
						barrier.setBarrier(); // 设置barrier
						System.out.println(" : setBarrier...");
						barrier.waitOnBarrier();// 等待其他进程移除barrier,此后所有的waitOnBarrier进程都将解除阻塞.
						System.err.println(" : run ing...");
					} catch (Exception e) {
					}
				}
			};
			service.execute(run);
		}
		Thread.currentThread().sleep(5000);
		for (DistributedBarrier b : barriers) {
			b.removeBarrier();
		}
		System.in.read();
	}

}