ZooKeeper ： Curator架構之分布式鎖InterProcessSemaphoreMutex

InterProcessSemaphoreMutex

​

​InterProcessSemaphoreMutex​

​類的源碼注釋：

A NON re-entrant mutex that works across JVMs. Uses Zookeeper to hold the lock. All processes in all JVMs that use the same lock path will achieve an inter-process critical section.

跨​

​JVM​

​​工作的不可重入互斥鎖。使用​

​Zookeeper​

​​來持有鎖。所有​

​JVM​

​中使用相同鎖路徑的所有程序都将實作程序間臨界區。

與​

​InterProcessMutex​

​​類似，也是公平的，每個使用者都将按照請求的順序獲得互斥鎖（從​

​ZK​

​的角度來看），但此鎖不可重入。

• ​​ZooKeeper ： Curator架構之分布式鎖InterProcessMutex​​

​

​InterProcessSemaphoreMutex​

​類的可用方法比較少：

測試

​

​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​

​需要的一些配置資訊）：

package com.kaven.zookeeper;

import org.apache.curator.RetryPolicy;
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);
}      

​

​InterProcessSemaphoreMutexRunnable​

​​類（實作了​

​Runnable​

​接口，模拟分布式節點擷取分布式鎖）：

package com.kaven.zookeeper;

import lombok.SneakyThrows;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;

import java.util.Random;

public class InterProcessSemaphoreMutexRunnable implements Runnable{
    @SneakyThrows
    @Override
    public void run() {
        // 使用不同的CuratorFramework執行個體，表示不同的分布式節點
        CuratorFramework curator = getCuratorFramework();
        curator.start();
        assert curator.getState().equals(CuratorFrameworkState.STARTED);

        // 模拟随機加入的分布式節點
        int randomSleep = new Random().nextInt(1000);
        Thread.sleep(randomSleep);

        // 分布式鎖的鎖路徑
        String path = "/kaven";

        // 建立InterProcessSemaphoreMutex執行個體，用于擷取分布式鎖
        InterProcessSemaphoreMutex mutex =
                new InterProcessSemaphoreMutex(curator, path);
                
        // 阻塞，直到擷取分布式鎖
        mutex.acquire();
        if(mutex.isAcquiredInThisProcess()) {
            System.out.println(Thread.currentThread().getName() + " 持有分布式鎖");
            Thread.sleep(5000);
            System.out.println(Thread.currentThread().getName() + " 釋放分布式鎖");
            mutex.release();
        }
        else {
            throw new RuntimeException("擷取分布式鎖時被中斷");
        }
    }

    private CuratorFramework getCuratorFramework() {

        // 建立CuratorFramework執行個體
        return CuratorFrameworkFactory.builder()
                .connectString(CuratorFrameworkProperties.CONNECT_ADDRESS)
                .retryPolicy(CuratorFrameworkProperties.RETRY_POLICY)
                .connectionTimeoutMs(CuratorFrameworkProperties.CONNECTION_TIMEOUT_MS)
                .sessionTimeoutMs(CuratorFrameworkProperties.SESSION_TIMEOUT_MS)
                .namespace(CuratorFrameworkProperties.NAMESPACE)
                .build();
    }
}      

啟動類：

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 InterProcessSemaphoreMutexRunnable());
        }
        Thread.sleep(10000000);
    }
}      

模拟​

​5​

​個分布式節點擷取分布式鎖，輸出如下所示：

pool-1-thread-4 持有分布式鎖
pool-1-thread-4 釋放分布式鎖
pool-1-thread-5 持有分布式鎖
pool-1-thread-5 釋放分布式鎖
pool-1-thread-3 持有分布式鎖
pool-1-thread-3 釋放分布式鎖
pool-1-thread-1 持有分布式鎖
pool-1-thread-1 釋放分布式鎖
pool-1-thread-2 持有分布式鎖
pool-1-thread-2 釋放分布式鎖      

​

​InterProcessSemaphoreMutex​

​​提供了帶有逾時機制的擷取鎖的方法（​

​acquire(long time, TimeUnit unit)​

​）。

// 阻塞，直到擷取分布式鎖或給定時間已到
        mutex.acquire(10, TimeUnit.SECONDS);
        if(mutex.isAcquiredInThisProcess()) {
            System.out.println(Thread.currentThread().getName() + " 持有分布式鎖");
            Thread.sleep(20000);
            System.out.println(Thread.currentThread().getName() + " 釋放分布式鎖");
            mutex.release();
        }
        else {
            System.out.println(Thread.currentThread().getName() + " 擷取分布式鎖逾時");
        }      

pool-1-thread-4 持有分布式鎖
pool-1-thread-1 擷取分布式鎖逾時
pool-1-thread-5 擷取分布式鎖逾時
pool-1-thread-3 擷取分布式鎖逾時
pool-1-thread-2 擷取分布式鎖逾時
pool-1-thread-4 釋放分布式鎖