5.实现简化版raft协议完成选举

1.设计

前面已经完成了netty的集成，接下来就是借助netty完成选举就行了。

针对选举，我们用到了VotRequestMessage、VotRespMessage、当节点下线时NodeOfflineMessage、NodeOnlineMessage、NodeOnlineRespMessage

1.1 节点详细的交互

1.2 对所有消息的处理使用策略模式

由于我们是Spring的应用，我可以借助Spring的容器完成对消息类型的选择

1.2.1 创建 IMessageService 接口

所有消息的处理都实现此接口

public interface IMessageService {byte getMessageType();void execute(ChannelHandlerContext ctx, DttaskMessage message);}

1.2.2 DttaskMessage

@Data
public class DttaskMessage {public static final byte COMMON_RESP = 0X00;public static final byte PING = 0X01;public static final byte PONG = 0X02;public static final byte VOTING = 0X03;public static final byte VOT_RESP = 0X04;public static final byte NODE_OFFLINE = 0X05;public static final byte NODE_ONLINE = 0X06;public static final byte NODE_ONLINE_RESP = 0X07;// 类型private byte type;// 消息实际信息private String info;public static DttaskMessage buildPingMessage(long serverId) {DttaskMessage dttaskMessage = new DttaskMessage();dttaskMessage.setType(PING);dttaskMessage.setInfo(JSON.toJSONString(new PingMessage(serverId)));return dttaskMessage;}public static DttaskMessage buildPongMessage(long serverId) {DttaskMessage dttaskMessage = new DttaskMessage();dttaskMessage.setType(PONG);dttaskMessage.setInfo(JSON.toJSONString(new PongMessage(serverId)));return dttaskMessage;}public static DttaskMessage buildCommonRespMessage(String message, boolean successFlag) {DttaskMessage dttaskMessage = new DttaskMessage();dttaskMessage.setType(COMMON_RESP);dttaskMessage.setInfo(JSON.toJSONString(new CommonRespMessage(message, successFlag)));return dttaskMessage;}public static DttaskMessage buildNodeOnlineRespMessage(long serverId) {DttaskMessage dttaskMessage = new DttaskMessage();dttaskMessage.setType(NODE_ONLINE_RESP);dttaskMessage.setInfo(JSON.toJSONString(new NodeOnlineRespMessage(serverId)));return dttaskMessage;}public static DttaskMessage buildNodeOnlineMessage(long serverId) {DttaskMessage dttaskMessage = new DttaskMessage();dttaskMessage.setType(NODE_ONLINE);dttaskMessage.setInfo(JSON.toJSONString(new NodeOnlineMessage(serverId)));return dttaskMessage;}public static DttaskMessage buildNodeOfflineMessage(long serverId) {DttaskMessage dttaskMessage = new DttaskMessage();dttaskMessage.setType(NODE_OFFLINE);dttaskMessage.setInfo(JSON.toJSONString(new VotRespMessage(serverId)));return dttaskMessage;}public static DttaskMessage buildVotRespMessage(long serverId) {DttaskMessage dttaskMessage = new DttaskMessage();dttaskMessage.setType(VOT_RESP);dttaskMessage.setInfo(JSON.toJSONString(new VotRespMessage(serverId)));return dttaskMessage;}public static DttaskMessage buildVotRequestMessage(Long lastControllerServerId, long fromServerId, long serverId, int version) {DttaskMessage dttaskMessage = new DttaskMessage();dttaskMessage.setType(VOTING);dttaskMessage.setInfo(JSON.toJSONString(new VotRequestMessage(lastControllerServerId, fromServerId, serverId, version)));return dttaskMessage;}}

1.2.3 VotingMessageService

这里以VotingMessageService为例，它实现了IMessageService接口，完成投票信息的处理。其它的*MessageService也是同样的，这里就不一一举例了，可以在 com.swsm.dttask.server.service.message包下查看

@Slf4j
@Component
public class VotingMessageService implements IMessageService {@Overridepublic byte getMessageType() {return DttaskMessage.VOTING;}@Overridepublic void execute(ChannelHandlerContext ctx, DttaskMessage message) {Channel channel = ctx.channel();VotRequestMessage votRequestMessage = JSON.parseObject(message.getInfo(), VotRequestMessage.class);long fromServerId = votRequestMessage.getFromServerId();Long lastControllerServerId = votRequestMessage.getLastControllerServerId();ServerInfo.addOtherNode(fromServerId, channel);boolean addRes = ServerInfo.addVotResult(votRequestMessage.getVersion(), votRequestMessage.getServerId());if (!addRes) {log.info("丢弃以前版本的投票信息={}", votRequestMessage);return;}// 归票VotResult votResult = ServerInfo.getVotResult();Map<Long, Integer> votMap = votResult.getVotMap();for (Map.Entry<Long, Integer> entry : votMap.entrySet()) {long controllerServerId = entry.getKey();if (votMap.get(controllerServerId) >= ServerInfo.getVotMax()) {// 归票成功log.info("本节点={}是controller", controllerServerId);ServerInfo.setStatus(ServerStatus.RUNNING);Controller controller = ServerInfo.initController();for (Long otherServerId : ServerInfo.getOtherNodeIds()) {Channel otherNodeChannel = ServerInfo.getChannelByServerId(otherServerId);otherNodeChannel.writeAndFlush(DttaskMessage.buildVotRespMessage(controllerServerId));}return;}}}
}

1.2.4 MessageServiceManager

Spring托管的bean，里面会将所有实现了IMessageService接口的类都管理起来，并在消息到来时进行选择

@Slf4j
@Component
public class MessageServiceManager {@Autowired(required = false)private List<IMessageService> messageServices;private Map<Byte, IMessageService> messageServiceMap = new HashMap<>();@PostConstructpublic void init() {if (messageServices != null) {for (IMessageService messageService : messageServices) {messageServiceMap.put(messageService.getMessageType(), messageService);}}}public IMessageService chooseMessageService(byte messageType) {if (messageServiceMap.containsKey(messageType)) {return messageServiceMap.get(messageType);}return messageServiceMap.get(Byte.MIN_VALUE);}}

1.2.5 借助netty实现节点与节点直接的心跳

@Slf4j
public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter {  @Override  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {  if (evt instanceof IdleStateEvent) {  IdleStateEvent event = (IdleStateEvent) evt;  if (event.state() == IdleState.READER_IDLE) {  log.warn("读取空闲...");} else if (event.state() == IdleState.WRITER_IDLE) {log.warn("写入空闲...");} else if (event.state() == IdleState.ALL_IDLE) {Long serverId = ServerInfo.getServerIdByChannelId(ctx.channel().id());log.warn("serverId={}与server通信读取或写入空闲...", serverId);if (serverId != null) {ctx.writeAndFlush(DttaskMessage.buildPingMessage(serverId));} else {ctx.close();}}}  }  
}

1.2.6 ServerInfo类完成所有信息的管理

ServerInfo类完成所有信息的管理，它里面会存储很多系统运行需要的数据如：当前节点信息(myNodeInfo)、Channel和节点的关系(nodeChannelMap、nodeChannelServerIdMap)、系统状态(status)、其它节点信息(otherNodeInfoMap)；

ServerInfo还肩负着确定节点角色(Controller或Follower)以及初始化角色的任务

@Slf4j
public class ServerInfo {private ServerInfo() {}private static NioEventLoopGroup bossGroup;public static void setBossGroup(NioEventLoopGroup bg) {bossGroup = bg;}public static NioEventLoopGroup getBossGroup() {return bossGroup;}private static NioEventLoopGroup workerGroup;private static Channel serverChannel;private static Bootstrap connectOtherNodeBootStrap;private static ServerBootstrap bootstrapForClient;private static NodeInfo myNodeInfo;private static Map<Long, Channel> nodeChannelMap = new ConcurrentHashMap<>();private static Map<ChannelId, Long> nodeChannelServerIdMap = new ConcurrentHashMap<>();private static volatile ServerStatus status;private static Map<Long, NodeInfo> otherNodeInfoMap = new ConcurrentHashMap<>();private static VotResult votResult = new VotResult();private static Controller controller;private static Follower follower;public static void init() {RedisUtil redisUtil = BeanUseHelper.redisUtil();DttaskServerConfig dttaskServerConfig = BeanUseHelper.dttaskServerConfig();long localServerId = dttaskServerConfig.getServerId();ServerInfo.setStatus(ServerStatus.STARTING);Long controllerServerId = redisUtil.getLongValue(Constant.RedisConstants.DTTASK_CONTROLLER);if (controllerServerId == null) {log.info("当前启动状态为：未确定controller");initNodeInfoByConfig();Long minServerId = ServerInfo.getMinNodeId();if (minServerId == localServerId) {log.info("就当前一个节点:{},此节点就是controller", localServerId);ServerInfo.setStatus(ServerStatus.RUNNING);} else {log.info("有多个节点，节点状态应为VOTING");ServerInfo.setStatus(ServerStatus.VOTING);}} else {log.info("当前启动状态为：已确定controller");ServerInfo.refreshNodeInfoByRedis();InetSocketAddress address = dttaskServerConfig.getServerInfoMap().get(localServerId);ServerInfo.setMyNodeInfo(localServerId, address.getHostString(), address.getPort(), null);ServerInfo.setStatus(ServerStatus.IDENTIFYING);}}private static void initNodeInfoByConfig() {DttaskServerConfig dttaskServerConfig = BeanUseHelper.dttaskServerConfig();long localServerId = dttaskServerConfig.getServerId();Map<Long, InetSocketAddress> serverInfoMap = dttaskServerConfig.getServerInfoMap();for (Map.Entry<Long, InetSocketAddress> entry : serverInfoMap.entrySet()) {long id = entry.getKey();InetSocketAddress address = serverInfoMap.get(id);if (localServerId != id) {ServerInfo.addOtherNode(id, address.getHostString(), address.getPort());} else {ServerInfo.setMyNodeInfo(localServerId, address.getHostString(), address.getPort(), null);}}}public static Controller initController() {long localServerId = ServerInfo.getServerId();RedisUtil redisUtil = BeanUseHelper.redisUtil();log.info("初始化本节点={}controller信息...", ServerInfo.getServerId());ServerInfo.setRole(ServerRole.CONTROLLER);redisUtil.setCacheObject(Constant.RedisConstants.DTTASK_CONTROLLER, localServerId);ServerInfo.setOtherNodeRole(localServerId);ServerInfo.refreshRedisNodeInfo();controller = Controller.getInstance();return controller;}public static Follower initFollower() {log.info("初始化本节点={}follower信息...", ServerInfo.getServerId());RedisUtil redisUtil = BeanUseHelper.redisUtil();Long controllerServerId = redisUtil.getLongValue(Constant.RedisConstants.DTTASK_CONTROLLER);if (controllerServerId == null) {log.error("init follower时，controller还没有确定...");throw new BusinessException("init follower时，controller还没有确定...");}ServerInfo.setRole(ServerRole.FOLLOWER);ServerInfo.setStatus(ServerStatus.RUNNING);ServerInfo.setOtherNodeRole(controllerServerId);follower = Follower.getInstance();return follower;}public static void setMyNodeInfo(long serverId, String ip, int port, ServerRole serverRole) {NodeInfo nodeInfo = new NodeInfo();nodeInfo.setServerId(serverId);nodeInfo.setIp(ip);nodeInfo.setPort(port);nodeInfo.setServerRole(serverRole);myNodeInfo = nodeInfo;}public static long getServerId() {return myNodeInfo.getServerId();}public static NodeInfo getMyNodeInfo() {return myNodeInfo;}public static Map<Long, NodeInfo> getOtherNodeInfoMap() {return otherNodeInfoMap;}public static int getVotMax() {return otherNodeInfoMap.size();}public static VotResult getVotResult() {return votResult;}public static synchronized void cacheChannelAnsServerIdRel(long serverId, Channel channel) {nodeChannelMap.put(serverId, channel);nodeChannelServerIdMap.put(channel.id(), serverId);}public static Channel getChannelByServerId(Long serverId) {return nodeChannelMap.get(serverId);}public static synchronized void removeChannel(ChannelId channelId) {Long serverId = nodeChannelServerIdMap.get(channelId);if (serverId != null) {log.info("删除和节点id={}的连接", serverId);nodeChannelServerIdMap.remove(channelId);nodeChannelMap.remove(serverId);otherNodeInfoMap.remove(serverId);}}public static synchronized void refreshRedisNodeInfo() {if (myNodeInfo != null && getRole() != null && getRole().isController()) {RedisUtil redisUtil = BeanUseHelper.redisUtil();Map<Long, NodeInfo> otherNodeInfoMap = ServerInfo.getOtherNodeInfoMap();List<NodeInfo> nodeInfoList = new ArrayList<>();nodeInfoList.addAll(otherNodeInfoMap.values());nodeInfoList.add(ServerInfo.getMyNodeInfo());log.info("controller刷新节点信息到redis:{}", nodeInfoList);redisUtil.setCacheObject(Constant.RedisConstants.DTTASK_NODE_INFO, JSON.toJSONString(nodeInfoList));}}public static Long getServerIdByChannelId(ChannelId channelId) {return nodeChannelServerIdMap.get(channelId);}public static synchronized boolean addVotResult(int version, long chooseServerId) {Integer curVersion = votResult.getVersion();if (version < votResult.getVersion()) {log.info("版本={}已失效,当前的为:{}", version, curVersion);return false;}votResult.setVersion(version);if (votResult.getVotMap().containsKey(chooseServerId)) {votResult.getVotMap().put(chooseServerId, votResult.getVotMap().get(chooseServerId) + 1);} else {votResult.getVotMap().put(chooseServerId, 1);}return true;}public static void addOtherNode(long serverId, String ip, int port) {NodeInfo nodeInfo = new NodeInfo();nodeInfo.setServerId(serverId);nodeInfo.setIp(ip);nodeInfo.setPort(port);otherNodeInfoMap.put(serverId, nodeInfo);}public static void addOtherNode(long serverId, Channel channel) {addOtherNode(serverId, channel, null);}public static void addOtherNode(long serverId, Channel channel, ServerRole serverRole) {InetSocketAddress address = BeanUseHelper.dttaskServerConfig().getServerInfoMap().get(serverId);if (address == null) {throw new BusinessException(CharSequenceUtil.format("id={}的没有配置在文件中", serverId));}if (channel != null && ServerInfo.getServerIdByChannelId(channel.id()) == null) {ServerInfo.cacheChannelAnsServerIdRel(serverId, channel);}NodeInfo nodeInfo = new NodeInfo();nodeInfo.setServerId(serverId);nodeInfo.setIp(address.getHostString());nodeInfo.setPort(address.getPort());nodeInfo.setServerRole(serverRole);otherNodeInfoMap.put(serverId, nodeInfo);}public static NodeInfo getNodeInfo(long serverId) {return otherNodeInfoMap.get(serverId);}public static Set<Long> getOtherNodeIds() {return otherNodeInfoMap.keySet();}public static long getMinNodeId() {if (getOtherNodeIds().isEmpty()) {return ServerInfo.getServerId();}return Collections.min(getOtherNodeIds());}public static void setOtherNodeRole(long controllerServerId) {for (Map.Entry<Long, NodeInfo> entry : otherNodeInfoMap.entrySet()) {long serverId = entry.getKey();if (serverId == controllerServerId) {otherNodeInfoMap.get(serverId).setServerRole(ServerRole.CONTROLLER);} else {otherNodeInfoMap.get(serverId).setServerRole(ServerRole.FOLLOWER);}}}public static ServerRole getRole() {return myNodeInfo.getServerRole();}public static void setStatus(ServerStatus s) {status = s;}public static void setRole(ServerRole r) {myNodeInfo.setServerRole(r);}public static Set<Channel> getOtherNodeChannel(Long serverId) {Set<Channel> res = new HashSet<>();for (Map.Entry<Long, Channel> entry : nodeChannelMap.entrySet()) {long id = entry.getKey();if (!Objects.equals(serverId, id)) {res.add(nodeChannelMap.get(id));}}return res;}/*** 这个方法针对，本节点并没有和要断开节点有连接的* @param offlineServerId 掉线的节点id*/public static void removeNode(long offlineServerId) {log.info("删除掉线节点id={}", offlineServerId);otherNodeInfoMap.remove(offlineServerId);}public static void refreshNodeInfoByRedis() {RedisUtil redisUtil = BeanUseHelper.redisUtil();Object obj = redisUtil.getCacheObject(Constant.RedisConstants.DTTASK_NODE_INFO);if (obj != null) {List<NodeInfo> nodeInfoList = JSON.parseObject(obj.toString(),new TypeReference<List<NodeInfo>>() {}.getType());for (NodeInfo nodeInfo : nodeInfoList) {otherNodeInfoMap.put(nodeInfo.getServerId(), nodeInfo);}}}public static boolean isIdentifying() {return status.isIdentifying();}public static boolean isVoting() {return status.isVoting();}public static boolean isRunning() {return status.isRunning();}public static void setWorkerGroup(NioEventLoopGroup bg) {workerGroup = bg;}public static NioEventLoopGroup getWorkerGroup() {return workerGroup;}public static void setServerChannel(Channel ch) {serverChannel = ch;}public static Channel getServerChannel() {return serverChannel;}public static void setConnectOtherNodeBootStrap(Bootstrap bs) {connectOtherNodeBootStrap = bs;}public static Bootstrap getConnectOtherNodeBootStrap() {return connectOtherNodeBootStrap;}public static void setBootstrapForClient(ServerBootstrap sbs) {bootstrapForClient = sbs;}public static ServerBootstrap getBootstrapForClient() {return bootstrapForClient;}}

1.2.7 SpringInitRunner -- 增加选举的逻辑

SpringInitRunner前面只启动了netty等待连接，这里将完成选举，以及当就自己一个节点则就认为自己是controller

@Component
@Slf4j
public class SpringInitRunner implements CommandLineRunner {@Autowiredprivate DttaskServerConfig dttaskServerConfig;@Autowiredprivate NetworkService networkService;@Autowiredprivate MessageServiceManager messageServiceManager;@Autowiredprivate RedisUtil redisUtil;@PostConstructpublic void init() {initServerBootStrap();initConnectOtherNodeBootStrap();}private void initConnectOtherNodeBootStrap() {ServerInfo.setConnectOtherNodeBootStrap(new Bootstrap());ServerInfo.getConnectOtherNodeBootStrap().group(new NioEventLoopGroup(4)).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) {socketChannel.pipeline().addLast(new DttaskMessageDecoder(MESSAGE_MAX_SIZE, MESSAGE_LENGTH_FILED_OFFSET, MESSAGE_LENGTH_FILED_LENGTH));socketChannel.pipeline().addLast(new IdleStateHandler(dttaskServerConfig.getReadIdleSecondTime(),dttaskServerConfig.getWriteIdleSecondTime(),dttaskServerConfig.getAllIdleSecondTime()));socketChannel.pipeline().addLast(new DttaskMessageEncoder());socketChannel.pipeline().addLast(new ServerClientChannelHandler(networkService, redisUtil, messageServiceManager));}});}private void initServerBootStrap() {ServerInfo.setBossGroup(new NioEventLoopGroup(4));ServerInfo.setWorkerGroup(new NioEventLoopGroup(8));ServerInfo.setBootstrapForClient(new ServerBootstrap());ServerInfo.getBootstrapForClient().group(ServerInfo.getBossGroup(), ServerInfo.getWorkerGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) {socketChannel.pipeline().addLast(new DttaskMessageDecoder(MESSAGE_MAX_SIZE, MESSAGE_LENGTH_FILED_OFFSET, MESSAGE_LENGTH_FILED_LENGTH));socketChannel.pipeline().addLast(new DttaskMessageEncoder());IdleStateHandler idleStateHandler = new IdleStateHandler(dttaskServerConfig.getReadIdleSecondTime(), dttaskServerConfig.getWriteIdleSecondTime(), dttaskServerConfig.getAllIdleSecondTime());socketChannel.pipeline().addLast(idleStateHandler);socketChannel.pipeline().addLast(new HeartBeatServerHandler());socketChannel.pipeline().addLast(new ServerClientChannelHandler(networkService, redisUtil, messageServiceManager));}});}@Overridepublic void run(String... args) {log.info("spring启动完成，接下来启动 netty");ServerInfo.init();try {log.info("启动监听其它节点端请求的服务端...");ServerInfo.setServerChannel(ServerInfo.getBootstrapForClient().bind(dttaskServerConfig.listenerPort()).sync().channel());} catch (Exception e) {log.error("启动 监听其它节点请求的服务端出现异常", e);System.exit(-1);}try {log.info("连接controller或开始vote...");if (ServerInfo.isIdentifying()) {log.info("连接controller...");RedisUtil redisUtil = BeanUseHelper.redisUtil();long controllerServerId = redisUtil.getLongValue(Constant.RedisConstants.DTTASK_CONTROLLER);networkService.connectController(controllerServerId);} else if (ServerInfo.isVoting()){log.info("开始vote...");long minNodeId = ServerInfo.getMinNodeId();networkService.startVote(null, minNodeId);} else if (ServerInfo.isRunning()) {log.info("已确认本节点={}就是controller...", ServerInfo.getServerId());Controller controller = ServerInfo.initController();}}  catch (Exception e) {log.error("连接controller或开始vote出现异常", e);System.exit(-1);}log.info("netty 启动成功...");}@PreDestroypublic void shutdown() {if (ServerInfo.getOtherNodeIds().isEmpty()) {redisUtil.setCacheObject(Constant.RedisConstants.DTTASK_CONTROLLER, null);redisUtil.setCacheObject(Constant.RedisConstants.DTTASK_NODE_INFO, null);}try {ServerInfo.getServerChannel().close().sync();} catch (InterruptedException e) {log.error("dttask-server netty shutdown 出现异常", e);Thread.currentThread().interrupt();} finally {ServerInfo.getWorkerGroup().shutdownGracefully();ServerInfo.getBossGroup().shutdownGracefully();}}
}

