[Curator] Barrier 的使用与分析
[Curator] Barrier 的使用与分析
秋雨霏霏 发表于7个月前
[Curator] Barrier 的使用与分析
  • 发表于 7个月前
  • 阅读 98
  • 收藏 0
  • 点赞 1
  • 评论 0

腾讯云 技术升级10大核心产品年终让利>>>   

摘要: Zookeeper Curator :Barrier 栅栏

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()
    {
        @Override
        public 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.barriers

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
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 = 0
        var i: Int = 0
        val target: Int = 100
        while (i < 5) {
            i++
            executorService.execute(Runnable {
                var sumSetp: Int = 0
                run {
                    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 += setp
                        println("${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

共有 人打赏支持
粉丝 115
博文 75
码字总数 138343
×
秋雨霏霏
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: