当前位置: 首页 > news >正文

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号节点

至此完成了所有验证

相关文章:

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

1.设计 前面已经完成了netty的集成&#xff0c;接下来就是借助netty完成选举就行了。 针对选举&#xff0c;我们用到了VotRequestMessage、VotRespMessage、当节点下线时NodeOfflineMessage、NodeOnlineMessage、NodeOnlineRespMessage 1.1 节点详细的交互 1.2 对所有消息的…...

服装管理系统 简单实现

服装管理系统 项目使用jsp servletmysql实现&#xff1b; 登陆注册 首页 首页显示服装信息 服装管理 1添加服装 2修改服装 3分页查询服装 4导出服装信息 5 导入服装信息 代码结构截图 百度网盘 链接&#xff1a;https://pan.baidu.com/s/1zfLHGMnrYd-JtnhzS5elYQ 提取码…...

深度学习项目实战:垃圾分类系统

简介&#xff1a; 今天开启深度学习另一板块。就是计算机视觉方向&#xff0c;这里主要讨论图像分类任务–垃圾分类系统。其实这个项目早在19年的时候&#xff0c;我就写好了一个版本了。之前使用的是python搭建深度学习网络&#xff0c;然后前后端交互的采用的是java spring …...

C#浅拷贝和深拷贝数据

目录 一、浅拷贝 二、深拷贝 一、浅拷贝 就是把原来的数据&#xff0c;复制一份&#xff0c;但是2份数据是共享地址的&#xff0c;修改第一份数据或者修改第二份数据&#xff0c;都会一起改变&#xff0c;这可能不是我们程序中需要的场景。 下面我们演示一下&#xff0c;首…...

【JVM】4.运行时数据区(程序计数器、虚拟机栈)

文章目录 4.JVM的运行时数据区4.1 程序计数器4.2 Java虚拟机栈4.3 虚拟机栈内存溢出 4.JVM的运行时数据区 4.1 程序计数器 程序计数器&#xff08;PC&#xff09;会记录着下一行字节码指令的地址。执行完当前指令后&#xff0c;PC刷新&#xff0c;JVM的执行引擎根据程序计数器…...

算法:程序员的数学读书笔记

目录 ​0的故事 ​一、按位计数法 二、不使用按位计数法的罗马数字 三、十进制转二进制​​​​​​​ ​四、0所起到的作用​​​​​​​ 逻辑 一、为何逻辑如此重要 二、兼顾完整性和排他性 三、逻辑 四、德摩根定律 五、真值表 六、文氏图 七、卡诺图 八、逻…...

机器学习算法---时间序列

类别内容导航机器学习机器学习算法应用场景与评价指标机器学习算法—分类机器学习算法—回归机器学习算法—聚类机器学习算法—异常检测机器学习算法—时间序列数据可视化数据可视化—折线图数据可视化—箱线图数据可视化—柱状图数据可视化—饼图、环形图、雷达图统计学检验箱…...

RK3568/RV1126/RV1109/RV1106 ISP调试方案

最近一直在做瑞芯微rv1126的开发&#xff0c;由于项目性质&#xff0c;与camera打的交道比较多&#xff0c;包括图像的采集&#xff0c;ISP处理&#xff0c;图像处理&#xff0c;H.264/H.265编解码等各个方面吧。学到了不少&#xff0c;在学习的过程中&#xff0c;也得到了不少…...

【TB作品】51单片机,语音出租车计价器

西交大题目 1.语音出租车计价器 一、功能要求: 1.具有可模拟出租车车轮转速传感器的硬件设计,可计量出租车所走的公 里数。 2.显示和语音播报里程、价格和等待红灯或堵车的计时价格: 3.具有等待计时功能 4.具有实时年月日显示和切换功能。 5.操作简单、界面友好。 二、设计建议…...

jmeter简单压测kafka

前言 这也是一个笔记&#xff0c;就是计划用jmeter做性能测试&#xff0c;但是这里是只要将数据放到kafka的topic里&#xff0c;后面查看下游业务处理能力。 一、方案 因为只要实现数据放到kafka&#xff0c;参考了下博友的方案&#xff0c;可行。 二、方案验证 详细过程就不…...