1.2.8 ServerClientChannelHandler -- 完成投票等消息处理

前面ServerClientChannelHandler只是完成了基本框架，代码较少，现在要在这里添加处理每个消息的逻辑，放心，代码也不多，因为我们已经对消息处理进行了拆分，只要加入一个策略选择就可以。

这里不得不感叹一下：策略模式的功能，否则这里将会有一大堆if else。

当节点与节点的通信断开时，会触发channelInactive和exceptionCaught方法，我们需要在这里处理断开的业务逻辑，注意断开的业务逻辑需要判断断开的是Controller还是Follower，这里的处理逻辑不同。

@Slf4j
public class ServerClientChannelHandler extends SimpleChannelInboundHandler<DttaskMessage> {private NetworkService networkService;private MessageServiceManager messageServiceManager;private RedisUtil redisUtil;public ServerClientChannelHandler(NetworkService networkService, RedisUtil redisUtil, MessageServiceManager messageServiceManager) {super();this.networkService = networkService;this.redisUtil = redisUtil;this.messageServiceManager = messageServiceManager;}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {log.info("channelActive={}", ctx.channel().id());}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, DttaskMessage message) throws Exception {log.info("收到客户端的请求:{}", message);messageServiceManager.chooseMessageService(message.getType()).execute(ctx, message);}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {log.warn("channelInactive...");stopChannel(ctx, null);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {log.warn("exceptionCaught...", cause);stopChannel(ctx, cause);}private void stopChannel(ChannelHandlerContext ctx, Throwable cause) {Channel channel = ctx.channel();long localServerId = ServerInfo.getServerId();Long serverId = ServerInfo.getServerIdByChannelId(ctx.channel().id());if (serverId == null) {return;}if (cause != null) {log.error("nodeId={}与本节点id={}通信出现异常", serverId, localServerId, cause);} else {log.error("nodeId={}与本节点id={}通信失效", serverId, localServerId);}if (channel.isActive()) {channel.close();}// 判断下线的是follower 还是 controllerNodeInfo nodeInfo = ServerInfo.getNodeInfo(serverId);if (!nodeInfo.getServerRole().isController()) {log.info("下线的是follower,id={}", serverId);Set<Channel> otherNodeChannels = ServerInfo.getOtherNodeChannel(serverId);for (Channel otherNodeChannel : otherNodeChannels) {otherNodeChannel.writeAndFlush(DttaskMessage.buildNodeOfflineMessage(serverId));}ServerInfo.removeChannel(channel.id());ServerInfo.refreshRedisNodeInfo();} else {log.info("下线的是controller,id={}", serverId);redisUtil.setCacheObject(Constant.RedisConstants.DTTASK_CONTROLLER, null);ServerInfo.removeChannel(channel.id());long minNodeId = ServerInfo.getMinNodeId();if (minNodeId != localServerId) {// 重新选举networkService.startVote(serverId, minNodeId);} else {// 当前就只剩自己一个节点ServerInfo.setStatus(ServerStatus.RUNNING);Controller controller = ServerInfo.initController();}ServerInfo.refreshRedisNodeInfo();}}
}

2. 验证

2.1 建立3个节点的配置

注意：Server.port需要是不一样的

注意：节点的serverId也要和serverInfo匹配

2.2 idea建立针对3个配置的启动Service

2.3 依次启动验证

可以按照本文最起那面的 节点详细交互图的 步骤进行测试。

• 确保redis，mysql都已ok
• 依次启动3个节点，可以看到1为controller，2 3为follower，redis的key --- 完成选举

• 下线3号节点

• 下线1号节点

2号节点称为Controller

• 上线1号节点、3号节点

所有节点启动完成，这时2是C、1 3是F

• 停止2号节点

至此完成了所有验证