使用ZooKeeper实现分布式队列、分布式锁和选举详解
java1234
共 36258字，需浏览 73分钟
· 2021-08-01
点击上方蓝色字体，选择“标星公众号”
优质文章，第一时间送达
1、分布式队列
1）、offer方法
public class DistributedQueue {
public boolean offer(byte[] data) throws KeeperException, InterruptedException {
for (; ; ) {
try {
zookeeper.create(dir + "/" + prefix, data, acl, CreateMode.PERSISTENT_SEQUENTIAL);
return true;
} catch (KeeperException.NoNodeException e) {
zookeeper.create(dir, new byte[0], acl, CreateMode.PERSISTENT);
}
}
}
2）、element方法
public class DistributedQueue {
public byte[] element() throws NoSuchElementException, KeeperException, InterruptedException {
Map<Long, String> orderedChildren;
while (true) {
try {
//获取所有排好序的子节点
orderedChildren = orderedChildren(null);
} catch (KeeperException.NoNodeException e) {
throw new NoSuchElementException();
}
if (orderedChildren.size() == 0) {
throw new NoSuchElementException();
}
//返回队头节点的数据
for (String headNode : orderedChildren.values()) {
if (headNode != null) {
try {
return zookeeper.getData(dir + "/" + headNode, false, null);
} catch (KeeperException.NoNodeException e) {
//另一个客户端已经移除了队头节点,尝试获取下一个节点
}
}
}
}
}
private Map<Long, String> orderedChildren(Watcher watcher) throws KeeperException, InterruptedException {
Map<Long, String> orderedChildren = new TreeMap<>();
List<String> childNames;
childNames = zookeeper.getChildren(dir, watcher);
for (String childName : childNames) {
try {
if (!childName.regionMatches(0, prefix, 0, prefix.length())) {
LOG.warn("Found child node with improper name: {}", childName);
continue;
}
String suffix = childName.substring(prefix.length());
Long childId = Long.parseLong(suffix);
orderedChildren.put(childId, childName);
} catch (NumberFormatException e) {
LOG.warn("Found child node with improper format : {}", childName, e);
}
}
return orderedChildren;
}
3）、remove方法
public class DistributedQueue {
public byte[] remove() throws NoSuchElementException, KeeperException, InterruptedException {
Map<Long, String> orderedChildren;
while (true) {
try {
//获取所有排好序的子节点
orderedChildren = orderedChildren(null);
} catch (KeeperException.NoNodeException e) {
throw new NoSuchElementException();
}
if (orderedChildren.size() == 0) {
throw new NoSuchElementException();
}
//移除队头节点
for (String headNode : orderedChildren.values()) {
String path = dir + "/" + headNode;
try {
byte[] data = zookeeper.getData(path, false, null);
zookeeper.delete(path, -1);
return data;
} catch (KeeperException.NoNodeException e) {
//另一个客户端已经移除了队头节点,尝试移除下一个节点
}
}
}
}
2、分布式锁
1）、排他锁
2）、羊群效应
3）、共享锁
4）、排他锁源码解析
public class WriteLock extends ProtocolSupport {
public synchronized boolean lock() throws KeeperException, InterruptedException {
if (isClosed()) {
return false;
}
//确认持久父节点是否存在
ensurePathExists(dir);
//真正获取锁的逻辑 调用ProtocolSupport的retryOperation()方法
return (Boolean) retryOperation(zop);
}
class ProtocolSupport {
protected Object retryOperation(ZooKeeperOperation operation)
throws KeeperException, InterruptedException {
KeeperException exception = null;
for (int i = 0; i < RETRY_COUNT; i++) {
try {
//调用LockZooKeeperOperation的execute()方法
return operation.execute();
} catch (KeeperException.SessionExpiredException e) {
LOG.warn("Session expired {}. Reconnecting...", zookeeper, e);
throw e;
} catch (KeeperException.ConnectionLossException e) {
if (exception == null) {
exception = e;
}
LOG.debug("Attempt {} failed with connection loss. Reconnecting...", i);
retryDelay(i);
}
}
throw exception;
}
public class WriteLock extends ProtocolSupport {
private class LockZooKeeperOperation implements ZooKeeperOperation {
private void findPrefixInChildren(String prefix, ZooKeeper zookeeper, String dir)
throws KeeperException, InterruptedException {
List<String> names = zookeeper.getChildren(dir, false);
for (String name : names) {
if (name.startsWith(prefix)) {
id = name;
LOG.debug("Found id created last time: {}", id);
break;
}
}
if (id == null) {
id = zookeeper.create(dir + "/" + prefix, data, getAcl(), EPHEMERAL_SEQUENTIAL);
LOG.debug("Created id: {}", id);
}
}
@SuppressFBWarnings(
value = "NP_NULL_PARAM_DEREF_NONVIRTUAL",
justification = "findPrefixInChildren will assign a value to this.id")
public boolean execute() throws KeeperException, InterruptedException {
do {
if (id == null) {
long sessionId = zookeeper.getSessionId();
String prefix = "x-" + sessionId + "-";
//创建临时顺序节点
findPrefixInChildren(prefix, zookeeper, dir);
idName = new ZNodeName(id);
}
//获取所有子节点
List<String> names = zookeeper.getChildren(dir, false);
if (names.isEmpty()) {
LOG.warn("No children in: {} when we've just created one! Lets recreate it...", dir);
id = null;
} else {
//对所有子节点进行排序
SortedSet<ZNodeName> sortedNames = new TreeSet<>();
for (String name : names) {
sortedNames.add(new ZNodeName(dir + "/" + name));
}
ownerId = sortedNames.first().getName();
SortedSet<ZNodeName> lessThanMe = sortedNames.headSet(idName);
//是否存在序号比自己小的节点
if (!lessThanMe.isEmpty()) {
ZNodeName lastChildName = lessThanMe.last();
lastChildId = lastChildName.getName();
LOG.debug("Watching less than me node: {}", lastChildId);
//有序号比自己小的节点,则调用exist()向前一个节点注册watcher
Stat stat = zookeeper.exists(lastChildId, new LockWatcher());
if (stat != null) {
return Boolean.FALSE;
} else {
LOG.warn("Could not find the stats for less than me: {}", lastChildName.getName());
}
}
//没有序号比自己小的节点,则获取锁
else {
if (isOwner()) {
LockListener lockListener = getLockListener();
if (lockListener != null) {
lockListener.lockAcquired();
}
return Boolean.TRUE;
}
}
}
}
while (id == null);
return Boolean.FALSE;
}
public class WriteLock extends ProtocolSupport {
public synchronized void unlock() throws RuntimeException {
if (!isClosed() && id != null) {
try {
//删除当前节点,此时会触发后一个节点的watcher
ZooKeeperOperation zopdel = () -> {
zookeeper.delete(id, -1);
return Boolean.TRUE;
};
zopdel.execute();
} catch (InterruptedException e) {
LOG.warn("Unexpected exception", e);
Thread.currentThread().interrupt();
} catch (KeeperException.NoNodeException e) {
} catch (KeeperException e) {
LOG.warn("Unexpected exception", e);
throw new RuntimeException(e.getMessage(), e);
} finally {
LockListener lockListener = getLockListener();
if (lockListener != null) {
lockListener.lockReleased();
}
id = null;
}
}
}
3、选举
public class LeaderElectionSupport implements Watcher {
public synchronized void start() {
state = State.START;
dispatchEvent(EventType.START);
LOG.info("Starting leader election support");
if (zooKeeper == null) {
throw new IllegalStateException(
"No instance of zookeeper provided. Hint: use setZooKeeper()");
}
if (hostName == null) {
throw new IllegalStateException(
"No hostname provided. Hint: use setHostName()");
}
try {
//发起选举请求 创建临时顺序节点
makeOffer();
//选举请求是否被满足
determineElectionStatus();
} catch (KeeperException | InterruptedException e) {
becomeFailed(e);
}
}
private void makeOffer() throws KeeperException, InterruptedException {
state = State.OFFER;
dispatchEvent(EventType.OFFER_START);
LeaderOffer newLeaderOffer = new LeaderOffer();
byte[] hostnameBytes;
synchronized (this) {
newLeaderOffer.setHostName(hostName);
hostnameBytes = hostName.getBytes();
newLeaderOffer.setNodePath(zooKeeper.create(rootNodeName + "/" + "n_",
hostnameBytes, ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL));
leaderOffer = newLeaderOffer;
}
LOG.debug("Created leader offer {}", leaderOffer);
dispatchEvent(EventType.OFFER_COMPLETE);
}
private void determineElectionStatus() throws KeeperException, InterruptedException {
state = State.DETERMINE;
dispatchEvent(EventType.DETERMINE_START);
LeaderOffer currentLeaderOffer = getLeaderOffer();
String[] components = currentLeaderOffer.getNodePath().split("/");
currentLeaderOffer.setId(Integer.valueOf(components[components.length - 1].substring("n_".length())));
//获取所有子节点并排序
List<LeaderOffer> leaderOffers = toLeaderOffers(zooKeeper.getChildren(rootNodeName, false));
for (int i = 0; i < leaderOffers.size(); i++) {
LeaderOffer leaderOffer = leaderOffers.get(i);
if (leaderOffer.getId().equals(currentLeaderOffer.getId())) {
LOG.debug("There are {} leader offers. I am {} in line.", leaderOffers.size(), i);
dispatchEvent(EventType.DETERMINE_COMPLETE);
//如果当前节点是第一个,则成为Leader
if (i == 0) {
becomeLeader();
}
//如果有选举请求在当前节点前面,则进行等待,调用exist()向前一个节点注册watcher
else {
becomeReady(leaderOffers.get(i - 1));
}
break;
}
}
}
作者 | 邋遢的流浪剑客
来源 | csdn.net/qq_40378034/article/details/117014648
评论
用 Shader 实现旗帜飘扬动画效果
我觉得对于刚入门 3D 编程的朋友来说，如果能够完成代码创建模型数据->创建材质->编写Shader动画这一系列，想必会有满满的成就感。今天就用 Cocos Creator 的 utils.MeshUtils.createMesh 接口，带大家感受一下这个流程。这个流程不仅可以用于新手学
COCOS
2
互联网晚报 | 大麦网已退款凤凰传奇演唱会“柱子票”；钟薛高再成被执行人；iPhone 16或取消实体音量键和电源键
大麦网回应凤凰传奇演唱会买到“柱子票”：已退票退款据报道，凤凰传奇2024巡回演唱会常州站演出结束的第二天，有网友称自己在大麦网买到“柱子票”，因为观看效果不佳，要求退款被拒。23日，记者从涉事网友处了解到，大麦方面给出了退款建议，但被其拒绝，“我希望平台退款加赔偿，并重视屡次出现的‘柱子票’问题。
产品刘
0
AI论文写作工具和生成器（一）
随着人工智能和大模型的迅猛发展，AI对研究人员和学生提供了极大的写作便利。本文将介绍市面上常用的AI论文写作工具，帮助你提高论文写作效率并遵循学术道德。请仅将AI论文生成器视为辅助参考手段，切勿直接挪用全文。XPaper AlXPaper AI是由点击式创作工具晓语台推出的一款论文写作生成平台，只需
IQ前端
0
使用 GitHub Actions 构建 Golang PGO
今年 2 月，我宣布 Dolt 版本现已构建为配置文件引导优化 (pgo) 二进制文件，利用 Golang 1.20 的强大功能将 Dolt 的读取性能提高 5.3%。在我宣布这一消息之前，我们的一位常驻 Golang 专家 Zach 试验并测试了 Golang 的 pgo 功能
GoCN
0
盘点一个使用超级鹰识别验证码并自动登录的案例
点击上方“Python共享之家”，进行关注回复“资源”即可获赠Python学习资料今日鸡汤江上几人在，天涯孤棹还。大家好，我是皮皮。一、前言前几天在Python钻石交流群【静惜】问了一个Python实现识别验证码并自动登录的问题，提问截图如下：验证码的截图如下所示：二、实现过程这里大家激烈的探讨，【
IT共享之家
0
Langchain使用 | 模型、提示和解析器、存储
零、LangChain介绍为各种不同基础模型提供统一接口- 帮助管理提示的框架- 一套中心化接口，用于处理长期记忆（参见Memory）、外部数据（参见Indexes）、其他 LLM（参见Chains）以及 LLM 无法处理的任务的其他代理（例如，计算或搜索）。总的来说，有六大核心模块：Models：
Python之王
0
你真的理解 devDependencies 和 dependencies 的区别吗？
点击上方 前端Q，关注公众号回复加群，加入前端Q技术交流群作者：井柏然原文：https://juejin.cn/post/7135795969370619918你是否真的理解 devDependencies 和 dependencies 的区别？如果不能确切的回答、理解还停留在模糊的阶段，
前端Q
0
Stability AI开放Stable Diffusion 3 API，在线免费使用
「Stability AI」宣布开放其最新文本到图像生成模型「Stable Diffusion 3」的API接口，供开发者和企业使用。该模型采用创新的多模态扩散转换器架构，在字体、细节还原、提示理解等方面表现优异，评测结果超越了业内其他顶尖系统。与DALL-E 3和Midjourney v6等最先进
IQ前端
0