在網(wǎng)絡(luò)傳輸中，怎么確保通道連接的可用性是一個(gè)很重要的問題，簡(jiǎn)單的說，在網(wǎng)絡(luò)通信中有客戶端和服務(wù)端，一個(gè)負(fù)責(zé)發(fā)送請(qǐng)求，一個(gè)負(fù)責(zé)接收請(qǐng)求，在保證連接有效性的背景下，這兩個(gè)物體扮演了什么角色，心跳機(jī)制能有效的保證連接的可用性，那它的機(jī)制是什么，下文中將會(huì)詳細(xì)講解。

網(wǎng)絡(luò)層的可用性

首先講一下TCP,在dubbo中的通信是基于TCP的，TCP本身并沒有長(zhǎng)短連接的區(qū)別，在短連接中，每次通信時(shí)，都會(huì)創(chuàng)建Socket,當(dāng)該次通信結(jié)束后，就會(huì)調(diào)用socket.close()；而在長(zhǎng)連接中，每次通信完畢后，不會(huì)關(guān)閉連接，這樣就可以做到連接的復(fù)用，長(zhǎng)連接的好處是省去了創(chuàng)建連接時(shí)的耗時(shí)。那么如何確保連接的有效性呢，在TCP中用到了KeepAlive機(jī)制，keepalive并不是TCP協(xié)議的一部分，但是大多數(shù)操作系統(tǒng)都實(shí)現(xiàn)了這個(gè)機(jī)制，在一定時(shí)間內(nèi)，在鏈路上如果沒有數(shù)據(jù)傳送的情況下，TCP層將會(huì)發(fā)送相應(yīng)的keepalive探針來確定連接可用性，探測(cè)失敗后重試10次（tcp_keepalive_probes）,每次間隔時(shí)間為75s(tcp_keepalive_intvl),所有探測(cè)失敗后，才認(rèn)為當(dāng)前連接已經(jīng)不可用了。

KeepAlive機(jī)制是在網(wǎng)絡(luò)層保證了連接的可用性，但在應(yīng)用層我們認(rèn)為這還是不夠的。

• KeepAlive的報(bào)活機(jī)制只有在鏈路空閑的情況下才會(huì)起作用，假如此時(shí)有數(shù)據(jù)發(fā)送，且物理鏈路已經(jīng)不通，操作系統(tǒng)這邊的鏈路狀態(tài)還是E STABLISHED,這時(shí)會(huì)發(fā)生TCP重傳機(jī)制，要知道默認(rèn)的TCP超時(shí)重傳，指數(shù)退避算法也是一個(gè)相當(dāng)長(zhǎng)的過程。
• KeepAlive本身是面向網(wǎng)絡(luò)的，并不是面向于應(yīng)用的，可能是由于本身GC問題，系統(tǒng)load高等情況，但網(wǎng)絡(luò)依然是通的，此時(shí)，應(yīng)用已經(jīng)失去了活性，所以連接自然認(rèn)為是不可用的。

應(yīng)用層的連接可用性：心跳機(jī)制

如何理解應(yīng)用層的心跳？簡(jiǎn)單的說，就是客戶端會(huì)開啟一個(gè)定時(shí)任務(wù)，定時(shí)對(duì)已經(jīng)建立連接的對(duì)端應(yīng)用發(fā)送請(qǐng)求，服務(wù)端則需要特殊處理該請(qǐng)求，返回響應(yīng)。如果心跳持續(xù)多次沒有收到響應(yīng)，客戶端會(huì)認(rèn)為連接不可用，主動(dòng)斷開連接。

客戶端如何得知請(qǐng)求失敗了？

在失敗的場(chǎng)景下，服務(wù)端是不會(huì)返回響應(yīng)的，所以只能在客戶端自身上設(shè)計(jì)了。
當(dāng)客戶端發(fā)起一個(gè)RPC請(qǐng)求時(shí)，會(huì)設(shè)置一個(gè)超時(shí)時(shí)間client_timeout,同時(shí)它也會(huì)開啟一個(gè)延遲的client_timeout的定時(shí)器。當(dāng)接收到正常響應(yīng)時(shí)，會(huì)移除該定時(shí)器；而當(dāng)計(jì)時(shí)器倒計(jì)時(shí)完畢后，還沒有被移除，則會(huì)認(rèn)為請(qǐng)求超時(shí)，構(gòu)造一個(gè)失敗的響應(yīng)傳遞給客戶端。

