其他分享
首页 > 其他分享> > 从零开始实现简单 RPC 框架 9:网络通信之心跳与重连机制

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

作者:互联网

一、心跳

什么是心跳

在 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) {
}

参数解析:

2. 事件处理

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

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

标签:void,ctx,RPC,从零开始,空闲,new,重连,public,channel
来源: https://blog.csdn.net/idesireccx/article/details/120123709