基礎

何為心跳

顧名思義, 所謂 心跳, 即在 TCP 長連接中, 客戶端和服務器之間定期發送的一種特殊的數據包, 通知對方自己還在線, 以確保 TCP 連接的有效性.

為什么需要心跳

因為網絡的不可靠性, 有可能在 TCP 保持長連接的過程中, 由于某些突發情況, 例如網線被拔出, 突然掉電等, 會造成服務器和客戶端的連接中斷. 在這些突發情況下, 如果恰好服務器和客戶端之間沒有交互的話, 那么它們是不能在短時間內發現對方已經掉線的. 為了解決這個問題, 我們就需要引入 心跳 機制. 心跳機制的工作原理是: 在服務器和客戶端之間一定時間內沒有數據交互時, 即處于 idle 狀態時, 客戶端或服務器會發送一個特殊的數據包給對方, 當接收方收到這個數據報文后, 也立即發送一個特殊的數據報文, 回應發送方, 此即一個 PING-PONG 交互. 自然地, 當某一端收到心跳消息后, 就知道了對方仍然在線, 這就確保 TCP 連接的有效性.

如何實現心跳

我們可以通過兩種方式實現心跳機制:

• 使用 TCP 協議層面的 keepalive 機制.

• 在應用層上實現自定義的心跳機制.

雖然在 TCP 協議層面上, 提供了 keepalive 保活機制, 但是使用它有幾個缺點:

它不是 TCP 的標準協議, 并且是默認關閉的.

TCP keepalive 機制依賴于操作系統的實現, 默認的 keepalive 心跳時間是 兩個小時, 并且對 keepalive 的修改需要系統調用(或者修改系統配置), 靈活性不夠.

TCP keepalive 與 TCP 協議綁定, 因此如果需要更換為 UDP 協議時, keepalive 機制就失效了.

雖然使用 TCP 層面的 keepalive 機制比自定義的應用層心跳機制節省流量, 但是基于上面的幾點缺點, 一般的實踐中, 人們大多數都是選擇在應用層上實現自定義的心跳.
既然如此, 那么我們就來大致看看在在 Netty 中是怎么實現心跳的吧. 在 Netty 中, 實現心跳機制的關鍵是 IdleStateHandler, 它可以對一個 Channel 的 讀/寫設置定時器, 當 Channel 在一定事件間隔內沒有數據交互時(即處于 idle 狀態), 就會觸發指定的事件.

使用 Netty 實現心跳

上面我們提到了, 在 Netty 中, 實現心跳機制的關鍵是 IdleStateHandler, 那么這個 Handler 如何使用呢? 我們來看看它的構造器:

public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS); }

實例化一個 IdleStateHandler 需要提供三個參數:

• readerIdleTimeSeconds, 讀超時. 即當在指定的時間間隔內沒有從 Channel 讀取到數據時, 會觸發一個 READER_IDLE 的 IdleStateEvent 事件.

• writerIdleTimeSeconds, 寫超時. 即當在指定的時間間隔內沒有數據寫入到 Channel 時, 會觸發一個 WRITER_IDLE 的 IdleStateEvent 事件.

• allIdleTimeSeconds, 讀/寫超時. 即當在指定的時間間隔內沒有讀或寫操作時, 會觸發一個 ALL_IDLE 的 IdleStateEvent 事件.

為了展示具體的 IdleStateHandler 實現的心跳機制, 下面我們來構造一個具體的EchoServer 的例子, 這個例子的行為如下:

在這個例子中, 客戶端和服務器通過 TCP 長連接進行通信.

TCP 通信的報文格式是:

+--------+-----+---------------+ | Length |Type | Content | | 17 | 1 |"HELLO, WORLD" | +--------+-----+---------------+

客戶端每隔一個隨機的時間后, 向服務器發送消息, 服務器收到消息后, 立即將收到的消息原封不動地回復給客戶端.

若客戶端在指定的時間間隔內沒有讀/寫操作, 則客戶端會自動向服務器發送一個 PING 心跳, 服務器收到 PING 心跳消息時, 需要回復一個 PONG 消息.

下面所使用的代碼例子可以在我的 Github github.com/yongshun/some_java_code 上找到.

通用部分

根據上面定義的行為, 我們接下來實現心跳的通用部分 CustomHeartbeatHandler:

