netty心跳_优秀的Dubbo心跳设计，你值得学习

前言

谈到RPC肯定绕不开TCP通信&＃xff0c;而主流的RPC框架都依赖于Netty等通信框架&＃xff0c;这时候我们还要考虑是使用长连接还是短连接&＃xff1a;

• 短连接&＃xff1a;每次通信结束后关闭连接&＃xff0c;下次通信需要重新创建连接&＃xff1b;优点就是无需管理连接&＃xff0c;无需保活连接&＃xff1b;
• 长连接&＃xff1a;每次通信结束不关闭连接&＃xff0c;连接可以复用&＃xff0c;保证了性能&＃xff1b;缺点就是连接需要统一管理&＃xff0c;并且需要保活&＃xff1b;

主流的RPC框架都会追求性能选择使用长连接&＃xff0c;所以如何保活连接就是一个重要的话题&＃xff0c;也是本文的主题&＃xff0c;下面会重点介绍一些保活策略&＃xff1b;

为什么需要保活

上面介绍的长连接、短连接并不是TCP提供的功能&＃xff0c;所以长连接是需要应用端自己来实现的&＃xff0c;包括&＃xff1a;连接的统一管理&＃xff0c;如何保活等&＃xff1b;如何保活之前我们了解一下为什么需要保活&＃xff1f;

主要原因是网络不是100%可靠的&＃xff0c;我们创建好的连接可能由于网络原因导致连接已经不可用了&＃xff0c;如果连接一直有消息往来&＃xff0c;那么系统马上可以感知到连接断开&＃xff1b;

但是我们系统可能长时间没有消息来往&＃xff0c;导致系统不能及时感知到连接不可用&＃xff0c;也就是不能及时处理重连或者释放连接&＃xff1b;常见的保活策略使用心跳机制由应用层来实现&＃xff0c;还有网络层提供的TCP Keepalive保活探测机制&＃xff1b;

TCP Keepalive机制

TCP Keepalive是操作系统实现的功能&＃xff0c;并不是TCP协议的一部分&＃xff0c;需要在操作系统下进行相关配置&＃xff0c;开启此功能后&＃xff0c;如果连接在一段时间内没有数据往来&＃xff0c;TCP将发送Keepalive探针来确认连接的可用性&＃xff0c;Keepalive几个内核参数配置&＃xff1a;

• tcp_keepalive_time&＃xff1a;连接多长时间没有数据往来发送探针请求&＃xff0c;默认为7200s(2h)&＃xff1b;
• tcp_keepalive_probes&＃xff1a;探测失败重试的次数默认为10次&＃xff1b;
• tcp_keepalive_intvl&＃xff1a;重试的间隔时间默认75s&＃xff1b;

以上参数可以修改到/etc/sysctl.conf文件中&＃xff1b;是否使用Keepalive用来保活就够了&＃xff0c;其实还不够&＃xff0c;Keepalive只是在网络层就行保活&＃xff0c;如果网络本身没有问题&＃xff0c;但是系统由于其他原因已经不可用了&＃xff0c;这时候Keepalive并不能发现&＃xff1b;所以往往还需要结合心跳机制来一起使用&＃xff1b;

心跳机制

何为心跳机制&＃xff0c;简单来讲就是客户端启动一个定时器用来定时发送请求&＃xff0c;服务端接到请求进行响应&＃xff0c;如果多次没有接受到响应&＃xff0c;那么客户端认为连接已经断开&＃xff0c;可以断开半打开的连接或者进行重连处理&＃xff1b;下面以Dubbo为例来看看是如何具体实施的&＃xff1b;

Dubbo2.6.X

在HeaderExchangeClient中启动了定时器

ScheduledThreadPoolExecutor来定期执行心跳请求&＃xff1a;

ScheduledThreadPoolExecutor scheduled &＃61; new ScheduledThreadPoolExecutor(2,     new NamedThreadFactory("dubbo-remoting-client-heartbeat", true));

在实例化HeaderExchangeClient时启动心跳定时器&＃xff1a;

private void startHeartbeatTimer() {    stopHeartbeatTimer();            if (heartbeat > 0) {        heartbeatTimer &＃61; scheduled.scheduleWithFixedDelay(            new HeartBeatTask(new HeartBeatTask.ChannelProvider() {                                        &＃64;Override                public Collection getChannels() {                                                return Collections.singletonList(HeaderExchangeClient.this);            }        }, heartbeat, heartbeatTimeout),        heartbeat, heartbeat, TimeUnit.MILLISECONDS);    }}

heartbeat默认为60秒&＃xff0c;heartbeatTimeout默认为heartbeat*3&＃xff0c;可以理解至少出现三次心跳请求还未收到回复才会任务连接已经断开&＃xff1b;HeartBeatTask为执行心跳的任务&＃xff1a;

