Curator典型使用场景之分布式Barrier。

た入场券 2024-02-18 17:09 18阅读 0赞

Barrier是一种用来控制多线程之间同步的经典方式，在JDK中也自带了CyclicBarrier实现。下面通过模拟一个赛跑比赛来演示CyclicBarrier的用法。

## 使用CyclicBarrier模拟一个赛跑比赛 ##

public class Recipes\_CyclicBarrier \{

> public static CyclicBarrier barrier = new CyclicBarrier(3);
> 
> public static void main(String\[\] args) throws IOException, InterruptedException \{
> 
> > ExecutorService executor = Executors.newFixedThreadPool(3);
> > 
> > executor.submit(new Thread(new Runner("1号选手")));
> > 
> > executor.submit(new Thread(new Runner("2号选手")));
> > 
> > executor.submit(new Thread(new Runner("3号选手")));
> > 
> > executor.shutdown();  
> > 
> 
> \}  
>

class Runner implements Runnable \{

> private String name;
> 
> public Runner(String name) \{
> 
> > this.name = name;  
> > 
> 
> \}
> 
> public void run() \{
> 
> > System.out.println(name + "准备好了.");
> > 
> > try \{
> > 
> > > Recipes\_CyclicBarrier.barrier.await();  
> > > 
> > 
> > \} catch (Exception e) \{\}
> > 
> > System.out.println(name + " 起跑！");  
> > 
> 
> \}  
>

上面就是一个使用JDK自带的CyclicBarrier实现的赛跑比赛程序，可以看到多线程在并发情况下，都会准确的等待所有县城都处于就绪状态后才开始同时执行其他业务逻辑。如果是在同一个JVM中的话，使用CyclicBarrier完全可以解决诸如此类的多线程同步问题。但是，如果在分布式环境中又该如何解决呢？Curator中提供DistributedBarrier就是用来实现分布式Barrier的。

## 使用Curator实现分布式Barrier ##

// 使用Curator实现分布式Barrier

public class Recipes\_Barrier \{

> static String barrier\_path = "/curator\_recipes\_barrier\_path";
> 
> static DistributedBarrier barrier;
> 
> public static void main(String\[\] args) throws Exception \{
> 
> > for(int i = 0; i <5; i ++) \{
> > 
> > > new Thread(new Runnable() \{
> > > 
> > > > public void run() \{
> > > > 
> > > > > try \{
> > > > > 
> > > > > > CuratorFramework client = CuratorFrameworkFactory.builder().connectString("domain1.book.zookeeper:2181").retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
> > > > > > 
> > > > > > client.start();
> > > > > > 
> > > > > > barrier = new DistributedBarrier(client, barrier\_path);
> > > > > > 
> > > > > > System.out.println(Thread.currentThread().getName() + "号barrier设置");
> > > > > > 
> > > > > > barrier.setBarrier();
> > > > > > 
> > > > > > barrier.waitOnBarrier();
> > > > > > 
> > > > > > System.err.println("启动...");
> > > > > 
> > > > > \} catch (Exception e) \{\}  
> > > > > 
> > > > 
> > > > \}  
> > > > 
> > > 
> > > \}).start();
> > > 
> > > Thread.sleep(2000);
> > > 
> > > barrier.removeBarrier();  
> > > 
> > 
> > \}  
> > 
> 
> \}  
>

运行程序，输出结果如下：

> ![20171213140122381][]

在上面这个实例程序中，我们模拟了5个线程，通过调用DistributedBarrier.setBarrier()方法来完成Barrier的设置，并通过调用DistributedBarrier.waitOnBarrier()方法来等待Barrier的释放。然后在主线程中，通过调用DistributedBarrier.removeBarrier()方法来释放Barrier，同时触发所有等待该Barrier的5个线程同时进行各自的逻辑。

和上面这种由主线程来触发Barrier释放不同的是，Curator还提供了另一种线程自发触发Barrier释放的模式，使用方法见下面示例。

## 使用Curator实现另一种线程自发触发Barrier释放的模式 ##

public class Recipes\_Barrier2 \{

> static String barrier\_path = "/curator\_recipes\_barrier\_path";
> 
> public static void main(String\[\] args) throws Exception \{
> 
> > for(int i = 0; i <5; i ++) \{
> > 
> > > new Thread(new Runnable() \{
> > > 
> > > > public void run() \{
> > > > 
> > > > > try \{
> > > > > 
> > > > > > CuratorFramework client = CuratorFrameworkFactory.builder().connectString("domain1.book.zookeeper:2181").retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
> > > > > > 
> > > > > > client.start();
> > > > > > 
> > > > > > DistributedBarrier barrier= new DistributedBarrier(client, barrier\_path, 5);
> > > > > > 
> > > > > > Thread.sleep(Math.round(Math.random() \* 3000));  
> > > > > > 
> > > > > > 
> > > > > > System.out.println(Thread.currentThread().getName() + "号barrier设置");
> > > > > > 
> > > > > > barrier.enter();
> > > > > > 
> > > > > > System.out.println("启动...");
> > > > > > 
> > > > > > Thread.sleep(Math.round(Math.random() \* 3000));
> > > > > > 
> > > > > > barrier.leave();
> > > > > > 
> > > > > > System.out.println("退出...");  
> > > > > > 
> > > > > 
> > > > > \} catch (Exception e) \{\}  
> > > > > 
> > > > 
> > > > \}  
> > > > 
> > > 
> > > \}).start();
> > > 
> > > Thread.sleep(2000);
> > > 
> > > barrier.removeBarrier();  
> > > 
> > 
> > \}  
> > 
> 
> \}  
>

运行程序，输出结果如下。

> ![20171213140754174][]

上面这个示例程序就是一个和JDK自带的CyclicBarrier非常类似的实现了，他们都指定了进入Barrier的成员数阈值，例如上面示例程序中的“5”.每个Barrier的参与者都会在调用DistributedDoubleBarrier.enter()方法之后进行等待，此时处于准备进入状态。一旦准备进入Barrier的成员数达到5个后，所有的成员会被同时触发进入。之后调用DistributedDoubleBarrier.leave()方法则会再次等待，此时处于退出状态。一旦准备退出Barrier的成员数达到5个后，所有的成员同样会被同时触发退出。因此，使用Curator的DistributedDoubleBarrier能够很好的实现一个分布式Barrier，并控制其同时进入和退出。

[20171213140122381]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/603136834a884fd9943940e286681a8e.png
[20171213140754174]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/8580f82ffb9749b89a1f595f759f5d6b.png