連接建立時(shí)創(chuàng)建定時(shí)器

HeaderExchangeClient類

 public HeaderExchangeClient(Client client, boolean needHeartbeat) {
        if (client == null) {
            throw new IllegalArgumentException("client == null");
        }
        this.client = client;
        // 創(chuàng)建信息交換通道
        this.channel = new HeaderExchangeChannel(client);
        // 獲得dubbo版本
        String dubbo = client.getUrl().getParameter(Constants.DUBBO_VERSION_KEY);
        //獲得心跳周期配置，如果沒有配置，并且dubbo是1.0版本的，則這只為1分鐘，否則設(shè)置為0
        this.heartbeat = client.getUrl().getParameter(Constants.HEARTBEAT_KEY, dubbo != null && dubbo.startsWith("1.0.") ? Constants.DEFAULT_HEARTBEAT : 0);
        // 獲得心跳超時(shí)配置，默認(rèn)是心跳周期的三倍
        this.heartbeatTimeout = client.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3);
                 
        if (needHeartbeat) {
            // 開啟心跳
          long tickDuration = calculateLeastDuration(heartbeat);
          heartbeatTimer = new HashedWheelTimer(new NamedThreadFactory("dubbo-client-heartbeat", true) , tickDuration, TimeUnit.MILLISECONDS, Constants.TICKS_PER_WHEEL);
          startHeartbeatTimer();
        }
    }

創(chuàng)建了一個(gè)HashedWheelTimer開啟心跳檢測(cè)，這是 Netty 所提供的一個(gè)經(jīng)典的時(shí)間輪定時(shí)器實(shí)現(xiàn)。

HeaderExchangeServer也同時(shí)開啟了定時(shí)器，代碼邏輯和上述差不多。

開啟兩個(gè)定時(shí)任務(wù)

private void startHeartbeatTimer() {
           long heartbeatTick = calculateLeastDuration(heartbeat); 
   long heartbeatTimeoutTick = calculateLeastDuration(heartbeatTimeout);
   HeartbeatTimerTask heartBeatTimerTask =new  HeartbeatTimerTask(cp, heartbeatTick, heartbeat);
   ReconnectTimerTask reconnectTimerTask = new ReconnectTimerTask(cp, heartbeatTimeoutTick, heartbeatTimeout);
    
  heartbeatTimer.newTimeout(heartBeatTimerTask, heartbeatTick, TimeUnit.MILLISECONDS); 
  heartbeatTimer.newTimeout(reconnectTimerTask, heartbeatTimeoutTick, TimeUnit.MILLISECONDS);
}

在該方法中主要開啟了兩個(gè)定時(shí)器

• HeartbeatTimerTask 主要是定時(shí)發(fā)送心跳請(qǐng)求
• ReconnectTimerTask 主要是心跳失敗后處理重連，斷連的邏輯

舊版的心跳處理HeartBeatTask類

