DistributedDoubleBarrier

DistributedDoubleBarrier類的原始碼注釋：

Double barriers enable clients to synchronize the beginning and the end of a computation. When enough processes have joined the barrier, processes start their computation and leave the barrier once they have finished.

雙屏障使客戶端能夠同步計算的開始和結束，當足夠多的行程加入屏障時，行程開始計算并在完成后離開屏障，

類比單體應用的屏障CyclicBarrier：

Java并發編程：CountDownLatch、CyclicBarrier、Semaphore初使用

測驗

pom.xml：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.kaven</groupId>
    <artifactId>zookeeper</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>5.2.0</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.22</version>
        </dependency>
    </dependencies>
</project>

CuratorFrameworkProperties類（提供CuratorFramework需要的一些配置資訊，以及創建CuratorFramework實體的方法）：

package com.kaven.zookeeper;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorFrameworkProperties {
    // 連接地址
    public static final String CONNECT_ADDRESS = "192.168.1.3:9000";
    // 連接超時時間
    public static final int CONNECTION_TIMEOUT_MS = 40000;
    // Session超時時間
    public static final int SESSION_TIMEOUT_MS = 10000;
    // 命名空間
    public static final String NAMESPACE = "MyNamespace";
    // 重試策略
    public static final RetryPolicy RETRY_POLICY = new ExponentialBackoffRetry(1000, 3);

    public static CuratorFramework getCuratorFramework() {
        // 創建CuratorFramework實體
        CuratorFramework curator = CuratorFrameworkFactory.builder()
                .connectString(CuratorFrameworkProperties.CONNECT_ADDRESS)
                .retryPolicy(CuratorFrameworkProperties.RETRY_POLICY)
                .connectionTimeoutMs(CuratorFrameworkProperties.CONNECTION_TIMEOUT_MS)
                .sessionTimeoutMs(CuratorFrameworkProperties.SESSION_TIMEOUT_MS)
                .namespace(CuratorFrameworkProperties.NAMESPACE)
                .build();
        curator.start();
        assert curator.getState().equals(CuratorFrameworkState.STARTED);
        return curator;
    }
}

DistributedDoubleBarrierRunnable類（實作了Runnable介面，模擬分布式節點進入與離開分布式屏障）：

package com.kaven.zookeeper;

import lombok.SneakyThrows;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;

import java.util.Random;

public class DistributedDoubleBarrierRunnable implements Runnable{
    @SneakyThrows
    @Override
    public void run() {
        // 使用不同的CuratorFramework實體，表示不同的分布式節點
        CuratorFramework curator = CuratorFrameworkProperties.getCuratorFramework();

        // 模擬隨機加入的分布式節點
        int randomSleep = new Random().nextInt(20000);
        Thread.sleep(randomSleep);

        // 分布式屏障的路徑
        String barrierPath = "/kaven";

        // 創建DistributedDoubleBarrier實體，用于提供分布式屏障功能
        DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(curator, barrierPath, 5);

        System.out.println(Thread.currentThread().getName() + " 等待進入屏障");
        long start = System.currentTimeMillis();
        // 等待進入屏障
        barrier.enter();
        System.out.println(Thread.currentThread().getName() + " 等待了 "
                + (System.currentTimeMillis() - start) / 1000 + " s");
        System.out.println(Thread.currentThread().getName() + " 進入屏障");
        Thread.sleep(1000);
        // 等待離開屏障
        barrier.leave();
        System.out.println(Thread.currentThread().getName() + " 離開屏障");
    }
}

啟動類：

package com.kaven.zookeeper;

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

public class Application {
    private static final ExecutorService EXECUTOR_SERVICE = Executors.newCachedThreadPool();

    public static void main(String[] args) throws Exception {
        // 分布式節點處理業務
        for (int i = 0; i < 5; i++) {
            EXECUTOR_SERVICE.execute(new DistributedDoubleBarrierRunnable());
        }
    }
}

模擬5個分布式節點進入與離開分布式屏障，輸出如下所示：