/*** @author xiongyongshun* @version 1.0* @email yongshun1228@gmail.com* @created 16/9/18 13:02*/ public abstract class CustomHeartbeatHandler extends SimpleChannelInboundHandler<ByteBuf> {public static final byte PING_MSG = 1;public static final byte PONG_MSG = 2;public static final byte CUSTOM_MSG = 3;protected String name;private int heartbeatCount = 0;public CustomHeartbeatHandler(String name) {this.name = name;}@Overrideprotected void channelRead0(ChannelHandlerContext context, ByteBuf byteBuf) throws Exception {if (byteBuf.getByte(4) == PING_MSG) {sendPongMsg(context);} else if (byteBuf.getByte(4) == PONG_MSG){System.out.println(name + " get pong msg from " + context.channel().remoteAddress());} else {handleData(context, byteBuf);}}protected void sendPingMsg(ChannelHandlerContext context) {ByteBuf buf = context.alloc().buffer(5);buf.writeInt(5);buf.writeByte(PING_MSG);context.writeAndFlush(buf);heartbeatCount++;System.out.println(name + " sent ping msg to " + context.channel().remoteAddress() + ", count: " + heartbeatCount);}private void sendPongMsg(ChannelHandlerContext context) {ByteBuf buf = context.alloc().buffer(5);buf.writeInt(5);buf.writeByte(PONG_MSG);context.channel().writeAndFlush(buf);heartbeatCount++;System.out.println(name + " sent pong msg to " + context.channel().remoteAddress() + ", count: " + heartbeatCount);}protected abstract void handleData(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf);@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {// IdleStateHandler 所產生的 IdleStateEvent 的處理邏輯.if (evt instanceof IdleStateEvent) {IdleStateEvent e = (IdleStateEvent) evt;switch (e.state()) {case READER_IDLE:handleReaderIdle(ctx);break;case WRITER_IDLE:handleWriterIdle(ctx);break;case ALL_IDLE:handleAllIdle(ctx);break;default:break;}}}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.err.println("---" + ctx.channel().remoteAddress() + " is active---");}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.err.println("---" + ctx.channel().remoteAddress() + " is inactive---");}protected void handleReaderIdle(ChannelHandlerContext ctx) {System.err.println("---READER_IDLE---");}protected void handleWriterIdle(ChannelHandlerContext ctx) {System.err.println("---WRITER_IDLE---");}protected void handleAllIdle(ChannelHandlerContext ctx) {System.err.println("---ALL_IDLE---");} }

類 CustomHeartbeatHandler 負責心跳的發送和接收, 我們接下來詳細地分析一下它的作用. 我們在前面提到, IdleStateHandler 是實現心跳的關鍵, 它會根據不同的 IO idle 類型來產生不同的 IdleStateEvent 事件, 而這個事件的捕獲, 其實就是在 userEventTriggered 方法中實現的.
我們來看看 CustomHeartbeatHandler.userEventTriggered 的具體實現:

@Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleStateEvent e = (IdleStateEvent) evt;switch (e.state()) {case READER_IDLE:handleReaderIdle(ctx);break;case WRITER_IDLE:handleWriterIdle(ctx);break;case ALL_IDLE:handleAllIdle(ctx);break;default:break;}} }

在 userEventTriggered 中, 根據 IdleStateEvent 的 state() 的不同, 而進行不同的處理. 例如如果是讀取數據 idle, 則 e.state() == READER_IDLE, 因此就調用 handleReaderIdle 來處理它. CustomHeartbeatHandler 提供了三個 idle 處理方法: handleReaderIdle, handleWriterIdle, handleAllIdle, 這三個方法目前只有默認的實現, 它需要在子類中進行重寫, 現在我們暫時略過它們, 在具體的客戶端和服務器的實現部分時再來看它們.

知道了這一點后, 我們接下來看看數據處理部分:

@Override protected void channelRead0(ChannelHandlerContext context, ByteBuf byteBuf) throws Exception {if (byteBuf.getByte(4) == PING_MSG) {sendPongMsg(context);} else if (byteBuf.getByte(4) == PONG_MSG){System.out.println(name + " get pong msg from " + context.channel().remoteAddress());} else {handleData(context, byteBuf);} }

在 CustomHeartbeatHandler.channelRead0 中, 我們首先根據報文協議:

+--------+-----+---------------+ | Length |Type | Content | | 17 | 1 |"HELLO, WORLD" | +--------+-----+---------------+

來判斷當前的報文類型, 如果是 PING_MSG 則表示是服務器收到客戶端的 PING 消息, 此時服務器需要回復一個 PONG 消息, 其消息類型是 PONG_MSG.
扔報文類型是 PONG_MSG, 則表示是客戶端收到服務器發送的 PONG 消息, 此時打印一個 log 即可.

客戶端部分

客戶端初始化

