Barrier

在分布式系统中，可以使用栅栏，对多个节点上的任务进行阻塞等待；直到满足某个定制的条件，所有的节点才可以继续执行下一步任务。

1. 关键 API

org.apache.curator.framework.recipes.barriers.DistributedBarrier

2. 机制说明

控制多节点上的多任务执行步进。

类似java中的java.util.concurrent.CyclicBarrier分布式实现。

3. 用法

3.1 创建

public DistributedBarrier(CuratorFramework client,String barrierPath)

3.2 使用

3.2.1 等待栅栏

public void waitOnBarrier()

3.2.2 设置/移除栅栏

setBarrier();
removeBarrier();

4. 错误处理

DistributedBarrier实例会监听链接丢失。在waitOnBarrier()时，如果发生丢失时，则会抛出异常。

5. 源码分析

5.1 类定义

public class DistributedBarrier {}

没有继承父类，也没有实现任何接口

5.2 成员变量

public class DistributedBarrier
{private final CuratorFramework client;private final String barrierPath;private final Watcher watcher = new Watcher(){@Overridepublic void process(WatchedEvent event){notifyFromWatcher();}};
}

• client
• barrierPath
  • 用作栅栏的zk节点path
• watcher
  • 用作zk链接的监听器

5.3 构造器

public DistributedBarrier(CuratorFramework client, String barrierPath)
{this.client = client;this.barrierPath = PathUtils.validatePath(barrierPath);
}

简单赋值

5.3 设置栅栏