public void run() {    long now &＃61; System.currentTimeMillis();            for (Channel channel : channelProvider.getChannels()) {                    if (channel.isClosed()) {                            continue;        }                    Long lastRead &＃61; (Long) channel.getAttribute(HeaderExchangeHandler.KEY_READ_TIMESTAMP);                    Long lastWrite &＃61; (Long) channel.getAttribute(HeaderExchangeHandler.KEY_WRITE_TIMESTAMP);                    if ((lastRead !&＃61; null && now - lastRead > heartbeat)                    || (lastWrite !&＃61; null && now - lastWrite > heartbeat)) {                            // 发送心跳        }                    if (lastRead !&＃61; null && now - lastRead > heartbeatTimeout) {                            if (channel instanceof Client) {                    ((Client) channel).reconnect();            } else {                channel.close();            }        }    }}

因为Dubbo双端都会发送心跳请求&＃xff0c;所以可以发现有两个时间点分别是&＃xff1a;lastRead和lastWrite&＃xff1b;当然时间和最后读取&＃xff0c;最后写的时间间隔大于heartbeat就会发送心跳请求&＃xff1b;

如果多次心跳未返回结果&＃xff0c;也就是最后读取消息时间大于heartbeatTimeout会判定当前是Client还是Server&＃xff0c;如果是Client会发起reconnect&＃xff0c;Server会关闭连接&＃xff0c;这样的考虑是合理的&＃xff0c;客户端调用是强依赖可用连接的&＃xff0c;而服务端可以等待客户端重新建立连接&＃xff1b;

以上只是介绍的Client&＃xff0c;同样Server端也有相同的心跳处理&＃xff0c;在可以查看HeaderExchangeServer&＃xff1b;

Dubbo2.7.0

Dubbo2.7.0的心跳机制在2.6.X的基础上得到了加强&＃xff0c;同样在HeaderExchangeClient中使用HashedWheelTimer开启心跳检测&＃xff0c;这是Netty提供的一个时间轮定时器&＃xff0c;在任务非常多&＃xff0c;并且任务执行时间很短的情况下&＃xff0c;HashedWheelTimer比Schedule性能更好&＃xff0c;特别适合心跳检测&＃xff1b;

HashedWheelTimer heartbeatTimer &＃61; new HashedWheelTimer(new NamedThreadFactory("dubbo-client-heartbeat", true), tickDuration,                    TimeUnit.MILLISECONDS, Constants.TICKS_PER_WHEEL);

分别启动了两个定时任务&＃xff1a;startHeartBeatTask和startReconnectTask&＃xff1a;

private void startHeartbeatTimer() {    AbstractTimerTask.ChannelProvider cp &＃61; () -> Collections.singletonList(HeaderExchangeClient.this);            long heartbeatTick &＃61; calculateLeastDuration(heartbeat);            long heartbeatTimeoutTick &＃61; calculateLeastDuration(heartbeatTimeout);    HeartbeatTimerTask heartBeatTimerTask &＃61; new HeartbeatTimerTask(cp, heartbeatTick, heartbeat);    ReconnectTimerTask reconnectTimerTask &＃61; new ReconnectTimerTask(cp, heartbeatTimeoutTick, heartbeatTimeout);        　　// init task and start timer.    heartbeatTimer.newTimeout(heartBeatTimerTask, heartbeatTick, TimeUnit.MILLISECONDS);    heartbeatTimer.newTimeout(reconnectTimerTask, heartbeatTimeoutTick, TimeUnit.MILLISECONDS);}

HeartbeatTimerTask&＃xff1a;用来定时发送心跳请求&＃xff0c;心跳间隔时间默认为60秒&＃xff1b;这里重新计算了时间&＃xff0c;其实就是在原来的基础上除以3&＃xff0c;其实就是缩短了检测间隔时间&＃xff0c;增大了及时发现死链的概率&＃xff1b;分别看一下两个任务&＃xff1a;

protected void doTask(Channel channel) {    Long lastRead &＃61; lastRead(channel);    Long lastWrite &＃61; lastWrite(channel);        if ((lastRead !&＃61; null && now() - lastRead > heartbeat)            || (lastWrite !&＃61; null && now() - lastWrite > heartbeat)) {        Request req &＃61; new Request();        req.setVersion(Version.getProtocolVersion());        req.setTwoWay(true);        req.setEvent(Request.HEARTBEAT_EVENT);        channel.send(req);    }}

同上检测最后读写时间和heartbeat的大小&＃xff0c;注&＃xff1a;普通请求和心跳请求都会更新读写时间&＃xff1b;Netty 在 Dubbo 中是如何应用的&＃xff1f;这篇推荐大家看一下。

protected void doTask(Channel channel) {    Long lastRead &＃61; lastRead(channel);    Long now &＃61; now();        if (lastRead !&＃61; null && now - lastRead > heartbeatTimeout) {            if (channel instanceof Client) {            ((Client) channel).reconnect();        } else {            channel.close();        }    }}

同样的在超时的情况下&＃xff0c;Client重连&＃xff0c;Server关闭连接&＃xff1b;同样Server端也有相同的心跳处理&＃xff0c;在可以查看HeaderExchangeServer&＃xff1b;

Dubbo2.7.1-X

在Dubbo2.7.1之后&＃xff0c;借助了Netty提供的IdleStateHandler来实现心跳机制服务&＃xff1a;

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

readerIdleTime&＃xff1a;读超时时间&＃xff1b;

writerIdleTime&＃xff1a;写超时时间&＃xff1b;

allIdleTime&＃xff1a;所有类型的超时时间&＃xff1b;

根据设置的超时时间&＃xff0c;循环检查读写事件多久没有发生了&＃xff0c;在pipeline中加入IdleSateHandler之后&＃xff0c;可以在此pipeline的任意Handler的userEventTriggered方法之中检测IdleStateEvent事件&＃xff1b;下面看看具体Client和Server端添加的IdleStateHandler&＃xff1a;

Client端

protected void initChannel(Channel ch) throws Exception {            final NettyClientHandler nettyClientHandler &＃61; new NettyClientHandler(getUrl(), this);            int heartbeatInterval &＃61; UrlUtils.getHeartbeat(getUrl());    ch.pipeline().addLast("client-idle-handler", new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS))                .addLast("handler", nettyClientHandler);}

Client端在NettyClient中添加了IdleStateHandler&＃xff0c;指定了读写超时时间默认为60秒&＃xff1b;60秒内没有读写事件发生&＃xff0c;会触发IdleStateEvent事件在NettyClientHandler处理&＃xff1a;

public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {        if (evt instanceof IdleStateEvent) {            try {            NettyChannel channel &＃61; NettyChannel.getOrAddChannel(ctx.channel(), url, handler);            Request req &＃61; new Request();            req.setVersion(Version.getProtocolVersion());            req.setTwoWay(true);            req.setEvent(Request.HEARTBEAT_EVENT);            channel.send(req);        } finally {            NettyChannel.removeChannelIfDisconnected(ctx.channel());        }   } else {            　　　　　super.userEventTriggered(ctx, evt);   }}

可以发现接收到IdleStateEvent事件发送了心跳请求&＃xff1b;

至于Client端如何处理重连&＃xff0c;同样在HeaderExchangeClient中使用HashedWheelTimer定时器启动了两个任务&＃xff1a;心跳任务和重连任务&＃xff0c;感觉这里已经不需要心跳任务了&＃xff0c;至于重连任务其实也可以放到userEventTriggered中处理&＃xff1b;

Server端

protected void initChannel(NioSocketChannel ch) throws Exception { int idleTimeout &＃61; UrlUtils.getIdleTimeout(getUrl());  final NettyServerHandler nettyServerHandler &＃61; new NettyServerHandler(getUrl(), this);    ch.pipeline().addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS))                .addLast("handler", nettyServerHandler);}

Server端指定的超时时间默认为60*3秒&＃xff0c;在NettyServerHandler中处理userEventTriggered

public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {        if (evt instanceof IdleStateEvent) {        NettyChannel channel &＃61; NettyChannel.getOrAddChannel(ctx.channel(), url, handler);            try {            channel.close();        } finally {            NettyChannel.removeChannelIfDisconnected(ctx.channel());        }    }            super.userEventTriggered(ctx, evt);}

Server端在指定的超时时间内没有发生读写&＃xff0c;会直接关闭连接&＃xff1b;相比之前现在只有Client发送心跳&＃xff0c;单向发送心跳&＃xff1b;

同样的在HeaderExchangeServer中并没有启动多个认为&＃xff0c;仅仅启动了一个CloseTimerTask&＃xff0c;用来检测超时时间关闭连接&＃xff1b;感觉这个任务是不是也可以不需要了&＃xff0c;IdleStateHandler已经实现了此功能&＃xff1b;

综上&＃xff1a;在使用IdleStateHandler的情况下来同时在HeaderExchangeClient启动心跳&＃43;重连机制&＃xff0c;HeaderExchangeServer启动了关闭连接机制&＃xff1b;主要是因为IdleStateHandler是Netty框架特有了&＃xff0c;而Dubbo是支持多种底层通讯框架的包括Mina&＃xff0c;Grizzy等&＃xff0c;应该是为了兼容此类框架存在的&＃xff1b;

总结

本文首先介绍了RPC中引入的长连接方式&＃xff0c;继而引出长连接的保活机制&＃xff0c;为什么需要保活&＃xff1f;然后分别介绍了网络层保活机制TCP Keepalive机制&＃xff0c;应用层心跳机制&＃xff1b;最后已Dubbo为例看各个版本中对心跳机制的进化。

作者&＃xff1a;ksfzhaohui317

https://segmentfault.com/a/1190000022591346