public class Client {public static void main(String[] args) {NioEventLoopGroup workGroup = new NioEventLoopGroup(4);Random random = new Random(System.currentTimeMillis());try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(workGroup).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {protected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline p = socketChannel.pipeline();p.addLast(new IdleStateHandler(0, 0, 5));p.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, -4, 0));p.addLast(new ClientHandler());}});Channel ch = bootstrap.remoteAddress("127.0.0.1", 12345).connect().sync().channel();for (int i = 0; i < 10; i++) {String content = "client msg " + i;ByteBuf buf = ch.alloc().buffer();buf.writeInt(5 + content.getBytes().length);buf.writeByte(CustomHeartbeatHandler.CUSTOM_MSG);buf.writeBytes(content.getBytes());ch.writeAndFlush(buf);Thread.sleep(random.nextInt(20000));}} catch (Exception e) {throw new RuntimeException(e);} finally {workGroup.shutdownGracefully();}} }

上面的代碼是 Netty 的客戶端端的初始化代碼, 使用過 Netty 的朋友對這個代碼應該不會陌生. 別的部分我們就不再贅述, 我們來看看 ChannelInitializer.initChannel 部分即可:

.handler(new ChannelInitializer<SocketChannel>() {protected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline p = socketChannel.pipeline();p.addLast(new IdleStateHandler(0, 0, 5));p.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, -4, 0));p.addLast(new ClientHandler());} });

我們給 pipeline 添加了三個 Handler, IdleStateHandler 這個 handler 是心跳機制的核心, 我們為客戶端端設置了讀寫 idle 超時, 時間間隔是5s, 即如果客戶端在間隔 5s 后都沒有收到服務器的消息或向服務器發送消息, 則產生 ALL_IDLE 事件.
接下來我們添加了 LengthFieldBasedFrameDecoder, 它是負責解析我們的 TCP 報文, 因為和本文的目的無關, 因此這里不詳細展開.
最后一個 Handler 是 ClientHandler, 它繼承于 CustomHeartbeatHandler, 是我們處理業務邏輯部分.

客戶端 Handler

public class ClientHandler extends CustomHeartbeatHandler {public ClientHandler() {super("client");}@Overrideprotected void handleData(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) {byte[] data = new byte[byteBuf.readableBytes() - 5];byteBuf.skipBytes(5);byteBuf.readBytes(data);String content = new String(data);System.out.println(name + " get content: " + content);}@Overrideprotected void handleAllIdle(ChannelHandlerContext ctx) {super.handleAllIdle(ctx);sendPingMsg(ctx);} }

ClientHandler 繼承于 CustomHeartbeatHandler, 它重寫了兩個方法, 一個是 handleData, 在這里面實現 僅僅打印收到的消息.
第二個重寫的方法是 handleAllIdle. 我們在前面提到, 客戶端負責發送心跳的 PING 消息, 當客戶端產生一個 ALL_IDLE 事件后, 會導致父類的 CustomHeartbeatHandler.userEventTriggered 調用, 而 userEventTriggered 中會根據 e.state() 來調用不同的方法, 因此最后調用的是 ClientHandler.handleAllIdle, 在這個方法中, 客戶端調用 sendPingMsg 向服務器發送一個 PING 消息.

服務器部分

服務器初始化

public class Server {public static void main(String[] args) {NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);NioEventLoopGroup workGroup = new NioEventLoopGroup(4);try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {protected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline p = socketChannel.pipeline();p.addLast(new IdleStateHandler(10, 0, 0));p.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, -4, 0));p.addLast(new ServerHandler());}});Channel ch = bootstrap.bind(12345).sync().channel();ch.closeFuture().sync();} catch (Exception e) {throw new RuntimeException(e);} finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}} }

服務器的初始化部分也沒有什么好說的, 它也和客戶端的初始化一樣, 為 pipeline 添加了三個 Handler.

服務器 Handler

public class ServerHandler extends CustomHeartbeatHandler {public ServerHandler() {super("server");}@Overrideprotected void handleData(ChannelHandlerContext channelHandlerContext, ByteBuf buf) {byte[] data = new byte[buf.readableBytes() - 5];ByteBuf responseBuf = Unpooled.copiedBuffer(buf);buf.skipBytes(5);buf.readBytes(data);String content = new String(data);System.out.println(name + " get content: " + content);channelHandlerContext.write(responseBuf);}@Overrideprotected void handleReaderIdle(ChannelHandlerContext ctx) {super.handleReaderIdle(ctx);System.err.println("---client " + ctx.channel().remoteAddress().toString() + " reader timeout, close it---");ctx.close();} }

ServerHandler 繼承于 CustomHeartbeatHandler, 它重寫了兩個方法, 一個是 handleData, 在這里面實現 EchoServer 的功能: 即收到客戶端的消息后, 立即原封不動地將消息回復給客戶端.
第二個重寫的方法是 handleReaderIdle, 因為服務器僅僅對客戶端的讀 idle 感興趣, 因此只重新了這個方法. 若服務器在指定時間后沒有收到客戶端的消息, 則會觸發 READER_IDLE 消息, 進而會調用 handleReaderIdle 這個方法. 我們在前面提到, 客戶端負責發送心跳的 PING 消息, 并且服務器的 READER_IDLE 的超時時間是客戶端發送 PING 消息的間隔的兩倍, 因此當服務器 READER_IDLE 觸發時, 就可以確定是客戶端已經掉線了, 因此服務器直接關閉客戶端連接即可.