【漏洞复现】红帆OA iorepsavexml.aspx文件上传漏洞

漏洞描述 广州红帆科技深耕医疗行业20余年,专注医院行政管控,与企业微信、阿里钉钉全方位结合,推出web移动一体化办公解决方案——iOffice20(医微云)。提供行政办公、专业科室应用、决策辅助等信息化工具,采取平台化管理模式,取代医疗机构过往多系统分散式管理,实现医…...

04_Web框架之Django一

Web框架之Django一 学习目标和内容 1、能够描述Django的作用 2、能够使用Django创建应用 3、能够使用GET和POST请求方式进行传参 4、能够使用Django的函数式方法定义视图 5、能够进行Django的配置文件修改 6、能够基本使用Django的路由定义 一、Django相关介绍 1、什么是Djan…...

单机架构到分布式架构的演变

目录 1.单机架构 2.应用数据分离架构 3.应用服务集群架构 4.读写分离 / 主从分离架构 5.引入缓存 —— 冷热分离架构 6.垂直分库 7.业务拆分 —— 微服务 8.容器化引入——容器编排架构 总结 1.单机架构 初期&#xff0c;我们需要利用我们精干的技术团队&#xff0c;快…...

1.新入手的32位单片机资源和资料总览

前言&#xff1a; 学了将近1年的linux驱动和uboot&#xff0c;感觉反馈不足&#xff0c;主要是一直在学各种框架&#xff0c;而且也遇到了门槛&#xff0c;比如驱动部分&#xff0c;还不能随心所欲地编程&#xff0c;原因是有些外设的原理还不够深刻、有些复杂的底层驱动的代码…...

jmeter判断’响应断言‘两个变量对象是否相等

1、首先需要设置变量&#xff0c;json、正则、csv文件等变量 2、然后在响应断言中 ①JMeter Variable Name to use —— 输入一个变量&#xff0c;变量名即可 ② 模式匹配规则 ——相等 ③测试模式 ——输入引用的变量命${变量名} &#xff08;注意这里是需要添加一个测试模式…...

【Linux基础命令使用】

文章目录 一. 操作系统和文件及文件路径介绍二. 基础指令介绍三. 结束语 一. 操作系统和文件及文件路径介绍 什么是操作系统&#xff1f;操作系统是一款进行软硬件资源管理的软件为什么要进行软硬件资源管理&#xff1f;对上提供良好的稳定的运行服务----工具Linux指令和图形化…...

【JNA与C++基本使用示例】

JNA中java与C使用注意事项和代码示例 JNA关系映射表使用案列注意代码示例C代码java代码 JNA关系映射表 使用案列 注意 JNA只支持C方式的dll使用C的char* 作为返回值时&#xff0c;需要返回的变量为malloc分配的地址C的strlen函数只获得除/0以外的字符串长度 代码示例 C代码…...

HttpRunner接口自动化测试框架

简介 HttpRunner是一款面向 HTTP(S) 协议的通用测试框架&#xff0c;只需编写维护一份 YAML/JSON 脚本&#xff0c;即可实现自动化测试、性能测试、线上监控、持续集成等多种测试需求。 项目地址&#xff1a;GitHub - httprunner/httprunner: HttpRunner 是一个开源的 API/UI…...

云计算:Vmware 安装 FreeNAS

目录 一、实验 1.Vmware 安装 FreeNAS 2.配置Web界面 二、问题 1.iSCSI如何限定名称 2.LUN和LVM的区别 一、实验 1.Vmware 安装 FreeNAS &#xff08;1&#xff09;环境准备 VMware Workstation 17 FreeNAS相关安装部署镜像: 官网地址&#xff1a; https://download…...

数据库交付运维高级工程师-腾讯云TDSQL

数据库交付运维高级工程师-腾讯云TDSQL上机指导&#xff0c;付费指导&#xff0c;暂定99...

Linux 文件类型,目录与路径,文件与目录管理

