Bully 算法实现
设定集群中有三个节点，通过Bully算法实现选主。节点之间的通信使用的是自我实现的Remoting组件，基于Netty开发，可以以同步，异步的方式发起通信。在后续《分布式通信》系列的文章中，会向大家详细分析Remoting组件。
分布式选举的项目名称：justin-distribute-election
整体结构如图：
image.png
节点和集群类：
Node：表示集群中的节点，包含节点ID，节点状态，节点元数据等属性。
// 节点ID
private final UUID nodeId = UUID.randomUUID();
// 节点状态
private final AtomicReference<NodeStatus> status =
new AtomicReference<NodeStatus>(NodeStatus.UNKNOWN);
// 选举周期，节点发起一次选举，周期值加1。所有的节点必须在相同的周期内。
private final AtomicInteger epoch = new AtomicInteger(0);
private final String[] nodesAddress = PropertiesUtil.getNodesAddress();
// 处理节点请求消息的线程池
private final ExecutorService processorExecutor =
Executors.newFixedThreadPool(PropertiesUtil.getProcessorThreads());
private final Timer timer = new Timer("NodeMessage", true);
// 节点元数据，用于节点间的数据传输
private final Metadata metadata;
// 集群，集群中存储所有节点的元数据，用于选举。
private final Cluster cluster;
Cluster：表示集群，包含所有节点的元数据，主节点，选举周期等属性。
// 选举周期，与各节点一致
private final AtomicInteger epoch = new AtomicInteger(0);
// Key：节点的nodeId，Value：节点的元数据
private final ConcurrentMap<String, Metadata> nodes =
new ConcurrentHashMap<String, Metadata>();
// 选举出的主节点
private AtomicReference<Metadata> leader =
new AtomicReference<Metadata>();
NodeStatus：表示节点的状态，包含三种状态，初始状态(UNKNOWN)、存活状态(ALIVE)、死亡状态(DEAD)。
public enum NodeStatus {
UNKNOWN,
ALIVE,
DEAD
}
节点启动：
-Node节点启动时，在内部启动了4个线程：
-server端线程，用于接收其他节点发送的消息；
-client端线程，用于向其他节点发送消息；
-元数据同步线程，用于在集群内同步各节点的元数据；
-选举线程，用于在集群内选举主节点；
public synchronized void start() {
logger.info("Prepare start node[id: {}, address: {}:{}]", nodeId, host, port);
if (status.get() != NodeStatus.ALIVE) {
try {
// 在集群中加入本地节点
cluster.addNode(nodeId, metadata);
// 启动Server，并注册消息处理器
server = new NettyRemotingServer(new NettyServerConfig(host, port));
server.registerProcessor(MessageType.CLUSTER_SYNC, new ClusterSyncRequestProcessor(this), processorExecutor);
server.registerProcessor(MessageType.ELECTION, new ElectionRequestProcessor(), processorExecutor);
server.registerProcessor(MessageType.VICTORY, new VictoryRequestProcessor(this), processorExecutor);
server.start();
// 启动Client
client = new NettyRemotingClient(new NettyClientConfig());
client.start();
// 本地节点状态设置为ALIVE
setStatus(NodeStatus.ALIVE);
metadata.setNodeStatus(NodeStatus.ALIVE);
// 周期线程，用于在集群内同步各节点的元数据
timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
Node.this.clusterSync(metadata);
}catch (Throwable e) {
logger.error("Cluster sync request failure: " + e.getMessage());
}
}
}, 2*1000, 10*1000);
// 周期线程，用于选主
timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
Node.this.electLeader();
}catch (Throwable e) {
logger.error("Elect leader failure: " + e.getMessage());
}
}
}, 5*1000, 5*1000);
}catch (Exception e) {
e.printStackTrace();
}
}else {
logger.info("Node is alive " + this.toString());
}
}
两种周期线程的作用，如下图：
image.png
Cluster Sync：
元数据同步线程，它的作用在于，在每个节点中维护一个集群所有节点的元数据列表，用于选主。还有就是探测节点是否处于存活状态。
private void clusterSync(Metadata metadata) throws Exception {
for(String nodeAddress : nodesAddress) {
if (!nodeAddress.equals(localAddr)) {
// 向其他节点发送本地节点的元数据
RemotingMessage response = client.invokeSync(nodeAddress, ClusterSyncMessage.getInstance().request(metadata), 3 * 1000);
String peerNodeId = cluster.getNodeIdByaddr(nodeAddress);
if (peerNodeId != null) {
// 如果超过3次没有响应，则设置节点状态为DEAD
if (response == null) {
if (nodeDetectFailNums.containsKey(peerNodeId)) {
int failNums = nodeDetectFailNums.get(peerNodeId);
if (failNums >= 3) {
cluster.getNodes().get(peerNodeId).setNodeStatus(NodeStatus.DEAD);
} else {
nodeDetectFailNums.put(peerNodeId, ++failNums);
}
} else {
nodeDetectFailNums.put(peerNodeId, 1);
}
} else {
nodeDetectFailNums.put(peerNodeId, 0);
}
}
}
}
}
元数据同步消息的处理类是：ClusterSyncRequestProcessor.class
public RemotingMessage processRequest(ChannelHandlerContext ctx,
RemotingMessage request) throws Exception {
Metadata peerNodeMetadata = ClusterSyncMessage.getInstance().getPeerNodeMetadata(request);
logger.info("[" + node.getMetadata() + " ===> " + peerNodeMetadata + "] cluster sync!");
// 将接收的节点元数据存入到本地元数据列表
node.getCluster().addNode(peerNodeMetadata.getNodeId(), peerNodeMetadata);
// 响应ALIVE消息
byte[] res = "ALIVE".getBytes(Charset.forName("UTF-8"));
request.setMessageBody(res);
return request;
}
Elect Leader：
选举线程，它的作用在于检测集群中是否有主节点或者主节点是否存活，否则发起选举。
private void electLeader() throws Exception {
logger.info("[" + metadata + "] cluster leader: " + cluster.getLeader());
if (cluster.getNodes().size() > 1) {
Metadata leader = cluster.getLeader();
// 判断集群中是否有主节点或者主节点是否存活，没有则发起选举。
if ((leader == null) || (leader.getNodeStatus() !=
NodeStatus.ALIVE)) {
logger.info("Starting election ...");
epoch.getAndIncrement();
metadata.setEpoch(epoch.get());
// 获取集群中比自己ID大的所有节点
List<Metadata> largerNodes = cluster.largerNodes(metadata);
// 如果没有则向其他节点发送Victory消息，宣布为主节点
if (largerNodes.isEmpty()) {
if (metadata.getNodeStatus() != NodeStatus.ALIVE) {
logger.warn("Node is not alive: " + metadata);
}else {
cluster.getNodes().get(metadata.getNodeId().toString()).setEpoch(epoch.get());
cluster.setLeader(metadata);
List<Metadata> otherNodes = cluster.otherNodes(metadata);
for (Metadata otherNode : otherNodes) {
client.invokeOneway(otherNode.getNodeAddress(), VictoryMessage.getInstance().request(metadata), 3*1000);
}
}
}else {
// 向比自己ID大的所有节点发送Election消息
for (Metadata largerNode : largerNodes) {
RemotingMessage response = client.invokeSync(largerNode.getNodeAddress(), ElectionMessage.getInstance().request(metadata), 3*1000);
String res = new String(response.getMessageBody(), Charset.forName("UTF-8"));
logger.info("Election response: " + res);
}
}
}
}
}
Election消息的处理类是：ElectionRequestProcessor.class
public RemotingMessage processRequest(ChannelHandlerContext ctx, RemotingMessage request) throws Exception {
Metadata peerNodeMetadata = ElectionMessage.getInstance().getPeerNodeMetadata(request);
logger.info("[ ===> " + peerNodeMetadata + "] election!");
// 响应ALIVE消息
byte[] res = "ALIVE".getBytes(Charset.forName("UTF-8"));
request.setMessageBody(res);
return request;
}
Victory消息的处理类是：VictoryRequestProcessor.class
public RemotingMessage processRequest(ChannelHandlerContext ctx, RemotingMessage request) throws Exception {
Metadata peerNodeMetadata = VictoryMessage.getInstance().getPeerNodeMetadata(request);
logger.info("[" + node.getMetadata() + " ===> " + peerNodeMetadata + "] victory!");
// 更新选举周期
node.setEpoch(peerNodeMetadata.getEpoch());
node.getCluster().setEpoch(peerNodeMetadata.getEpoch());
// 在集群中设置主节点
node.getCluster().setLeader(peerNodeMetadata);
return null;
}
至此，根据Bully算法的选主代码实现完成。
下一篇文章《分布式选举-Raft算法-1 Leader选举 原理》讲解典型的多数派投票选举算法Raft的选举原理。
