【从入门到放弃-ZooKeeper】ZooKeeper实战-分布式竞选

前言

上文【从入门到放弃-ZooKeeper】ZooKeeper实战-分布式锁-升级版中,我们通过利用ZooKeeper的临时节点和Watcher特性,实现了一个分布式锁。
本文我们结合实际场景,完成一个分布式竞争选举。

设计

这里我们实现一个公平的选举方式,即先参加选举的优先被选为leader。
具体的实现思路 参考了ZooKeeper提供的官方示例:zookeeper-recipes-election

【从入门到放弃-ZooKeeper】ZooKeeper实战-分布式竞选

  • START:服务器开始竞选
  • OFFER:创建临时顺序结点
  • DETERMINE:开始决策,将临时节点按末尾序号从小到大排序,如果当前节点的序号最小,则竞选成功,否则,则Watch前一个节点,当前一个节点被删除时,再次进行决策
  • ELECTED:当前节点是序号最小的节点,竞选成功
  • READY:当前节点不是序号最小的节点,竞选不成功,Watch前一个节点,进入READY态
  • FAILED:当出现异常情况时,为失败状态
  • STOP:结束竞选

LeaderElectionSupport

public class LeaderElectionSupport implements LeaderElection{
private static Logger logger = LoggerFactory.getLogger(LeaderElectionSupport.class);
//ZooKeeper客户端,进行ZooKeeper操作
private ZooKeeper zooKeeper;
//根节点名称
private String dir;
//节点前缀
private String node;
//ZooKeeper鉴权信息
private List<ACL> acls;
//要加锁节点
private String fullPath;
//选举状态
private State state;
//监听器
private Set<LeaderElectionListener> listeners;
//存当前节点的信息
private volatile LeaderNode leaderNode;
//监察器
private Watcher watcher;
/**
* Constructor.
*
* @param zooKeeper the zoo keeper
* @param dir       the dir
* @param node      the node
* @param acls      the acls
*/
public LeaderElectionSupport(ZooKeeper zooKeeper, String dir, String node, List<ACL> acls) {
this.zooKeeper = zooKeeper;
this.dir = dir;
this.node = node;
this.acls = acls;
this.fullPath = dir.concat("/").concat(this.node);
init();
state = State.STOP;
listeners = Collections.synchronizedSet(new HashSet<>());
}
/**
* 初始化根节点、检查器等
* */
private void init() {
try {
watcher = new LeaderWatcher();
Stat stat = zooKeeper.exists(dir, false);
if (stat == null) {
zooKeeper.create(dir, null, acls, CreateMode.PERSISTENT);
}
} catch (Exception e) {
logger.error("[LeaderElectionSupport#init] error : " + e.toString(), e);
}
}
}

start

/**
* Start.
* 开始竞选
*/
@Override
public void start() {
synchronized (this) {
state = State.START;
dispatchEvent(EventType.START);
offerElection();
determineElection();
}
}

offerElection

/**
* 创建临时节点,参加竞选,并将主机信息保存在node中
* */
private void offerElection() {
dispatchEvent(EventType.OFFER_START);
state = State.OFFER;
if (leaderNode == null) {
synchronized (this) {
try {
if (leaderNode == null) {
InetAddress ia = InetAddress.getLocalHost();
LeaderNode tmpNode = new LeaderNode();
tmpNode.setHostName(ia.getHostName());
String path = zooKeeper.create(fullPath, ConversionUtil.objectToBytes(ia.getHostName()), acls, CreateMode.EPHEMERAL_SEQUENTIAL);
tmpNode.setNodePath(path);
tmpNode.setId(NodeUtil.getNodeId(path));
leaderNode = tmpNode;
}
} catch (Exception e) {
becomeFailed(e);
}
}
}
dispatchEvent(EventType.OFFER_COMPLETE);
}

determineElection

/**
* 决定竞选结果
* 1、竞选节点序号最低的赢取选举
* 2、未赢得选举的节点,监听上一个节点,直到上一个节点被删除,则尝试重新竞选
* */
private void determineElection() {
dispatchEvent(EventType.DETERMINE_START);
state = State.DETERMINE;
synchronized (this) {
TreeSet<String> nodePathSet = getNodePathSet();
if (nodePathSet.isEmpty()) {
becomeFailed(new Exception("no node"));
return;
}
String leaderPath = nodePathSet.first();
if (leaderNode.getNodePath().equalsIgnoreCase(leaderPath)) {
becomeLeader();
} else {
becomeReady(nodePathSet.headSet(leaderNode.getNodePath()).last());
}
}
dispatchEvent(EventType.DETERMINE_COMPLETE);
}

