天天看點
傳回

ZooKeeper ： Curator架構之分布式鎖InterProcessReadWriteLock

InterProcessReadWriteLock

跨​

​JVM​

​​工作的可重入讀/寫互斥鎖，使用​

​Zookeeper​

​​來持有鎖，所有​

​JVM​

​​中使用相同鎖路徑的所有程序都将實作程序間臨界區。這個互斥鎖是公平的，每個使用者都會按照請求的順序獲得互斥鎖（從​

​ZK​

​的角度來看）。

讀寫鎖維護一對關聯的鎖，一個用于隻讀操作，一個用于寫操作。隻要沒有寫鎖，讀鎖可以被多個使用者同時持有，而寫鎖是獨占的。

讀寫鎖允許從寫鎖降級為讀鎖，方法是先擷取寫鎖，然後就可以擷取讀鎖。但是，無法從讀鎖更新到寫鎖。

測試

​

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

​

​InterProcessReadWriteLockRunnable​

​​類（實作了​

​Runnable​

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

package com.kaven.zookeeper;

import lombok.SneakyThrows;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;

import java.nio.charset.StandardCharsets;
import java.util.Random;

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

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

        // 建立InterProcessReadWriteLock執行個體，用于提供分布式鎖的功能
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(curator, basePath,
                "分布式讀寫鎖".getBytes(StandardCharsets.UTF_8));

        // 根據随機數來決定擷取寫鎖還是讀鎖
        Random random = new Random();
        if(random.nextInt(10000) > 5000) {
            // 擷取寫鎖
            readWriteLock.writeLock().acquire();
            System.out.println(Thread.currentThread().getName() + "擷取寫鎖");
            Thread.sleep(2000);
            System.out.println(Thread.currentThread().getName() + "釋放寫鎖");
            // 釋放寫鎖
            readWriteLock.writeLock().release();
        }
        else {
            // 擷取讀鎖
            readWriteLock.readLock().acquire();
            System.out.println(Thread.currentThread().getName() + "擷取讀鎖");
            Thread.sleep(2000);
            System.out.println(Thread.currentThread().getName() + "釋放讀鎖");
            // 釋放讀鎖
            readWriteLock.readLock().release();
        }
    }
}

啟動類：

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 < 15; i++) {
            EXECUTOR_SERVICE.execute(new InterProcessReadWriteLockRunnable());
        }
    }
}

模拟​

​15​

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

pool-1-thread-8擷取寫鎖
pool-1-thread-8釋放寫鎖
pool-1-thread-13擷取寫鎖
pool-1-thread-13釋放寫鎖
pool-1-thread-12擷取讀鎖
pool-1-thread-15擷取讀鎖
pool-1-thread-12釋放讀鎖
pool-1-thread-15釋放讀鎖
pool-1-thread-1擷取寫鎖
pool-1-thread-1釋放寫鎖
pool-1-thread-6擷取讀鎖
pool-1-thread-2擷取讀鎖
pool-1-thread-14擷取讀鎖
pool-1-thread-9擷取讀鎖
pool-1-thread-2釋放讀鎖
pool-1-thread-6釋放讀鎖
pool-1-thread-14釋放讀鎖
pool-1-thread-9釋放讀鎖
pool-1-thread-4擷取寫鎖
pool-1-thread-4釋放寫鎖
pool-1-thread-11擷取寫鎖
pool-1-thread-11釋放寫鎖
pool-1-thread-5擷取讀鎖
pool-1-thread-5釋放讀鎖
pool-1-thread-10擷取寫鎖
pool-1-thread-10釋放寫鎖
pool-1-thread-7擷取寫鎖
pool-1-thread-7釋放寫鎖
pool-1-thread-3擷取讀鎖
pool-1-thread-3釋放讀鎖

為了驗證輸出是否符合預期，可以通過​

​ZooKeeper​

​提供的用戶端擷取鎖路徑下的所有節點，如下圖所示：

ZooKeeper ： Curator架構之分布式鎖InterProcessReadWriteLock

排序後如下所示，和輸出是對應的（讀鎖可以被多個使用者同時持有，而寫鎖是獨占的）。

