Dubbo进阶（十）：心跳机制

Bertha 。 2023-02-10 05:22 39阅读 0赞

Dubbo客户端和Dubbo服务端之间存在心跳，目的是维持provider和consumer之间的长链接。由Dubbo客户端主动发起，可参见Dubbo源码 `HeartbeatTimerTask`和`ReconnectTimerTask`。

谈到RPC肯定绕不开TCP通信，而主流的RPC框架都依赖于Netty等通信框架，这时候我们还要考虑是使用长连接还是短连接。主流的RPC框架都会追求性能选择使用长连接，所以如何保活连接就是一个重要的话题，本文就会以这个为中心来介绍一下保活策略。

##### KeepAlive机制 #####

Dubbo中的通信是基于TCP的，TCP本身并没有长短连接的区别。

*  在短连接中，每次通信时，都会创建Socket，当该次通信结束后，就会调用socket.close()，下次通信需要重新创建连接。优点就是无需管理连接，无需保活连接；缺点就是每次创建连接需要耗费时间。
 *  在长连接中，每次通信完毕后，不会关闭连接，这样子就实现了连接可以复用，保证了性能；长连接的优点是省去了创建连接时所耗费的时间；缺点就是连接需要统一管理，并且需要保活。

那么如何确保连接的有效性呢，在TCP中用到了KeepAlive机制，keepalive并不是TCP协议的一部分，但是大多数操作系统都实现了这个机制，在一定时间内，在链路上如果没有数据传送的情况下，TCP层将会发送相应的keepalive探针来确定连接可用性，Keepalive几个内核参数配置：

*  tcp\_keepalive\_time：连接多长时间没有数据往来发送探针请求，默认为7200s（2h）；
 *  tcp\_keepalive\_probes：探测失败重试的次数默认为10次；
 *  tcp\_keepalive\_intvl：重试的间隔时间默认75s；

##### Dubbo心跳机制 #####

有了KeepAlive机制往往是不够用的，还需要配合心跳机制来一起使用。何为心跳机制，简单来讲就是客户端启动一个定时器用来定时发送请求，服务端接到请求进行响应，如果多次没有接受到响应，那么客户端认为连接已经断开，可以断开半打开的连接或者进行重连处理。

接下来我们以Dubbo的源码来分析一下Dubbo的心跳机制是如何实现的。

*  首先要知道provider绑定和consumer连接的入口：

public class HeaderExchanger implements Exchanger {
    
        public static final String NAME = "header";
    
        @Override
        public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
            return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
        }
    
        @Override
        public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
            return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
        }
    }

*  来看一下第一个定时任务：发送心跳请求`HeartbeatTimerTask#doTask`

@Override
    protected void doTask(Channel channel) {
        try {
            // 最后一次读的时间
            Long lastRead = lastRead(channel);
            // 最后一次写的时间
            Long lastWrite = lastWrite(channel);
            // 如果最后读的时间间隔或者最后写的时间间隔大于heartbeat，就会发送心跳信息
            // heartbeat默认值是60s
            if ((lastRead != null && now() - lastRead > heartbeat)
                    || (lastWrite != null && now() - lastWrite > heartbeat)) {
                Request req = new Request();
                req.setVersion(Version.getProtocolVersion());
                
                req.setTwoWay(true);
                req.setEvent(HEARTBEAT_EVENT);
                channel.send(req);
                if (logger.isDebugEnabled()) {
                    logger.debug("Send heartbeat to remote channel " + channel.getRemoteAddress()
                            + ", cause: The channel has no data-transmission exceeds a heartbeat period: "
                            + heartbeat + "ms");
                }
            }
        } catch (Throwable t) {
            logger.warn("Exception when heartbeat to remote channel " + channel.getRemoteAddress(), t);
        }
    }

*  再看一下第二个定时任务：处理重连和断连`ReconnectTimerTask#doTask`

@Override
    protected void doTask(Channel channel) {
        try {
            // 最后一次读的时间
            Long lastRead = lastRead(channel);
            Long now = now();
    
            // 无法初始化链接，则进行重连
            if (!channel.isConnected()) {
                try {
                    logger.info("Initial connection to " + channel);
                    ((Client) channel).reconnect();
                } catch (Exception e) {
                    logger.error("Fail to connect to " + channel, e);
                }
            // 如果最后读的时间间隔大于idleTimeout，则进行重连
            // idleTimeout的默认时间60s * 3
            } else if (lastRead != null && now - lastRead > idleTimeout) {
                logger.warn("Reconnect to channel " + channel + ", because heartbeat read idle time out: "
                        + idleTimeout + "ms");
                try {
                    ((Client) channel).reconnect();
                } catch (Exception e) {
                    logger.error(channel + "reconnect failed during idle time.", e);
                }
            }
        } catch (Throwable t) {
            logger.warn("Exception when reconnect to remote channel " + channel.getRemoteAddress(), t);
        }
    }