pool-1-thread-3 等待進入屏障
pool-1-thread-5 等待進入屏障
pool-1-thread-2 等待進入屏障
pool-1-thread-1 等待進入屏障
pool-1-thread-4 等待進入屏障
pool-1-thread-4 等待了 0 s
pool-1-thread-4 進入屏障
pool-1-thread-3 等待了 9 s
pool-1-thread-3 進入屏障
pool-1-thread-2 等待了 6 s
pool-1-thread-2 進入屏障
pool-1-thread-1 等待了 5 s
pool-1-thread-1 進入屏障
pool-1-thread-5 等待了 8 s
pool-1-thread-5 進入屏障
pool-1-thread-1 離開屏障
pool-1-thread-3 離開屏障
pool-1-thread-5 離開屏障
pool-1-thread-2 離開屏障
pool-1-thread-4 離開屏障

創建DistributedDoubleBarrier實體，將memberQty屬性設定為5，當想進入屏障的節點數不少于memberQty的值時，這些節點就可以進入障礙，

DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(curator, barrierPath, 5);

memberQty屬性是一個閾值，而不是限制，閾值的約束是比較寬松的，接下來進行測驗，

public class Application {
    private static final ExecutorService EXECUTOR_SERVICE = Executors.newCachedThreadPool();

    public static void main(String[] args) throws Exception {
        // 分布式節點處理業務
        for (int i = 0; i < 7; i++) {
            EXECUTOR_SERVICE.execute(new DistributedDoubleBarrierRunnable());
        }
    }
}

修改應用的運行配置，
在這里插入圖片描述
允許多個應用實體一起運行，

運行兩個應用實體，

第一個應用實體的輸出如下所示：

pool-1-thread-5 等待進入屏障
pool-1-thread-6 等待進入屏障
pool-1-thread-2 等待進入屏障
pool-1-thread-4 等待進入屏障
pool-1-thread-1 等待進入屏障
pool-1-thread-1 等待了 0 s
pool-1-thread-1 進入屏障
pool-1-thread-6 等待了 7 s
pool-1-thread-6 進入屏障
pool-1-thread-5 等待了 8 s
pool-1-thread-5 進入屏障
pool-1-thread-4 等待了 3 s
pool-1-thread-4 進入屏障
pool-1-thread-2 等待了 5 s
pool-1-thread-2 進入屏障
pool-1-thread-3 等待進入屏障
pool-1-thread-3 等待了 0 s
pool-1-thread-3 進入屏障
pool-1-thread-7 等待進入屏障
pool-1-thread-7 等待了 0 s
pool-1-thread-7 進入屏障
pool-1-thread-5 離開屏障
pool-1-thread-4 離開屏障
pool-1-thread-7 離開屏障
pool-1-thread-6 離開屏障
pool-1-thread-3 離開屏障
pool-1-thread-2 離開屏障
pool-1-thread-1 離開屏障

第二個應用實體的輸出如下所示：

pool-1-thread-1 等待進入屏障
pool-1-thread-6 等待進入屏障
pool-1-thread-2 等待進入屏障
pool-1-thread-3 等待進入屏障
pool-1-thread-7 等待進入屏障
pool-1-thread-7 等待了 0 s
pool-1-thread-7 進入屏障
pool-1-thread-6 等待了 1 s
pool-1-thread-6 進入屏障
pool-1-thread-2 等待了 0 s
pool-1-thread-2 進入屏障
pool-1-thread-3 等待了 0 s
pool-1-thread-3 進入屏障
pool-1-thread-1 等待了 3 s
pool-1-thread-1 進入屏障
pool-1-thread-6 離開屏障
pool-1-thread-3 離開屏障
pool-1-thread-2 離開屏障
pool-1-thread-1 離開屏障
pool-1-thread-7 離開屏障
pool-1-thread-4 等待進入屏障
pool-1-thread-4 等待了 0 s
pool-1-thread-4 進入屏障
pool-1-thread-5 等待進入屏障
pool-1-thread-5 等待了 0 s
pool-1-thread-5 進入屏障
pool-1-thread-5 離開屏障
pool-1-thread-4 離開屏障