/MyNamespace/kaven/_c_cd949fbc-c779-46e6-a7d8-4c4c90f13f23-__WRIT__0000000000
/MyNamespace/kaven/_c_a4934270-e9c4-40c0-8b89-5fb369a7cfaa-__WRIT__0000000001
/MyNamespace/kaven/_c_5a03751b-7f45-4813-99f6-49671c473367-__READ__0000000002
/MyNamespace/kaven/_c_babe2eec-7259-484f-9f76-51a5fc45d607-__READ__0000000003
/MyNamespace/kaven/_c_405e6cc8-481b-4d9a-8cf1-a83ea69912dd-__WRIT__0000000004
/MyNamespace/kaven/_c_a870b175-c082-4661-9383-4a82a021e283-__READ__0000000005
/MyNamespace/kaven/_c_3fa721b5-1015-46e8-8da3-20b333b76ca4-__READ__0000000006
/MyNamespace/kaven/_c_03e0208c-fb25-4229-9534-4dc6c295d6a8-__READ__0000000007
/MyNamespace/kaven/_c_5428e1af-c9fc-4afd-92f7-76fbbb545dcc-__READ__0000000008
/MyNamespace/kaven/_c_395407f3-c402-40b4-abe2-5750f43ddd9d-__WRIT__0000000009
/MyNamespace/kaven/_c_c7a1a90c-8b76-47fc-8c9e-1dc15c43a8bd-__WRIT__0000000010
/MyNamespace/kaven/_c_800861ea-8da2-4ff5-b237-19d8b939393e-__READ__0000000011
/MyNamespace/kaven/_c_3d357ec6-5288-4719-ba3b-7f98eceb32d6-__WRIT__0000000012
/MyNamespace/kaven/_c_e8067da5-bf08-4876-8200-1746dbc2ce77-__WRIT__0000000013
/MyNamespace/kaven/_c_8d4e07a2-08b9-4c1d-91d4-3b0d76844b33-__READ__0000000014

鎖降級 

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

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

        // 建立InterProcessReadWriteLock執行個體，用于提供分布式鎖的功能
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(curator, basePath,
                "分布式讀寫鎖".getBytes(StandardCharsets.UTF_8));

        // 擷取寫鎖
        readWriteLock.writeLock().acquire();
        System.out.println(Thread.currentThread().getName() + "擷取寫鎖");
        Thread.sleep(2000);
        // 鎖降級
        readWriteLock.readLock().acquire();
        System.out.println(Thread.currentThread().getName() + "擷取讀鎖，鎖降級成功");
        Thread.sleep(2000);
        // 釋放讀鎖
        System.out.println(Thread.currentThread().getName() + "釋放讀鎖");
        readWriteLock.readLock().release();
        // 釋放寫鎖
        System.out.println(Thread.currentThread().getName() + "釋放寫鎖");
        readWriteLock.writeLock().release();
    }
}      
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 InterProcessReadWriteLockRunnable());
        }
    }
}

輸出如下所示：

pool-1-thread-3擷取寫鎖
pool-1-thread-3擷取讀鎖，鎖降級成功
pool-1-thread-3釋放讀鎖
pool-1-thread-3釋放寫鎖
pool-1-thread-1擷取寫鎖
pool-1-thread-1擷取讀鎖，鎖降級成功
pool-1-thread-1釋放讀鎖
pool-1-thread-1釋放寫鎖
pool-1-thread-4擷取寫鎖
pool-1-thread-4擷取讀鎖，鎖降級成功
pool-1-thread-4釋放讀鎖
pool-1-thread-4釋放寫鎖
pool-1-thread-5擷取寫鎖
pool-1-thread-5擷取讀鎖，鎖降級成功
pool-1-thread-5釋放讀鎖
pool-1-thread-5釋放寫鎖
pool-1-thread-2擷取寫鎖
pool-1-thread-2擷取讀鎖，鎖降級成功
pool-1-thread-2釋放讀鎖
pool-1-thread-2釋放寫鎖

​

​InterProcessReadWriteLock​

​​類除了構造方法外，就隻有​

​readLock​

​​和​

​writeLock​

​​這兩個方法可以調用，而這兩個方法的傳回值是​

​InterProcessMutex​

​​執行個體，是以分布式讀寫鎖​

​InterProcessReadWriteLock​

​​的實作是基于分布式鎖​

​InterProcessMutex​

​。

  • ​​ZooKeeper ： Curator架構之分布式鎖InterProcessMutex​​
zookeeper 分布式 apache 讀鎖 分布式鎖
上一篇: 具有國際影響的《數書九章》
下一篇: iOS Get同步與異步

繼續閱讀