*  最后看一下关闭连接的定时任务：`CloseTimerTask#doTask`

@Override
    protected void doTask(Channel channel) {
        try {
            // 最后一次读的时间
            Long lastRead = lastRead(channel);
            // 最后一次写的时间
            Long lastWrite = lastWrite(channel);
            Long now = now();
            // 如果最后一次读的时间间隔大于idleTimeout，或者最后一次写的时间间隔大于idleTimeout，就断开链接
            // idleTimeout的默认时间60s * 3
            if ((lastRead != null && now - lastRead > idleTimeout)
                    || (lastWrite != null && now - lastWrite > idleTimeout)) {
                logger.warn("Close channel " + channel + ", because idleCheck timeout: "
                        + idleTimeout + "ms");
                channel.close();
            }
        } catch (Throwable t) {
            logger.warn("Exception when close remote channel " + channel.getRemoteAddress(), t);
        }
    }

*  连接建立时创建定时器（客户端）

HeaderExchangeClient中使用HashedWheelTimer开启心跳检测，这是Netty提供的一个时间轮定时器，在任务非常多，并且任务执行时间很短的情况下，HashedWheelTimer比Schedule性能更好，特别适合心跳检测。

public class HeaderExchangeClient implements ExchangeClient {
    
        private final Client client;
        private final ExchangeChannel channel;
    
        // 这是Netty提供的一个时间轮定时器
        private static final HashedWheelTimer IDLE_CHECK_TIMER = new HashedWheelTimer(
                new NamedThreadFactory("dubbo-client-idleCheck", true), 1, TimeUnit.SECONDS, TICKS_PER_WHEEL);
        private HeartbeatTimerTask heartBeatTimerTask;
        private ReconnectTimerTask reconnectTimerTask;
    
        public HeaderExchangeClient(Client client, boolean startTimer) {
            Assert.notNull(client, "Client can't be null");
            this.client = client;
            this.channel = new HeaderExchangeChannel(client);
    
            if (startTimer) {
                URL url = client.getUrl();
                // 客户端发起重连
                startReconnectTask(url);
                // 客户端发送心跳
                startHeartBeatTask(url);
            }
        }
        
        // 省略。。。
        
        private void startHeartBeatTask(URL url) {
            if (!client.canHandleIdle()) {
                AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(HeaderExchangeClient.this);
                // 默认值是60s
                int heartbeat = getHeartbeat(url);
                // 重连、断连，执行的频率均为各自检测周期的 1/3。定时发送心跳的任务负责在连接空闲时，
                // 向对端发送心跳包。定时重连、断连的任务负责检测 lastRead 是否在超时周期内仍未被更新，
                // 如果判定为超时，客户端处理的逻辑是重连，服务端则采取断连的措施。
                long heartbeatTick = calculateLeastDuration(heartbeat);
                // 发送心跳的定时任务
                this.heartBeatTimerTask = new HeartbeatTimerTask(cp, heartbeatTick, heartbeat);
                IDLE_CHECK_TIMER.newTimeout(heartBeatTimerTask, heartbeatTick, TimeUnit.MILLISECONDS);
            }
        }
    
        private void startReconnectTask(URL url) {
            if (shouldReconnect(url)) {
                AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(HeaderExchangeClient.this);
                // idleTimeout应该至少是heartBeat的两倍以上，因为客户端可能会重试。
                int idleTimeout = getIdleTimeout(url);
                long heartbeatTimeoutTick = calculateLeastDuration(idleTimeout);
                this.reconnectTimerTask = new ReconnectTimerTask(cp, heartbeatTimeoutTick, idleTimeout);
                IDLE_CHECK_TIMER.newTimeout(reconnectTimerTask, heartbeatTimeoutTick, TimeUnit.MILLISECONDS);
            }
        }
    }

*  服务端的逻辑和客户端差不多，所以就不做详细介绍了。

##### 总结Dubbo的心跳方案 #####

Dubbo 对于建立的每一个连接，同时在客户端和服务端开启了 2 个定时器，一个用于定时发送心跳，一个用于定时重连、断连，执行的频率均为各自检测周期的 1/3。定时发送心跳的任务负责在连接空闲时，向对端发送心跳包。定时重连、断连的任务负责检测 lastRead 是否在超时周期内仍未被更新，如果判定为超时，客户端处理的逻辑是重连，服务端则采取断连的措施。