總結

使用 Netty 實現心跳機制的關鍵就是利用 IdleStateHandler 來產生對應的 idle 事件.

一般是客戶端負責發送心跳的 PING 消息, 因此客戶端注意關注 ALL_IDLE 事件, 在這個事件觸發后, 客戶端需要向服務器發送 PING 消息, 告訴服務器"我還存活著".

服務器是接收客戶端的 PING 消息的, 因此服務器關注的是 READER_IDLE 事件, 并且服務器的 READER_IDLE 間隔需要比客戶端的 ALL_IDLE 事件間隔大(例如客戶端ALL_IDLE 是5s 沒有讀寫時觸發, 因此服務器的 READER_IDLE 可以設置為10s)

當服務器收到客戶端的 PING 消息時, 會發送一個 PONG 消息作為回復. 一個 PING-PONG 消息對就是一個心跳交互.

實現客戶端的斷線重連

public class Client {private NioEventLoopGroup workGroup = new NioEventLoopGroup(4);private Channel channel;private Bootstrap bootstrap;public static void main(String[] args) throws Exception {Client client = new Client();client.start();client.sendData();}public void sendData() throws Exception {Random random = new Random(System.currentTimeMillis());for (int i = 0; i < 10000; i++) {if (channel != null && channel.isActive()) {String content = "client msg " + i;ByteBuf buf = channel.alloc().buffer(5 + content.getBytes().length);buf.writeInt(5 + content.getBytes().length);buf.writeByte(CustomHeartbeatHandler.CUSTOM_MSG);buf.writeBytes(content.getBytes());channel.writeAndFlush(buf);}Thread.sleep(random.nextInt(20000));}}public void start() {try {bootstrap = new Bootstrap();bootstrap.group(workGroup).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {protected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline p = socketChannel.pipeline();p.addLast(new IdleStateHandler(0, 0, 5));p.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, -4, 0));p.addLast(new ClientHandler(Client.this));}});doConnect();} catch (Exception e) {throw new RuntimeException(e);}}protected void doConnect() {if (channel != null && channel.isActive()) {return;}ChannelFuture future = bootstrap.connect("127.0.0.1", 12345);future.addListener(new ChannelFutureListener() {public void operationComplete(ChannelFuture futureListener) throws Exception {if (futureListener.isSuccess()) {channel = futureListener.channel();System.out.println("Connect to server successfully!");} else {System.out.println("Failed to connect to server, try connect after 10s");futureListener.channel().eventLoop().schedule(new Runnable() {@Overridepublic void run() {doConnect();}}, 10, TimeUnit.SECONDS);}}});}}

上面的代碼中, 我們抽象出 doConnect 方法, 它負責客戶端和服務器的 TCP 連接的建立, 并且當 TCP 連接失敗時, doConnect 會 通過 "channel().eventLoop().schedule" 來延時10s 后嘗試重新連接.

客戶端 Handler

public class ClientHandler extends CustomHeartbeatHandler {private Client client;public ClientHandler(Client client) {super("client");this.client = client;}@Overrideprotected void handleData(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) {byte[] data = new byte[byteBuf.readableBytes() - 5];byteBuf.skipBytes(5);byteBuf.readBytes(data);String content = new String(data);System.out.println(name + " get content: " + content);}@Overrideprotected void handleAllIdle(ChannelHandlerContext ctx) {super.handleAllIdle(ctx);sendPingMsg(ctx);}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {super.channelInactive(ctx);client.doConnect();} }

斷線重連的關鍵一點是檢測連接是否已經斷開. 因此我們改寫了 ClientHandler, 重寫了 channelInactive 方法. 當 TCP 連接斷開時, 會回調 channelInactive 方法, 因此我們在這個方法中調用 client.doConnect() 來進行重連.

完整代碼可以在我的 Github github.com/yongshun/some_java_code 上找到.

本文由 yongshun 發表于個人博客, 采用署名-非商業性使用-相同方式共享 3.0 中國大陸許可協議.
非商業轉載請注明作者及出處. 商業轉載請聯系作者本人
Email: yongshun1228@gmail.com
本文標題為: 淺析 Netty 實現心跳機制與斷線重連
本文鏈接為: https://segmentfault.com/a/1190000006931568

總結

以上是生活随笔為你收集整理的浅析 Netty 实现心跳机制与断线重连的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。