文件类型 后面的字符表示文件类型标志 普通文件&#xff1a;-&#xff08;纯文本文件&#xff0c;二进制文件&#xff0c;数据格式文件&#xff09; 如文本文件、图片、程序文件等。 目录文件&#xff1a;d&#xff08;directory&#xff09; 用来存放其他文件或子目录。 设备…...

java_网络服务相关_gateway_nacos_feign区别联系

1. spring-cloud-starter-gateway 作用&#xff1a;作为微服务架构的网关&#xff0c;统一入口&#xff0c;处理所有外部请求。 核心能力&#xff1a; 路由转发&#xff08;基于路径、服务名等&#xff09;过滤器&#xff08;鉴权、限流、日志、Header 处理&#xff09;支持负…...

8k长序列建模,蛋白质语言模型Prot42仅利用目标蛋白序列即可生成高亲和力结合剂

蛋白质结合剂&#xff08;如抗体、抑制肽&#xff09;在疾病诊断、成像分析及靶向药物递送等关键场景中发挥着不可替代的作用。传统上&#xff0c;高特异性蛋白质结合剂的开发高度依赖噬菌体展示、定向进化等实验技术&#xff0c;但这类方法普遍面临资源消耗巨大、研发周期冗长…...

Cesium1.95中高性能加载1500个点

一、基本方式&#xff1a; 图标使用.png比.svg性能要好 <template><div id"cesiumContainer"></div><div class"toolbar"><button id"resetButton">重新生成点</button><span id"countDisplay&qu…...

Qt Http Server模块功能及架构

Qt Http Server 是 Qt 6.0 中引入的一个新模块&#xff0c;它提供了一个轻量级的 HTTP 服务器实现&#xff0c;主要用于构建基于 HTTP 的应用程序和服务。 功能介绍&#xff1a; 主要功能 HTTP服务器功能&#xff1a; 支持 HTTP/1.1 协议 简单的请求/响应处理模型 支持 GET…...

WEB3全栈开发——面试专业技能点P2智能合约开发(Solidity)

一、Solidity合约开发 下面是 Solidity 合约开发 的概念、代码示例及讲解&#xff0c;适合用作学习或写简历项目背景说明。 &#x1f9e0; 一、概念简介&#xff1a;Solidity 合约开发 Solidity 是一种专门为 以太坊&#xff08;Ethereum&#xff09;平台编写智能合约的高级编…...

前端开发面试题总结-JavaScript篇(一)

文章目录 JavaScript高频问答一、作用域与闭包1.什么是闭包&#xff08;Closure&#xff09;&#xff1f;闭包有什么应用场景和潜在问题&#xff1f;2.解释 JavaScript 的作用域链&#xff08;Scope Chain&#xff09; 二、原型与继承3.原型链是什么&#xff1f;如何实现继承&a…...

自然语言处理——Transformer

自然语言处理——Transformer 自注意力机制多头注意力机制Transformer 虽然循环神经网络可以对具有序列特性的数据非常有效&#xff0c;它能挖掘数据中的时序信息以及语义信息&#xff0c;但是它有一个很大的缺陷——很难并行化。 我们可以考虑用CNN来替代RNN&#xff0c;但是…...

优选算法第十二讲:队列 + 宽搜 优先级队列

优选算法第十二讲&#xff1a;队列 宽搜 && 优先级队列 1.N叉树的层序遍历2.二叉树的锯齿型层序遍历3.二叉树最大宽度4.在每个树行中找最大值5.优先级队列 -- 最后一块石头的重量6.数据流中的第K大元素7.前K个高频单词8.数据流的中位数 1.N叉树的层序遍历 2.二叉树的锯…...

Yolov8 目标检测蒸馏学习记录

yolov8系列模型蒸馏基本流程&#xff0c;代码下载&#xff1a;这里本人提交了一个demo:djdll/Yolov8_Distillation: Yolov8轻量化_蒸馏代码实现 在轻量化模型设计中&#xff0c;**知识蒸馏&#xff08;Knowledge Distillation&#xff09;**被广泛应用&#xff0c;作为提升模型…...