从零开始实现简单 RPC 框架 9：网络通信之心跳与重连机制

2024-02-18 17:48:46

一、心跳

什么是心跳

在 TPC 中，客户端和服务端建立连接之后，需要定期发送数据包，来通知对方自己还在线，以确保 TPC 连接的有效性。如果一个连接长时间没有心跳，需要及时断开，否则服务端会维护很多无用连接，浪费服务端的资源。

IdleStateHandler

Netty 已经为我们提供了心跳的 Handler：IdleStateHandler。当连接的空闲时间（读或者写）太长时，IdleStateHandler 将会触发一个 IdleStateEvent 事件，传递的下一个 Handler。我们可以通过在 Pipeline Handler 中重写 userEventTrigged 方法来处理该事件，注意我们自己的 Handler 需要在 IdleStateHandler 后面。

下面我们来看看 IdleStateHandler 的源码。

1. 构造函数

最完整的构造函数如下：

public IdleStateHandler(boolean observeOutput,
            long readerIdleTime, long writerIdleTime, long allIdleTime,
            TimeUnit unit) {
}

参数解析：

observeOutput：是否考虑出站时较慢的情况。如果 true：当出站时间太长，超过空闲时间，那么将不触发此次事件。如果 false，超过空闲时间就会触发事件。默认 false。
readerIdleTime：读空闲的时间，0 表示禁用读空闲事件。
writerIdleTime：写空闲的时间，0 表示禁用写空闲事件。
allIdleTime：读或写空闲的时间，0 表示禁用事件。
unit：前面三个时间的单位。

2. 事件处理

IdleStateHandler 继承 ChannelDuplexHandler，重写了出站和入站的事件，我们来看看代码。
当 channel 添加、注册、活跃的时候，会初始化 initialize(ctx)，删除、不活跃的时候销毁 destroy()，读写的时候设置 lastReadTime 和 lastWriteTime 字段。

public class IdleStateHandler extends ChannelDuplexHandler {

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isActive() && ctx.channel().isRegistered()) {
            initialize(ctx);
        }
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        destroy();
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isActive()) {
            initialize(ctx);
        }
        super.channelRegistered(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        initialize(ctx);
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        destroy();
        super.channelInactive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 判断是否开启 读空闲 或者 读写空闲 监控
        if (readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
            // 设置 reading 标志位
            reading = true;
            firstReaderIdleEvent = firstAllIdleEvent = true;
        }
        ctx.fireChannelRead(msg);
    }

    // 读完成之后
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 判断是否开启 读空闲 或者 读写空闲 监控，检查 reading 标志位
        if ((readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) && reading) {
            // 设置 lastReadTime，后面判断读超时有用
            lastReadTime = ticksInNanos();
            reading = false;
        }
        ctx.fireChannelReadComplete();
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        // 判断是否开启 写空闲 或者 读写空闲 监控
        if (writerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
            // writeListener 的方法在下面，主要是设置 lastWriteTime
            ctx.write(msg, promise.unvoid()).addListener(writeListener);
        } else {
            ctx.write(msg, promise);
        }
    }
    
    private final ChannelFutureListener writeListener = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            lastWriteTime = ticksInNanos();
            firstWriterIdleEvent = firstAllIdleEvent = true;
        }
    };
}

3. 初始化

当 channel 添加、注册、活跃的时候，会初始化 initialize(ctx)，下面我们就来看看初始化的代码：

private void initialize(ChannelHandlerContext ctx) {
    // Avoid the case where destroy() is called before scheduling timeouts.
    // See: https://github.com/netty/netty/issues/143
    switch (state) {
    case 1:
    case 2:
        return;
    }
    state = 1;
    initOutputChanged(ctx);
    lastReadTime = lastWriteTime = ticksInNanos();
    if (readerIdleTimeNanos > 0) {
        readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
                readerIdleTimeNanos, TimeUnit.NANOSECONDS);
    }
    if (writerIdleTimeNanos > 0) {
        writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
                writerIdleTimeNanos, TimeUnit.NANOSECONDS);
    }
    if (allIdleTimeNanos > 0) {
        allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
                allIdleTimeNanos, TimeUnit.NANOSECONDS);
    }
}