becomeLeader

/**
* 竞选成功
* */
private void becomeLeader() {
dispatchEvent(EventType.ELECTED_START);
state = State.ELECTED;
dispatchEvent(EventType.ELECTED_COMPLETE);
}

becomeReady

/**
* 竞选失败进入就绪态
* */
private void becomeReady(String path) {
try {
Stat stat = zooKeeper.exists(path, watcher);
if (stat == null) {
determineElection();
} else {
dispatchEvent(EventType.READY_START);
state = State.READY;
dispatchEvent(EventType.READY_COMPLETE);
}
} catch (KeeperException e) {
becomeFailed(e);
} catch (InterruptedException e) {
becomeFailed(e);
}
}

becomeFailed

/**
* 当发生异常时,更新为FAILED状态
* */
private void becomeFailed(Exception e) {
state = State.FAILED;
dispatchEvent(EventType.FAILED);
logger.error("[LeaderElectionSupport#becomeFailed] error : " + e.toString(), e);
}

getNodePathSet

/**
* 获取参加竞选的节点信息
* */
private TreeSet<String> getNodePathSet() {
TreeSet<String> nodeSet = new TreeSet<>();
try {
List<String> nodes = zooKeeper.getChildren(dir, false);
for (String node : nodes) {
nodeSet.add(dir.concat("/").concat(node));
}
} catch (KeeperException e) {
becomeFailed(e);
} catch (InterruptedException e) {
becomeFailed(e);
}
return nodeSet;
}

stop

/**
* Stop.
* 停止竞选
*/
@Override
public void stop() {
synchronized (this) {
dispatchEvent(EventType.STOP_START);
deleteNode();
state = State.STOP;
dispatchEvent(EventType.STOP_COMPLETE);
}
}

deleteNode

/**
* 停止时,删除节点,退出竞选
* */
private void deleteNode() {
try {
if (leaderNode != null) {
synchronized (this) {
zooKeeper.delete(leaderNode.getNodePath(), -1);
leaderNode = null;
}
}
} catch (InterruptedException e) {
becomeFailed(e);
} catch (KeeperException e) {
becomeFailed(e);
}
}

getLeaderHostName

/**
* Gets get leader host name.
*
* @return the get leader host name
*/
@Override
public String getLeaderHostName() {
synchronized (this) {
TreeSet<String> nodePathSet = getNodePathSet();
if (!nodePathSet.isEmpty()) {
try {
String leaderPath = nodePathSet.first();
return (String) ConversionUtil.bytesToObject(zooKeeper.getData(leaderPath, false, null));
} catch (KeeperException e) {
logger.error("[LeaderWatcher#getLeaderHostName] error : " + e.toString(), e);
} catch (InterruptedException e) {
logger.error("[LeaderWatcher#getLeaderHostName] error : " + e.toString(), e);
} catch (IOException e) {
logger.error("[LeaderWatcher#getLeaderHostName] error : " + e.toString(), e);
} catch (ClassNotFoundException e) {
logger.error("[LeaderWatcher#getLeaderHostName] error : " + e.toString(), e);
}
}
return null;
}
}

getLeaderNodePath

/**
* Gets get leader node path.
*
* @return the get leader node path
*/
@Override
public String getLeaderNodePath() {
synchronized (this) {
TreeSet<String> nodePathSet = getNodePathSet();
return nodePathSet.isEmpty() ? null : nodePathSet.first();
}
}

LeaderWatcher

/**
* 内部watcher类,当竞选失败时,watch前一个节点,当前一个节点别移除时,再次发起决策
* */
private class LeaderWatcher implements Watcher {
/**
* Process.
*
* @param watchedEvent the watched event
*/
@Override
public void process(WatchedEvent watchedEvent) {
try {
if (Event.EventType.NodeDeleted.equals(watchedEvent.getType()) && !State.STOP.equals(state)) {
determineElection();
}
} catch (Exception e) {
logger.error("[LeaderWatcher#process] error : " + e.toString(), e);
}
}
}

总结

以上就是我们利用ZooKeeper的临时节点和Watcher特性实现的公平模式分布式竞选。

可以进行简单的选主操作,适用于如执行单机定时任务、心跳检测等场景。实际上是实现的Master-Slave模型。

源代码可见:aloofJr

而对高可用要求较多的复杂选举场景,如分布式存储、同步等,则需要考虑集群一致性、脑裂等各种情况,则需要实现如Paxos、raft、Zab等一致性算法协议。如ZooKeeper集群的选举模式就是使用的Zab算法。
我们后续会进行深入的探讨。

更多文章

见我的博客:https://nc2era.com

written by AloofJr,转载请注明出处