Zookeeper实现分布式锁代码实例
作者：外星喵
Zookeeper分布式锁原理
临时顺序节点
该类型的节点创建完成之后，若客户端与服务端断开连接之前没有执行delete操作，Zookeeper会自动删除该类型的节点，同时该类型的节点创建之后，Zookeeper会为该类型的节点在节点名称的基础上增加一个单调递增的编号；
Zookeeper 分布式锁应用了其 临时顺序节点 的特性。实现步骤如下：
获取锁
首先在Zookeeper中创建一个持久节点ParentLock，当第一个客户端要获取锁时，在ParentLock节点下创建一个临时顺序节点Lock1；
接下来客户端1会获取ParentLock下的所有临时顺序子节点并进行排序，然后与自身创建的Lock1比较，判断Lock1是不是最小的（最靠前的），如果是最靠前的，则获取锁成功；
此时，假设又有一个客户端2前来获取锁，则在ParentLock下创建一个临时顺序节点Lock2；
接下来客户端2会获取ParentLock下的所有临时顺序子节点并进行排序，然后与自身创建的Lock2比较，判断Lock2是不是最小的（最靠前的），结果发现Lock2不是最小的；
于是，Lock2向排序比它靠前的第一个节点Lock1注册一个Watcher，用于监听Lock1是否存在，此时意味着客户端2抢锁失败，客户端2进入等待状态；
此时，假设现在又有一个客户端3前来获取锁，则在ParentLock下创建一个临时顺序节点Lock3；
接下来客户端3将获取ParentLock下的所有临时顺序节点并排序，与自身创建的节点Lock3比较，判断Lock3是不是最小的（最靠前的），结果发现Lock3不是最小的；
于是，Lock3向比它靠前的第一个节点Lock2注册一个Watcher，用于监听Lock2是否存在，此时意味着Lock3也抢锁失败，进入阻塞状态； 这样的话，客户端1获取到了锁，客户端2监听了Lock1、客户端3监听了Lock2；
客户端崩溃释放锁（避免死锁）
假设由于网络原因或者其他物理原因，导致客户端1与Zookeeper失去连接，根据临时节点的特性，Zookeeper会自动删除相关联的节点Lock1。 Lock1删除之后，因为客户端2在Lock1上注册了Watcher，因此Lock1删除之后，客户端2会立即收到通知，这时客户端2会再次获取ParentLock下的所有临时顺序子节点并进行排序，然后与自身创建的Lock2比较，判断Lock2是不是最小的（最靠前的），结果发现Lock2是最小的，因此客户端2获取锁成功；
客户端2崩溃同理；
显示释放（客户端任务执行完毕，主动释放锁）
当客户端2执行完任务之后，调用delete方法删除Lock2节点，因为客户端3在Lock2上注册了Watcher，因此Lock2删除之后，客户端3会立即收到通知，这时客户端3会再次获取ParentLock下的所有临时顺序子节点并进行排序，然后与自身创建的Lock3比较，判断Lock3是不是最小的（最靠前的），结果发现Lock3是最小的，因此客户端3获取锁成功；
zookeeper分布式锁实现
配置文件pom.xml
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipesss</artifactId>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>4.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.0.1</version>
</dependency>
配置文件application.properties
##ZooKeeper 集成 Curator zk.url = 127.0.0.1:2181
conf
@Configuration
@Slf4j
public class ZooKeeperConf {
@Value("${zk.url}")
private String zkUrl;
@Bean
public CuratorFramework getCuratorFramework() {
// 用于重连策略，1000毫秒是初始化的间隔时间，3代表尝试重连次数
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
CuratorFramework client = CuratorFrameworkFactory.newClient(zkUrl, retryPolicy);
//必须调用start开始连接ZooKeeper
client.start();
// /**
// * 使用Curator，可以通过LeaderSelector来实现领导选取；
// * 领导选取：选出一个领导节点来负责其他节点；如果领导节点不可用，则在剩下的机器里再选出一个领导节点
// */
// // 构造一个监听器
// LeaderSelectorListenerAdapter listener = new LeaderSelectorListenerAdapter() {
// @Override
// public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
// log.info("get leadership");
// // 领导节点，方法结束后退出领导。zk会再次重新选择领导
//
// }
// };
// LeaderSelector selector = new LeaderSelector(client, "/schedule", listener);
// selector.autoRequeue();
// selector.start();
return client;
}
}controller
@RestController
@RequestMapping("/zookeeper")
public class ZooKeeperController {
@Autowired
private CuratorFramework zkClient;
@Autowired
private ZooKeeperImpl zooKeeper;
/**
* zookeeper 获取节点下的数据
* <p>
* post请求： http://localhost:8082/zookeeper/makeOrder?path=/task
*
* @param path
* @return
*/
@PostMapping("/getData")
public String getData(@RequestParam String path) {
byte[] bytes = null;
try {
bytes = zkClient.getData().forPath(path);
} catch (Exception e) {
e.printStackTrace();
}
String str = new String(bytes);
return str;
}
@PostMapping("/create")
public String create(@RequestParam String path) {
try {
zkClient.create().forPath(path);
} catch (Exception e) {
e.printStackTrace();
}
return "success";
}
@PostMapping("/delete")
public String delete(@RequestParam String path) {
try {
zkClient.delete().forPath(path);
} catch (Exception e) {
e.printStackTrace();
}
return "success";
}
@PostMapping("/setData")
public String setData(@RequestParam(value = "path") String path, @RequestParam(value = "data") String data) {
try {
zkClient.setData().forPath(path, data.getBytes());
} catch (Exception e) {
e.printStackTrace();
}
return "success";
}
@PostMapping("/check")
public String check(@RequestParam(value = "path") String path) {
Stat stat = null;
try {
stat = zkClient.checkExists().forPath(path);
} catch (Exception e) {
e.printStackTrace();
}
return "stat" + stat;
}
@PostMapping("/children")
public String children(@RequestParam(value = "path") String path) {
List<String> children = null;
try {
children = zkClient.getChildren().forPath(path);
} catch (Exception e) {
e.printStackTrace();
}
return "children" + children;
}
@PostMapping("/watch")
public String watch(@RequestParam(value = "path") String path) {
Stat stat = null;
try {
stat = zkClient.checkExists().watched().forPath(path);
} catch (Exception e) {
e.printStackTrace();
}
return "watch " + stat;
}
/**
* zookeeper分布式锁
*
* @param product
* @return
*/
@PostMapping("/makeOrder")
public String makeOrder(@RequestParam(value = "product") String product) {
zooKeeper.makeOrder(product);
return "success";
}
}service
@Service
@Slf4j
public class ZooKeeperImpl {
private static final String lockPath = "/lock/order";
@Autowired
private CuratorFramework zkClient;
public void makeOrder(String product) {
log.info("try do job for " + product);
String path = lockPath + "/" + product;
try {
// InterProcessMutex 构建一个分布式锁
InterProcessMutex lock = new InterProcessMutex(zkClient, path);
try {
if (lock.acquire(5, TimeUnit.HOURS)) {
// 模拟业务处理耗时5秒
Thread.sleep(5*1000);
log.info("do job " + product + "done");
}
} finally {
// 释放该锁
lock.release();
}
} catch (Exception e) {
// zk异常
e.printStackTrace();
}
}
}到此这篇关于Zookeeper实现分布式锁代码实例的文章就介绍到这了,更多相关Zookeeper分布式锁内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！