反問面試官：如何實現集群內選主
作者：程序員半支煙
本文主要演示了一個簡易的多Server的選主過程，以下代碼是一個簡單的基于Netty實現的集群選舉過程的示例。
面試官經常喜歡問什么zookeeper選主原理、什么CAP理論、什么數據一致性。經常都被問煩了，我就想問問面試官，你自己還會實現一個簡單的集群內選主呢？估計大部分面試官自己也寫不出來。
本篇使用 Java 和 Netty 實現簡單的集群選主過程的示例。
這個示例展示了多個節點通過投票選舉一個新的主節點的過程。Netty 用于節點間的通信，而每個節點則負責發起和響應選舉消息。
集群選主流程
1.選主流程
咱們且不說zookeeper如何選主，單說人類選主，也是采用少數服從多數的原則。人類選主時，中間會經歷如下過程：
- 如果我沒有熟悉的或者沒找到能力比我強的，首先投給自己一票。
- 隨著時間推移，可能后面的人介紹了各自的特點和實力，那我可能會改投給別人。
- 所有人將投票信息放入到統計箱中。
- 最終票數最多的人是領導者。
同樣的，zookeeper在選主時，也是這樣的流程。假設有五大服務器：
- 服務器1先給自身投票
- 后續起來的服務器2也會投自身一票，然后服務器1觀察到服務器2的id比較大，則會改投服務器2
- 后續起來的服務器3也會投自身一票，然后服務1和服務器2發現服務器3的id比較大，則都會改投服務器3。服務器3被確定為領導者。
- 服務器4起來后也會投自身一票，然后發現服務器3已經有3票了，立馬改投服務器3。
- 服務器5與服務器4的操作一樣。
2.選主協議
在選主過程中采用的是超過半數的協議。在選主過程中，會需要如下幾類消息：
- 投票請求：節點發出自己的投票請求。
- 接受投票：其余節點作出判斷，如果覺得id較大，則接受投票。
- 選舉勝出：當選主節點后，廣播勝出消息。
代碼實現
下面模擬3個節點的選主過程，核心步驟如下：
(1) 定義消息類型、消息對象、節點信息
public enum MessageType {
VOTE_REQUEST, // 投票請求
VOTE, // 投票
ELECTED // 選舉完成后的勝出消息
}
public class ElectionMessage implements Serializable {
private MessageType type;
private int nodeId; // 節點ID
private long zxId; // ZXID：類似于ZooKeeper中的邏輯時鐘，用于比較
private int voteFor; // 投票給的節點ID
}
public class ElectionNode {
private int nodeId; // 當前節點ID
private long zxId; // 當前節點的ZXID
private volatile int leaderId; // 當前選舉的Leader ID
private String host;
private int port;
private ConcurrentHashMap<Integer, Integer> voteMap = new ConcurrentHashMap<>(); // 此節點對每個節點的投票情況
private int totalNodes; // 集群總節點數
}(2) 每個節點利用Netty啟動Server
public void start() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(
new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
new ObjectEncoder(),
new ElectionHandler(ElectionNode.this));
}
});
ChannelFuture future = serverBootstrap.bind(port).sync();
System.out.println("Node " + nodeId + " started on port " + port);
// 啟動后開始選舉過程
startElection();
// future.channel().closeFuture().sync();
} catch (Exception e) {
} finally {
// bossGroup.shutdownGracefully();
// workerGroup.shutdownGracefully();
}
}(3) 啟動后利用Netty發送投票請求
public void sendVoteRequest(String targetHost, int targetPort) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(
new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
new ObjectEncoder(),
new ElectionHandler(ElectionNode.this));
}
});
ChannelFuture future = bootstrap.connect(targetHost, targetPort).sync();
ElectionMessage voteRequest = new ElectionMessage(ElectionMessage.MessageType.VOTE_REQUEST, nodeId, zxId, nodeId);
future.channel().writeAndFlush(voteRequest);
// future.channel().closeFuture().sync();
} catch (Exception e) {
} finally {
// group.shutdownGracefully();
}
}(4) 節點接受到投票請求后，做相關處理
節點在收到消息后，做相關邏輯處理：處理投票請求、處理確認投票、處理選主結果。
處理投票請求：判斷是否是否接受投票信息。只有在主節點沒確定并且zxId較大時，才發送投票消息。如果接受了投票請求的話，則更新本地的投票邏輯，然后給投票節點發送接受投票的消息
處理確認投票：如果投票消息被接受了，則更新本地的投票邏輯。
處理選主結果：如果收到了選主結果的消息，則更新本地的主節點。
public class ElectionHandler extends ChannelInboundHandlerAdapter {
private final ElectionNode node;
public ElectionHandler(ElectionNode node) {
this.node = node;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ElectionMessage electionMessage = (ElectionMessage) msg;
System.out.println("Node " + node.getNodeId() + " received: " + electionMessage);
if (electionMessage.getType() == ElectionMessage.MessageType.VOTE_REQUEST) {
// 判斷是否是否接受投票信息。只有在主節點沒確定并且zxId較大時，才發送投票消息
// 如果接受了投票請求的話，則更新本地的投票邏輯，然后給投票節點發送接受投票的消息
if (electionMessage.getZxId() >= node.getZxId() && node.getLeaderId() == 0) {
node.receiveVote(electionMessage.getNodeId());
ElectionMessage voteMessage = new ElectionMessage(ElectionMessage.MessageType.VOTE, electionMessage.getNodeId(), electionMessage.getZxId(), electionMessage.getNodeId());
ctx.writeAndFlush(voteMessage);
} else {
// 如果已經確定主節點了，直接發送ELECTED消息
sendLeaderInfo(ctx);
}
} else if (electionMessage.getType() == ElectionMessage.MessageType.VOTE) {
// 如果投票消息被接受了，則更新本地的投票邏輯。
if (electionMessage.getZxId() >= node.getZxId() && node.getLeaderId() == 0) {
node.receiveVote(electionMessage.getNodeId());
} else {
// 如果已經確定主節點了，直接發送ELECTED消息
sendLeaderInfo(ctx);
}
} else if (electionMessage.getType() == ElectionMessage.MessageType.ELECTED) {
if (node.getLeaderId() == 0) {
node.setLeaderId(electionMessage.getVoteFor());
}
}
}(5) 接受別的節點的投票
這里是比較關鍵的一步，當確定接受某個節點時，則更新本地的投票數，然后判斷投票數是否超過半數，超過半數則確定主節點。同時，再將主節點廣播出去。
此時，其余節點接收到選主確認的消息后，都會更新自己的本地的主節點信息。
public void receiveVote(int nodeId) {
voteMap.merge(nodeId, 1, Integer::sum);
// 比較出votes里值，取出最大的那個對應的key
int currentVotes = voteMap.values().stream().max(Integer::compareTo).get();
if (currentVotes > totalNodes / 2 && leaderId == 0) {
setLeaderId(nodeId);
broadcastElected();
}
}(6) 廣播選主結果
/**
* 廣播選舉結果
*/
private void broadcastElected() {
for (int i = 1; i <= totalNodes; i++) {
if (i != nodeId) {
sendElectedMessage(host, 9000 + i);
}
}
}
/**
* 發送選舉結果
*
* @param targetHost
* @param targetPort
*/
public void sendElectedMessage(String targetHost, int targetPort) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(
new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
new ObjectEncoder(),
new ElectionHandler(ElectionNode.this));
}
});
ChannelFuture future = bootstrap.connect(targetHost, targetPort).sync();
ElectionMessage electedMessage = new ElectionMessage(ElectionMessage.MessageType.ELECTED, leaderId, zxId, leaderId);
future.channel().writeAndFlush(electedMessage);
// future.channel().closeFuture().sync();
} catch (Exception e) {
} finally {
// group.shutdownGracefully();
}
}(7) 完整代碼
完整代碼：https://gitee.com/yclxiao/specialty/blob/master/javacore/src/main/java/com/ycl/election/ElectionHandler.java
總結
本文主要演示了一個簡易的多Server的選主過程，以上代碼是一個簡單的基于Netty實現的集群選舉過程的示例。在實際場景中，選舉邏輯遠比這個復雜，需要處理更多的網絡異常、重復消息、并發問題等。
責任編輯：趙寧寧
來源：
程序員半支煙
