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 {

    @Override
    public byte getMessageType() {
        return DttaskMessage.VOTING;
    }

    @Override
    public 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<>();
    
    @PostConstruct
    public 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 {

    @Autowired
    private DttaskServerConfig dttaskServerConfig;
    @Autowired
    private NetworkService networkService;
    @Autowired
    private MessageServiceManager messageServiceManager;
    @Autowired
    private RedisUtil redisUtil;

    @PostConstruct
    public void init() {
        initServerBootStrap();
        initConnectOtherNodeBootStrap();
    }

    private void initConnectOtherNodeBootStrap() {
        ServerInfo.setConnectOtherNodeBootStrap(new Bootstrap());
        ServerInfo.getConnectOtherNodeBootStrap().group(new NioEventLoopGroup(4))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected 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>() {
                    @Override
                    protected 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));
                    }
                });
    }

    @Override
    public 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 启动成功...");
    }

    @PreDestroy
    public 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;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("channelActive={}", ctx.channel().id());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DttaskMessage message) throws Exception {
        log.info("收到客户端的请求:{}", message);
        messageServiceManager.chooseMessageService(message.getType()).execute(ctx, message);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.warn("channelInactive...");
        stopChannel(ctx, null);
    }

    @Override
    public 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 还是 controller
        NodeInfo 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号节点

至此完成了所有验证