兩個應用實體一起運行，相當于14個節點進入與離開分布式屏障，在本次測驗中，它們都成功進入與離開分布式屏障，而14%5!=0，這是因為memberQty屬性是一個閾值，而不是限制，閾值的約束是比較寬松的，當想進入屏障的節點數不少于memberQty的值時，這些節點就可以進入障礙，而分布式環境是很復雜的，同時想要進入屏障的節點數可能有多個，只要這些節點還沒有離開障礙就會一直存在，因此14個節點是有可能全部成功進入與離開屏障，因此可以知道，某個節點想要進入屏障，只要此時節點數滿足閾值條件就可以進入，而不是限制每次只能進入memberQty個節點，而離開屏障沒有過多的約束，但離開屏障是一個集體操作，會讓該節點離開屏障，并且等待當前其他的節點全部離開屏障，才算離開屏障成功，博主以后會分析DistributedDoubleBarrier的原始碼，到時候再詳細介紹DistributedDoubleBarrier的實作原理，

因為memberQty屬性是一個閾值，而不是限制，因此多次測驗的結果可能會有差異，下面是第二次測驗的結果，

第一個應用實體的輸出如下所示：

pool-1-thread-2 等待進入屏障
pool-1-thread-7 等待進入屏障
pool-1-thread-5 等待進入屏障
pool-1-thread-5 等待了 0 s
pool-1-thread-5 進入屏障
pool-1-thread-2 等待了 8 s
pool-1-thread-2 進入屏障
pool-1-thread-7 等待了 2 s
pool-1-thread-7 進入屏障
pool-1-thread-3 等待進入屏障
pool-1-thread-3 等待了 0 s
pool-1-thread-3 進入屏障
pool-1-thread-2 離開屏障
pool-1-thread-5 離開屏障
pool-1-thread-7 離開屏障
pool-1-thread-3 離開屏障
pool-1-thread-4 等待進入屏障
pool-1-thread-6 等待進入屏障
pool-1-thread-6 等待了 0 s
pool-1-thread-6 進入屏障
pool-1-thread-4 等待了 3 s
pool-1-thread-4 進入屏障
pool-1-thread-4 離開屏障
pool-1-thread-6 離開屏障
pool-1-thread-1 等待進入屏障

第二個應用實體的輸出如下所示：

pool-1-thread-3 等待進入屏障
pool-1-thread-7 等待進入屏障
pool-1-thread-3 等待了 6 s
pool-1-thread-3 進入屏障
pool-1-thread-7 等待了 3 s
pool-1-thread-7 進入屏障
pool-1-thread-7 離開屏障
pool-1-thread-3 離開屏障
pool-1-thread-6 等待進入屏障
pool-1-thread-1 等待進入屏障
pool-1-thread-2 等待進入屏障
pool-1-thread-6 等待了 5 s
pool-1-thread-6 進入屏障
pool-1-thread-2 等待了 0 s
pool-1-thread-2 進入屏障
pool-1-thread-1 等待了 3 s
pool-1-thread-1 進入屏障
pool-1-thread-6 離開屏障
pool-1-thread-2 離開屏障
pool-1-thread-1 離開屏障
pool-1-thread-4 等待進入屏障
pool-1-thread-5 等待進入屏障

很顯然，這次測驗的14個節點并沒有都成功進入與離開分布式屏障，

DistributedDoubleBarrier類提供了帶有超時機制的進入與離開屏障的方法：enter(long maxWait, TimeUnit unit)、leave(long maxWait, TimeUnit unit)，這里就不進行演示了，比較簡單，

Curator框架的分布式屏障DistributedDoubleBarrier就介紹到這里，如果博主有說錯的地方或者大家有不同的見解，歡迎大家評論補充，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/415133.html

標籤：其他

上一篇：ZooKeeper ： Curator框架之共享計數器SharedCount

下一篇：JAVA基礎階段知識總結

ZooKeeper ： Curator框架之分布式屏障DistributedDoubleBarrier

DistributedDoubleBarrier

測驗