final class HeartBeatTask implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(HeartBeatTask.class);

    /**
     * 通道管理
     */
    private ChannelProvider channelProvider;

    /**
     * 心跳間隔 單位：ms
     */
    private int heartbeat;

    /**
     * 心跳超時(shí)時(shí)間 單位：ms
     */
    private int heartbeatTimeout;

    HeartBeatTask(ChannelProvider provider, int heartbeat, int heartbeatTimeout) {
        this.channelProvider = provider;
        this.heartbeat = heartbeat;
        this.heartbeatTimeout = heartbeatTimeout;
    }

    @Override
    public void run() {
        try {
            long now = System.currentTimeMillis();
            // 遍歷所有通道
            for (Channel channel : channelProvider.getChannels()) {
                // 如果通道關(guān)閉了，則跳過
                if (channel.isClosed()) {
                    continue;
                }
                try {
                    // 最后一次接收到消息的時(shí)間戳
                    Long lastRead = (Long) channel.getAttribute(
                            HeaderExchangeHandler.KEY_READ_TIMESTAMP);
                    // 最后一次發(fā)送消息的時(shí)間戳
                    Long lastWrite = (Long) channel.getAttribute(
                            HeaderExchangeHandler.KEY_WRITE_TIMESTAMP);
                    // 如果最后一次接收或者發(fā)送消息到時(shí)間到現(xiàn)在的時(shí)間間隔超過了心跳間隔時(shí)間
                    if ((lastRead != null && now - lastRead > heartbeat)
                            || (lastWrite != null && now - lastWrite > heartbeat)) {
                        // 創(chuàng)建一個(gè)request
                        Request req = new Request();
                        // 設(shè)置版本號(hào)
                        req.setVersion(Version.getProtocolVersion());
                        // 設(shè)置需要得到響應(yīng)
                        req.setTwoWay(true);
                        // 設(shè)置事件類型，為心跳事件
                        req.setEvent(Request.HEARTBEAT_EVENT);
                        // 發(fā)送心跳請(qǐng)求
                        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");
                        }
                    }
                    // 如果最后一次接收消息的時(shí)間到現(xiàn)在已經(jīng)超過了超時(shí)時(shí)間
                    if (lastRead != null && now - lastRead > heartbeatTimeout) {
                        logger.warn("Close channel " + channel
                                + ", because heartbeat read idle time out: " + heartbeatTimeout + "ms");
                        // 如果該通道是客戶端，也就是請(qǐng)求的服務(wù)器掛掉了，客戶端嘗試重連服務(wù)器
                        if (channel instanceof Client) {
                            try {
                                // 重新連接服務(wù)器
                                ((Client) channel).reconnect();
                            } catch (Exception e) {
                                //do nothing
                            }
                        } else {
                            // 如果不是客戶端，也就是是服務(wù)端返回響應(yīng)給客戶端，但是客戶端掛掉了，則服務(wù)端關(guān)閉客戶端連接
                            channel.close();
                        }
                    }
                } catch (Throwable t) {
                    logger.warn("Exception when heartbeat to remote channel " + channel.getRemoteAddress(), t);
                }
            }
        } catch (Throwable t) {
            logger.warn("Unhandled exception when heartbeat, cause: " + t.getMessage(), t);
        }
    }

    interface ChannelProvider {
        // 獲得所有的通道集合，需要心跳的通道數(shù)組
        Collection<Channel> getChannels();
    }

}

它首先遍歷所有的Channel,在服務(wù)端對(duì)用的是所有客戶端連接，在客戶端對(duì)應(yīng)的是服務(wù)端連接，判斷當(dāng)前TCP連接是否空閑，如果空閑就發(fā)送心跳報(bào)文，判斷是否空閑，根據(jù)Channel是否有讀或?qū)憗頉Q定，比如一分鐘內(nèi)沒有讀或?qū)懢桶l(fā)送心跳報(bào)文，然后是處理超時(shí)的問題，處理客戶端超時(shí)重新建立TCP連接，目前的策略是檢查是否在3分鐘內(nèi)都沒有成功接受或發(fā)送報(bào)文，如果在服務(wù)端檢測(cè)則就會(huì)主動(dòng)關(guān)閉遠(yuǎn)程客戶端連接。

新版本的心跳機(jī)制

定時(shí)任務(wù)一： 發(fā)送心跳請(qǐng)求

在新版本下，去除了HeartBeatTask類,添加了HeartbeatTimerTask和ReconnectTimerTask類

public class HeartbeatTimerTask extends AbstractTimerTask {

    private static final Logger logger = LoggerFactory.getLogger(HeartbeatTimerTask.class);

    private final int heartbeat;

    HeartbeatTimerTask(ChannelProvider channelProvider, Long heartbeatTick, int heartbeat) {
        super(channelProvider, heartbeatTick);
        this.heartbeat = heartbeat;
    }

    @Override
    protected void doTask(Channel channel) {
        try {
            Long lastRead = lastRead(channel);
            Long lastWrite = lastWrite(channel);
            if ((lastRead != null && now() - lastRead > heartbeat)
                    || (lastWrite != null && now() - lastWrite > heartbeat)) {
                Request req = new Request();
                req.setVersion(Version.getProtocolVersion());
                req.setTwoWay(true);
                req.setEvent(Request.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);
        }
    }
}

Dubbo采取的是雙向心跳設(shè)計(jì)，即服務(wù)端會(huì)向客戶端發(fā)送心跳，客戶端也會(huì)向服務(wù)端發(fā)送心跳，接收的一方更新lastread字段，發(fā)送的一方更新lastWrite字段，超過心跳間隙的時(shí)間，便發(fā)送心跳請(qǐng)求給對(duì)端。

定時(shí)任務(wù)二： 處理重連和斷連

public class ReconnectTimerTask extends AbstractTimerTask {