其实初始化很简单，就是根据构造函数给的读写空闲时间去决定初始化哪些定时任务，分别是：ReaderIdleTimeoutTask(读空闲超时任务)、WriterIdleTimeoutTask(写空闲超时任务)、AllIdleTimeoutTask(读写空闲超时任务)。

4. 定时任务

我们来看看 ReaderIdleTimeoutTask，剩下两个的原理跟 ReaderIdleTimeoutTask 差不多，感兴趣的同学自行阅读源码吧。

private final class ReaderIdleTimeoutTask extends AbstractIdleTask {
    ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
        super(ctx);
    }
    @Override
    protected void run(ChannelHandlerContext ctx) {
        // 查看是否超时
        long nextDelay = readerIdleTimeNanos;
        if (!reading) {
            nextDelay -= ticksInNanos() - lastReadTime;
        }
        if (nextDelay <= 0) {
            // 超时了，重新启动一个新的定时器，然后触发事件
            // Reader is idle - set a new timeout and notify the callback.
            readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
            boolean first = firstReaderIdleEvent;
            firstReaderIdleEvent = false;
            try {
                // 构造事件
                IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
                // 触发事件
                channelIdle(ctx, event);
            } catch (Throwable t) {
                ctx.fireExceptionCaught(t);
            }
        } else {
            // 没有超时，设置新的定时器，不过这次的时间是更短的时间
            // Read occurred before the timeout - set a new timeout with shorter delay.
            readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
        }
    }
}

从上面的代码可以看出：
① 如果读空闲超时了，则重新起一个定时器，然后触发事件
② 如果读空闲未超时，则新起一个时间更短(readerIdleTimeNanos - ticksInNanos() - lastReadTime)的定时器

5. 触发事件

上面的触发事件方法是：channelIdle，经过重重代码拨开，其实最终就是调用到了下面的代码：

private void invokeUserEventTriggered(Object event) {
    if (invokeHandler()) {
        try {
            // 触发事件，说白了，就是直接调用 userEventTriggered 方法而已
            ((ChannelInboundHandler) handler()).userEventTriggered(this, event);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    } else {
        fireUserEventTriggered(event);
    }
}

其实触发事件，就是把事件传给下一个 Handler (next)，就是调用 userEventTriggered 方法而已。所以我们处理心跳的 Handler 一定要写到 IdleStateHandler。

ccx-rpc 心跳实现

1. 客户端

IdleStateHandler 放到启动类的 PipleLine 注册上，业务处理器 NettyClientHandler 在其后面。

public class NettyClient {
    // ... 忽略其他代码
    private NettyClient() {
        bootstrap = new Bootstrap()
                // ... 省略其他代码
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        // 设定 IdleStateHandler 心跳检测每 5 秒进行一次写检测
                        // write()方法超过 5 秒没调用，就调用 userEventTrigger
                        p.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
                        // 编码器
                        p.addLast(new RpcMessageEncoder());
                        // 解码器
                        p.addLast(new RpcMessageDecoder());
                        // 业务处理器
                        p.addLast(new NettyClientHandler());
                    }
                });
    }
}

接下来我们来看看 NettyClientHandler 是如何处理心跳事件的：