public synchronized void setBarrier() throws Exception
{try{client.create().creatingParentContainersIfNeeded().forPath(barrierPath);}catch ( KeeperException.NodeExistsException ignore ){// ignore}
}

• synchronized同步控制

可见，设置栅栏的过程，就是创建barrierPath节点（普通节点）

5.4 等待

public synchronized void waitOnBarrier() throws Exception
{waitOnBarrier(-1, null);
}public synchronized boolean waitOnBarrier(long maxWait, TimeUnit unit) throws Exception
{long            startMs = System.currentTimeMillis();boolean         hasMaxWait = (unit != null);long            maxWaitMs = hasMaxWait ? TimeUnit.MILLISECONDS.convert(maxWait, unit) : Long.MAX_VALUE;boolean         result;for(;;){result = (client.checkExists().usingWatcher(watcher).forPath(barrierPath) == null);if ( result ){break;}if ( hasMaxWait ){long        elapsed = System.currentTimeMillis() - startMs;long        thisWaitMs = maxWaitMs - elapsed;if ( thisWaitMs <= 0 ){break;}wait(thisWaitMs);}else{wait();}}return result;
}

• 使用synchronized同步控制
• 不断检查barrierPath栅栏节点是否存在
  • 如果栅栏不存在了，则栅栏放开了，返回true
  • 如果栅栏还在，则进行等待

5.5 移除栅栏

public synchronized void removeBarrier() throws Exception
{try{client.delete().forPath(barrierPath);}catch ( KeeperException.NoNodeException ignore ){// ignore}
}

直接删除栅栏节点

6. 测试

对于栅栏的示例，最先想到的就是赛马游戏的场景。所以，这里用DistributedBarrier来实现一个赛马游戏。

package com.roc.curator.demo.barriersimport 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
import org.junit.Test
import java.util.*
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.ThreadFactory
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicReference/*** Created by roc on 2017/6/1.*/
class HorseRaceLampTest {val PATH: String = "/test/barrier/horse"// 由于栅栏是监听链接，所以对于一个链接上的多个栅栏监听，会存在问题// 这里为每一个任务创建一个链接fun connect(): CuratorFramework {val client: CuratorFramework = CuratorFrameworkFactory.builder().connectString("0.0.0.0:8888").connectionTimeoutMs(5000).retryPolicy(ExponentialBackoffRetry(1000, 10)).sessionTimeoutMs(3000).build()client.start()return client}@Test fun runTest() {val name: String = "HORSE-"val count: AtomicInteger = AtomicInteger()val threadFactory: ThreadFactory = ThreadFactory { r ->val t: Thread = Thread(r, name + count.incrementAndGet())t}var champion: AtomicReference<String> = AtomicReference()champion.set("")val executorService: ExecutorService = Executors.newFixedThreadPool(5, threadFactory)//使用一个计步器来判定每一匹马是否完成当轮比赛val countStep: AtomicInteger = AtomicInteger()var round: Int = 0var i: Int = 0val target: Int = 100while (i < 5) {i++executorService.execute(Runnable {var sumSetp: Int = 0run {val client: CuratorFramework = connect()val barrier: DistributedBarrier = DistributedBarrier(client, PATH)//设置栅栏barrier.setBarrier()println("${Thread.currentThread()} 准备完毕 ${Date()}")//通知准备完成countStep.incrementAndGet()//等待开赛barrier.waitOnBarrier()//需要对线程是否中断进行判断while (!Thread.interrupted() || sumSetp < target) {var setp: Int = (Math.random() * 20).toInt()setp = maxOf(setp, 1)setp = minOf(setp, target - sumSetp)sumSetp += setpprintln("${Thread.currentThread()} 第 $round 轮，跑了 $setp 步，合计 $sumSetp / $target ")//如果已经到达终点，自行标记冠军//示例而已，如果要严谨一点的话，这个应该交给裁判来判断if (sumSetp >= target) {champion.set(Thread.currentThread().name)}//设立栅栏barrier.setBarrier()//通知完成此轮比赛countStep.incrementAndGet()//等待下一轮barrier.waitOnBarrier()}}})}val client: CuratorFramework = connect()val barrier: DistributedBarrier = DistributedBarrier(client, PATH)var pStep: Int = countStep.get()while (champion.get().isBlank()) {if (countStep.get() > pStep && countStep.get() % 5 == 0) {round++println("------------------------------------------------")println("裁判： 开始第 $round 轮 ${Date()}")println("------------------------------------------------")pStep = countStep.get()barrier.removeBarrier()}TimeUnit.MILLISECONDS.sleep(500)}println("裁判： 冠军是 $champion")executorService.shutdown()barrier.removeBarrier()}
}

输出：

Thread[HORSE-2,5,main] 准备完毕 Thu Jun 01 20:18:20 CST 2017
Thread[HORSE-1,5,main] 准备完毕 Thu Jun 01 20:18:20 CST 2017
Thread[HORSE-3,5,main] 准备完毕 Thu Jun 01 20:18:20 CST 2017
Thread[HORSE-5,5,main] 准备完毕 Thu Jun 01 20:18:20 CST 2017
Thread[HORSE-4,5,main] 准备完毕 Thu Jun 01 20:18:20 CST 2017
------------------------------------------------
裁判： 开始第 1 轮 Thu Jun 01 20:18:20 CST 2017
------------------------------------------------
Thread[HORSE-4,5,main] 第 1 轮，跑了 17 步，合计 17 / 100
Thread[HORSE-3,5,main] 第 1 轮，跑了 13 步，合计 13 / 100
Thread[HORSE-2,5,main] 第 1 轮，跑了 8 步，合计 8 / 100
Thread[HORSE-5,5,main] 第 1 轮，跑了 14 步，合计 14 / 100
Thread[HORSE-1,5,main] 第 1 轮，跑了 18 步，合计 18 / 100
------------------------------------------------
裁判： 开始第 2 轮 Thu Jun 01 20:18:21 CST 2017
------------------------------------------------
Thread[HORSE-3,5,main] 第 2 轮，跑了 11 步，合计 24 / 100
Thread[HORSE-4,5,main] 第 2 轮，跑了 1 步，合计 18 / 100
Thread[HORSE-5,5,main] 第 2 轮，跑了 14 步，合计 28 / 100
Thread[HORSE-2,5,main] 第 2 轮，跑了 1 步，合计 9 / 100
Thread[HORSE-1,5,main] 第 2 轮，跑了 19 步，合计 37 / 100
------------------------------------------------
裁判： 开始第 3 轮 Thu Jun 01 20:18:21 CST 2017
------------------------------------------------
Thread[HORSE-3,5,main] 第 3 轮，跑了 10 步，合计 34 / 100
Thread[HORSE-4,5,main] 第 3 轮，跑了 13 步，合计 31 / 100
Thread[HORSE-1,5,main] 第 3 轮，跑了 10 步，合计 47 / 100
Thread[HORSE-2,5,main] 第 3 轮，跑了 8 步，合计 17 / 100
Thread[HORSE-5,5,main] 第 3 轮，跑了 18 步，合计 46 / 100
------------------------------------------------
裁判： 开始第 4 轮 Thu Jun 01 20:18:22 CST 2017
------------------------------------------------
Thread[HORSE-2,5,main] 第 4 轮，跑了 3 步，合计 20 / 100
Thread[HORSE-4,5,main] 第 4 轮，跑了 9 步，合计 40 / 100
Thread[HORSE-3,5,main] 第 4 轮，跑了 16 步，合计 50 / 100
Thread[HORSE-1,5,main] 第 4 轮，跑了 14 步，合计 61 / 100
Thread[HORSE-5,5,main] 第 4 轮，跑了 12 步，合计 58 / 100
------------------------------------------------
裁判： 开始第 5 轮 Thu Jun 01 20:18:22 CST 2017
------------------------------------------------
Thread[HORSE-3,5,main] 第 5 轮，跑了 6 步，合计 56 / 100
Thread[HORSE-1,5,main] 第 5 轮，跑了 17 步，合计 78 / 100
Thread[HORSE-2,5,main] 第 5 轮，跑了 12 步，合计 32 / 100
Thread[HORSE-4,5,main] 第 5 轮，跑了 10 步，合计 50 / 100
Thread[HORSE-5,5,main] 第 5 轮，跑了 3 步，合计 61 / 100
------------------------------------------------
裁判： 开始第 6 轮 Thu Jun 01 20:18:23 CST 2017
------------------------------------------------
Thread[HORSE-4,5,main] 第 6 轮，跑了 6 步，合计 56 / 100
Thread[HORSE-2,5,main] 第 6 轮，跑了 3 步，合计 35 / 100
Thread[HORSE-1,5,main] 第 6 轮，跑了 5 步，合计 83 / 100
Thread[HORSE-5,5,main] 第 6 轮，跑了 7 步，合计 68 / 100
Thread[HORSE-3,5,main] 第 6 轮，跑了 1 步，合计 57 / 100
------------------------------------------------
裁判： 开始第 7 轮 Thu Jun 01 20:18:23 CST 2017
------------------------------------------------
Thread[HORSE-3,5,main] 第 7 轮，跑了 14 步，合计 71 / 100
Thread[HORSE-2,5,main] 第 7 轮，跑了 3 步，合计 38 / 100
Thread[HORSE-1,5,main] 第 7 轮，跑了 14 步，合计 97 / 100
Thread[HORSE-4,5,main] 第 7 轮，跑了 1 步，合计 57 / 100
Thread[HORSE-5,5,main] 第 7 轮，跑了 11 步，合计 79 / 100
------------------------------------------------
裁判： 开始第 8 轮 Thu Jun 01 20:18:24 CST 2017
------------------------------------------------
Thread[HORSE-4,5,main] 第 8 轮，跑了 15 步，合计 72 / 100
Thread[HORSE-2,5,main] 第 8 轮，跑了 8 步，合计 46 / 100
Thread[HORSE-3,5,main] 第 8 轮，跑了 10 步，合计 81 / 100
Thread[HORSE-5,5,main] 第 8 轮，跑了 9 步，合计 88 / 100
Thread[HORSE-1,5,main] 第 8 轮，跑了 3 步，合计 100 / 100
裁判： 冠军是 HORSE-1

zk节点：

get /test/barrier/horse
192.168.60.165
cZxid = 0x1e991
ctime = Thu Jun 01 20:18:23 CST 2017
mZxid = 0x1e991
mtime = Thu Jun 01 20:18:23 CST 2017
pZxid = 0x1e991
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 14
numChildren = 0