【Netty】心跳机制与断线重连

用户3467126

发布于 2019-07-03 17:24:04

2.1K0

发布于 2019-07-03 17:24:04

文章被收录于专栏：爱编码爱编码

欢迎关注公众号：【爱编码】如果有需要后台回复2019赠送1T的学习资料哦！！

心跳是啥

在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性.

心跳机制的工作原理

心跳机制的工作原理是: 在服务器和客户端之间一定时间内没有数据交互时, 即处于 idle 状态时, 客户端或服务器会发送一个特殊的数据包给对方, 当接收方收到这个数据报文后, 也立即发送一个特殊的数据报文, 回应发送方, 此即一个 PING-PONG 交互. 自然地, 当某一端收到心跳消息后, 就知道了对方仍然在线, 这就确保 TCP 连接的有效性.

实现心跳

在 Netty 中, 实现心跳机制的关键是 IdleStateHandler, 它可以对一个 Channel 的读/写设置定时器, 当 Channel 在一定事件间隔内没有数据交互时(即处于 idle 状态), 就会触发指定的事件.

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

IdleStateHandler构造函数需要提供三个参数:

readerIdleTimeSeconds, 读超时. 即当在指定的时间间隔内没有从 Channel 读取到数据时, 会触发一个READER_IDLE 的 IdleStateEvent 事件.
writerIdleTimeSeconds, 写超时. 即当在指定的时间间隔内没有数据写入到 Channel 时, 会触发一个 WRITER_IDLE的 IdleStateEvent 事件.
allIdleTimeSeconds, 读/写超时. 即当在指定的时间间隔内没有读或写操作时, 会触发一个 ALL_IDLE 的 IdleStateEvent 事件.

服务端

服务端启动初始化代码

服务器的初始化部分为 pipeline 添加了三个 Handler，其中IdleStateHandler就是心跳处理Handler

public class HeartbeatServer {
 public static void main(String[] args) throws InterruptedException {
 EventLoopGroup bossGroup = new NioEventLoopGroup();
 EventLoopGroup workerGroup = new NioEventLoopGroup();
 try {
 ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
 .option(ChannelOption.SO_BACKLOG, 1024)
 .childHandler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new IdleStateHandler(5, 5, 5, TimeUnit.SECONDS));
                            socketChannel.pipeline().addLast(new StringDecoder());
                            socketChannel.pipeline().addLast(new HeartBeatServerHandler());
 }
 });
 ChannelFuture f = b.bind(8089).sync();
            f.channel().closeFuture().sync();
 }catch (Exception e){
            e.printStackTrace();
 }finally {
            workerGroup.shutdownGracefully().sync();
            bossGroup.shutdownGracefully().sync();
 }
 }
}

服务端心跳处理Handler

IdleStateHandler 是实现心跳的关键, 它会根据不同的 IO idle 类型来产生不同的 IdleStateEvent 事件, 而这个事件的捕获, 其实就是在 userEventTriggered 方法中实现的.

public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter {
 private int lossConnectCount = 0;
 @Override
 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
 System.out.println("已经5秒未收到客户端的消息了！");
 if (evt instanceof IdleStateEvent){
 IdleStateEvent event = (IdleStateEvent)evt;
 if (event.state()== IdleState.READER_IDLE){
                lossConnectCount++;
 if (lossConnectCount>2){
 System.out.println("关闭这个不活跃通道！");
                    ctx.channel().close();
 }
 }
 }else {
 super.userEventTriggered(ctx,evt);
 }
 }
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        lossConnectCount = 0;
 System.out.println("client says: "+msg.toString());
 }
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
 }
}

客户端启动初始化代码

心跳的配置也是跟服务端一样，往pipeline中添加IdleStateHandler，其中的参数可以自己随意配置。

public class HeartBeatClient {
 public static void main(String[] args) throws Exception {
 EventLoopGroup group = new NioEventLoopGroup();
 try {
 Bootstrap b = new Bootstrap();
        b.group(group).channel(NioSocketChannel.class)
 .handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new IdleStateHandler(0,4,0, TimeUnit.SECONDS));
                        socketChannel.pipeline().addLast(new StringEncoder());
                        socketChannel.pipeline().addLast(new HeartBeatClientHandler());
 }
 });
 ChannelFuture f = b.connect(new InetSocketAddress(8089)).sync();
            f.channel().closeFuture().sync();
 } catch (InterruptedException e) {
            e.printStackTrace();
 }finally {
 group.shutdownGracefully().sync();
 }
 }
}