    private static final Logger logger = LoggerFactory.getLogger(ReconnectTimerTask.class);

    private final int idleTimeout;

    public ReconnectTimerTask(ChannelProvider channelProvider, Long heartbeatTimeoutTick, int idleTimeout) {
        super(channelProvider, heartbeatTimeoutTick);
        this.idleTimeout = idleTimeout;
    }

    @Override
    protected void doTask(Channel channel) {
        try {
            Long lastRead = lastRead(channel);
            Long now = now();

            // Rely on reconnect timer to reconnect when AbstractClient.doConnect fails to init the connection
            if (!channel.isConnected()) {
                try {
                    logger.info("Initial connection to " + channel);
                    ((Client) channel).reconnect();
                } catch (Exception e) {
                    logger.error("Fail to connect to " + channel, e);
                }
            // check pong at client
            } 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);
        }
    }
}

不同類型處理機(jī)制不同，當(dāng)超過設(shè)置的心跳總時(shí)間后，客戶端選擇的是重新連接，服務(wù)端是選擇直接斷開連接。

心跳改進(jìn)方案

Netty對(duì)空閑連接的檢測(cè)提供了天然的支持，使用IdleStateHandler可以很方便的實(shí)現(xiàn)空閑檢測(cè)邏輯。

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

• readerIdleTime: 讀超時(shí)的時(shí)間
• writerIdleTime: 寫超時(shí)的時(shí)間
• allIdleTime: 所有類型的超時(shí)時(shí)間
  客戶端和服務(wù)端配置
  客戶端：

bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
    @Override
    protected void initChannel(NioSocketChannel ch) throws Exception {
        ch.pipeline().addLast("clientIdleHandler", new IdleStateHandler(60, 0, 0));
    }
});

服務(wù)端：

serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
    @Override
    protected void initChannel(NioSocketChannel ch) throws Exception {
        ch.pipeline().addLast("serverIdleHandler",new IdleStateHandler(0, 0, 200));
    }
}

從上面看出，客戶端配置了read超時(shí)為60s，服務(wù)端配置了write/read超時(shí)未200s，

空閑超時(shí)邏輯-客戶端

對(duì)于空閑超時(shí)的處理邏輯，客戶端和服務(wù)端是不同的，首先來看客戶端的：

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof IdleStateEvent) {
        // send heartbeat
        sendHeartBeat();
    } else {
        super.userEventTriggered(ctx, evt);
    }
}

檢測(cè)到空閑超時(shí)后，采取的行為是向服務(wù)端發(fā)送心跳包，

public void sendHeartBeat() {
    Invocation invocation = new Invocation();
    invocation.setInvocationType(InvocationType.HEART_BEAT);
    channel.writeAndFlush(invocation).addListener(new CallbackFuture() {
        @Override
        public void callback(Future future) {
            RPCResult result = future.get();
            //超時(shí) 或者 寫失敗
            if (result.isError()) {
                channel.addFailedHeartBeatTimes();
                if (channel.getFailedHeartBeatTimes() >= channel.getMaxHeartBeatFailedTimes()) {
                    channel.reconnect();
                }
            } else {
                channel.clearHeartBeatFailedTimes();
            }
        }
    });
}

構(gòu)造一個(gè)心跳包發(fā)送到服務(wù)端，接受響應(yīng)結(jié)果

• 響應(yīng)成功，清除請(qǐng)求失敗標(biāo)記
• 響應(yīng)失敗，心跳失敗標(biāo)記+1，如果超過配置的失敗次數(shù)，則重新連接

空閑超時(shí)邏輯 - 服務(wù)端

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof IdleStateEvent) {
        channel.close();
    } else {
        super.userEventTriggered(ctx, evt);
    }
}

服務(wù)端直接關(guān)閉連接。