Netty-实现心跳机制-以简易手写RPC框架为例

Netty-实现心跳机制-以简易手写RPC框架为例

所谓心跳机制，其本质作用就是保证服务端与客户端的连接的可用性。

当客户端出现宕机，或因网络故障无法连接工作时。服务端应该及时发现，并将该连接断开，避免服务端或重启后的客户端仍使用这个连接读写数据，造成异常。

为了实现这个功能，我们可以在客户端，一段事件没有向服务端写数据时，向服务端发送一个心跳包。服务端接收到该心跳包，则表示当前客户端仍在线。当服务端在一段事件后没有获取到来自客户端的心跳包时，则可判定客户端掉线，将与该客户端的连接断开。

Netty中为我们提供了IdleStateHandler类，我们可以用它来计时监听channel当中的读写事件

 public IdleStateHandler(
            long readerIdleTime, long writerIdleTime, long allIdleTime,
            TimeUnit unit) {
        this(false, readerIdleTime, writerIdleTime, allIdleTime, unit);
    }

readerIdleTime读空闲超时时间设定，如果channelRead()方法超过readerIdleTime时间未被调用则会触发超时事件调用userEventTrigger()方法；

writerIdleTime写空闲超时时间设定，如果write()方法超过writerIdleTime时间未被调用则会触发超时事件调用userEventTrigger()方法；

allIdleTime所有类型的空闲超时时间设定，包括读空闲和写空闲；

unit时间单位，包括时分秒等，默认为秒；

所以有了上述前提，我们就可以在代码实现

NettyServer:

ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .handler(new LoggingHandler(LogLevel.INFO))
        .option(ChannelOption.SO_BACKLOG, 256)
        // 当设置为true的时候，TCP会实现监控连接是否有效
        .option(ChannelOption.SO_KEEPALIVE, true)
        .childOption(ChannelOption.TCP_NODELAY, true)
        .childHandler(new ChannelInitializer() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new CommonEncoder(serializer));
                pipeline.addLast(new CommonDecoder());
                // 服务端30秒没有读事件，关闭连接
                pipeline.addLast(new IdleStateHandler(30, 0, 0));
                pipeline.addLast(new NettyServerHandler());
            }
        });
ChannelFuture future = serverBootstrap.bind(port).sync();
future.channel().closeFuture().sync();

NettyServerHandler:重写userEventTrigger()方法

@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) throws Exception {
    try {
        // 收到心跳
        if(msg.getHeartBeat() != null && msg.getHeartBeat() == true) {
            logger.info("read idle happen: [{}]", ctx.channel().remoteAddress());
            return;
        }
        ...
        }
    } finally {
        ...
    }
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    // 如果一段时间未监听到心跳包，说明客户端掉线，断开连接
    if (evt instanceof IdleStateEvent) {
        IdleState state = ((IdleStateEvent) evt).state();
        if (state == IdleState.READER_IDLE) {
            logger.info("长时间未收到心跳包，断开连接:{}", ctx.channel().toString());
            ctx.close();
        }
    } else {
        super.userEventTriggered(ctx, evt);
    }
}

NettyClient：

bootstrap.handler(new ChannelInitializer() {
    @Override
    protected void initChannel(SocketChannel ch) {
        
        // RpcResponse -> ByteBuf
        ch.pipeline().addLast(new CommonEncoder(serializer))
                // 客户端5秒后没有写 *** 作，发送心跳
                .addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS))
                .addLast(new CommonDecoder())
                .addLast(new NettyClientHandler());
    }
});

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof IdleStateEvent) {
        IdleState state = ((IdleStateEvent) evt).state();
        if (state == IdleState.WRITER_IDLE) {
            logger.info("write idle happen: [{}]", ctx.channel().toString());
            RpcRequest rpcRequest = new RpcRequest();
            rpcRequest.setHeartBeat(true);
     		       ctx.channel().writeAndFlush(rpcRequest).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
        }
    } else {
        super.userEventTriggered(ctx, evt);
    }
}

另外，当服务端出现宕机，或因网络故障无法连接工作时。客户端应该及时发现，并重新建立连接。所以我们需要在客户端处理此类事件，Netty的ChannelInboundHandler类中为我们提供了channelInactive方法，我们只需要在客户端重写该方法，实现重连逻辑即可。

channelInactive方法会在服务端与客户端的连接断开时，自动被回调。