客户端心跳处理代码

关键也是在userEventTriggered方法中实现的，主要的逻辑就是往服务端发送心跳，发了3次就不发了，这时候就会触发服务端的userEventTriggered中lossConnectCount 如果超过2次就把这个通道给断开。也就是把这个客户端给断开。

public class HeartBeatClientHandler extends ChannelInboundHandlerAdapter {
 private int curTime = 0;
 private int beatTime = 3;
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 System.out.println("==channelRead===");
 System.out.println(msg.toString());
 }
 @Override
 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
 System.out.println("客户端循环心跳监测发送: "+new Date());
 if (evt instanceof IdleStateEvent){
 IdleStateEvent event = (IdleStateEvent)evt;
 if (event.state()== IdleState.WRITER_IDLE){
 if (curTime<beatTime) {
                    curTime++;
                    ctx.writeAndFlush("biubiu");
 }
 }
 }
 }
}

小结

Netty心跳的做法大致就是如此，

1.利用 IdleStateHandler来产生对应的 idle 事件。 2.userEventTriggered中做好心跳交互逻辑。

至于更加复杂的逻辑，还是各位遇到的时候自己发挥。

断线重连

服务端代码依旧是上面的不变。

客户端

主要工作以及初始化代码如下：

1.通过 channel().eventLoop().schedule来延时10s 后尝试重新连接.

public class ReconnectClient {
 private NioEventLoopGroup workGroup = new NioEventLoopGroup(4);
 private Channel channel;
 private Bootstrap bootstrap;
 public static void main(String[] args) throws Exception {
 ReconnectClient client = new ReconnectClient();
        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()) {
                channel.writeAndFlush("ReconnectClient心跳來了呀.....");
 }
 Thread.sleep(random.nextInt(20000));
 }
 }
 public void start() {
 try {
            bootstrap = new Bootstrap();
            bootstrap
 .group(workGroup)
 .channel(NioSocketChannel.class)
 .handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new IdleStateHandler(0,4,0, TimeUnit.SECONDS));
                            ch.pipeline().addLast(new StringEncoder());
                            ch.pipeline().addLast(new ReconnectClientHandler(ReconnectClient.this));
 }
 });
            doConnect();
 } catch (Exception e) {
 throw new RuntimeException(e);
 }
 }
 protected void doConnect() {
 if (channel != null && channel.isActive()) {
 return;
 }
 ChannelFuture future = bootstrap.connect(new InetSocketAddress(8089));
        future.addListener(new ChannelFutureListener() {
 @Override
 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() {
 @Override
 public void run() {
                                doConnect();
 }
 }, 10, TimeUnit.SECONDS);
 }
 }
 });
 }
}

断线重连处理Handler

2.断线重连的关键一点是检测连接是否已经断开. 因此我们重写了 channelInactive 方法. 当 TCP 连接断开时, 会回调channelInactive方法, 因此我们在这个方法中调用 client.doConnect() 来进行重连.

public class ReconnectClientHandler extends ChannelInboundHandlerAdapter {
 private int curTime = 0;
 private int beatTime = 3;
 private ReconnectClient client;
 public ReconnectClientHandler(ReconnectClient client) {
 this.client = client;
 }
 @Override
 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
 System.out.println("ReconnectClientHandler客户端循环心跳监测发送: "+new Date());
 if (evt instanceof IdleStateEvent){
 IdleStateEvent event = (IdleStateEvent)evt;
 if (event.state()== IdleState.WRITER_IDLE){
 if (curTime<beatTime) {
                    curTime++;
                    ctx.writeAndFlush("ReconnectClientHandler=biubiu.....");
 }
 }
 }
 }
 @Override
 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 super.channelInactive(ctx);
        client.doConnect();
 System.out.println("重新連接了呀。。。。");
 }
}

总结

心跳机制与断线重连的基本步骤如上所述。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2019-06-25，如有侵权请联系 cloudcommunity@tencent.com 删除

android

java

本文分享自爱编码微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

android

java

登录后参与评论

0 条评论

热度