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 释放分布式锁
Curator框架的分布式锁InterProcessSemaphoreMutex就介绍到这里，如果博主有说错的地方或者大家有不同的见解，欢迎大家评论补充。