public class NettyClientHandler extends SimpleChannelInboundHandler<RpcMessage> {
    // ... 忽略其他代码
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            // 根据上面的配置，超过 5 秒没有写请求，会触发 WRITER_IDLE 事件
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) {
                log.info("write idle happen [{}]", ctx.channel().remoteAddress());
                Channel channel = ctx.channel();
                // 触发写空闲事件后，就应该发心跳了。
                // 组装消息
                RpcMessage rpcMessage = new RpcMessage();
                rpcMessage.setSerializeType(SerializeType.PROTOSTUFF.getValue());
                rpcMessage.setCompressTye(CompressType.DUMMY.getValue());
                rpcMessage.setMessageType(MessageType.HEARTBEAT.getValue());
                // 发心跳消息
                channel.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

2. 服务端

同样，服务端的 IdleStateHandler 放到启动类的 PipleLine 注册上，业务处理器 NettyServerHandler 在其后面。

public class NettyServerBootstrap {
    public void start() {
        ServerBootstrap bootstrap = new ServerBootstrap()
                // ... 忽略其他代码
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        // 30 秒之内没有收到客户端请求的话就关闭连接
                        p.addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS));
                        // 编解码器
                        p.addLast(new RpcMessageEncoder());
                        p.addLast(new RpcMessageDecoder());
                        // RPC 消息处理器
                        p.addLast(serviceHandlerGroup, new NettyServerHandler());
                    }
                });
        // ... 忽略其他代码
    }
}

服务端收到超过 30 秒没有读请求的事件后，调用 ctx.close 将连接关闭。
同时，如果收到了客户端发来的心跳消息，直接忽略即可。如果每个心跳都要去响应，会加重服务器的负担的。

NettyServerHandler 的代码如下

public class NettyServerHandler extends SimpleChannelInboundHandler<RpcMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcMessage requestMsg) {
        // 不处理心跳消息
        if (requestMsg.getMessageType() != MessageType.REQUEST.getValue()) {
            return;
        }
    }
    
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        // 处理空闲状态的
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                log.info("idle check happen, so close the connection");
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

二、重连机制

很多时候服务端和客户端连接断开，仅仅是因为网络问题或者处理程序慢，并不是程序挂了。那么客户端想再发起请求，就发不出去了。此时需要一个功能：当发现连接断了之后，如果想往连接写数据，就自动重新连接上，这个就是重连机制。

客户端想请求服务端的接口，先从注册中心中，获得服务端的地址，然后跟服务端连接，然后写数据。
简单代码如下：

protected RpcResult doInvoke(RpcRequest request, URL selected) throws RpcException {
    // ... 忽略其他代码
    // 服务端地址
    InetSocketAddress socketAddress = new InetSocketAddress(selected.getHost(), selected.getPort());
    // 获取连接(Channel)
    Channel channel = nettyClient.getChannel(socketAddress);
    // 构建消息
    RpcMessage rpcMessage = buildRpcMessage(request);
    // 写消息(发请求)
    channel.writeAndFlush(rpcMessage);
}

这个 nettyClient.getChannel(socketAddress) 是重连机制的秘密：

/**
 * 获取和指定地址连接的 channel，如果获取不到，则连接
 *
 * @param address 指定要连接的地址
 * @return channel
 */
public Channel getChannel(SocketAddress address) {
    // 根据地址从缓存中获取 Channel
    Channel channel = CHANNEL_MAP.get(address);
    // 如果获取不到，或者 channel 已经断开，则重连，然后放到 CHANNEL_MAP 缓存起来
    if (channel == null || !channel.isActive()) {
        // 连接
        channel = connect(address);
        CHANNEL_MAP.put(address, channel);
    }
    return channel;
}

代码一目了然，就是使用了 CHANNEL_MAP 作为缓存，发现找不到或者已断开，就重新连接，然后放到 CHANNEL_MAP 中，以便下次获取。

总结

心跳是用于服务端和客户端保持有效连接的一种手段，客户端每隔一小段时间发一个心跳包，服务端收到之后不用响应，但是会记下客户端最后一次读的时间。服务器起定时器，定时检测客户端上次读请求的时间超过配置的值，超过就会触发事件，断开连接。
重连机制是连接断开之后，要使用的时候自动重连的机制。

心跳和重连机制，结合起来让服务端和客户端的连接使用更加合理，该断开的断开节省服务端资源，该重连的重连提高可用性。

ccx-rpc 代码已经开源
Github：https://github.com/chenchuxin/ccx-rpc
Gitee：https://gitee.com/imccx/ccx-rpc

码农公寓