转载

Netty学习篇④-心跳机制及断线重连

心跳检测

  1. 前言

    客户端和服务端的连接属于socket连接，也属于长连接，往往会存在客户端在连接了服务端之后就没有任何操作了，但还是占用了一个连接；当越来越多类似的客户端出现就会浪费很多连接，netty中可以通过心跳检测来找出一定程度（自定义规则判断哪些连接是无效链接）的无效链接并断开连接，保存真正活跃的连接。

  2. 什么叫心跳检测

    我理解的心跳检测应该是客户端/服务端定时发送一个数据包给服务端/客户端，检测对方是否有响应；
如果是存活的连接，在一定的时间内应该会收到响应回来的数据包；
如果在一定时间内还是收不到接收方的响应的话，就可以当做是挂机，可以断开此连接；
如果检测到了掉线之后还可以进行重连；

  3. 心跳检测的实现

    • TCP自带心跳检测，协议层采用Keeplive机制默认2小时频率触发一次检测，但是它存在缺陷:检测不出网线拔出、防火墙、使用起来不灵活、依赖操作系统等
    • Netty可以通过IdleStateHandler来实现心跳检测，使用起来也非常方便清晰

  4. IdleStateHandler原理

    idleStateHandler在通道注册之后会开启一个定时任务，定时去检测通道中后续是否还有进行数据传输，如果在规定的时间内没有进行数据传输则会触发对应的超时事件，使用者可以根据对应的事件自定义规则来判别当前连接是否是活跃，是否需要关闭连接等来进行操作。
一般idleStateHandler触发的事件IdleStateEvent会在心跳handler中的userEventTriggered方法中捕获到对应的超时事件。

    IdleStateHandler的继承关系：通过ChannelDuplexHandler类继承ChannelInboundHandler和实现ChannelOutboundHandler来实现对入站和出站的重写和监控

    Netty学习篇④-心跳机制及断线重连

  5. 源码分析

    • IdleStateHandler初始化：为0代表不监控

      /**
* @observeOutput 观察输出
* @readerIdleTime 读超时时间 自定义时间内检测Channel通道有没有读取到数据，为0代表不监控
* @writerIdleTime 写超时时间 自定义时间内检测Channel通道有没有write数据，为0代表不监控
* @allIdleTime 总超时时间 自定义时间内检测Channel通道有没有读/写数据，为0代表不监控
* @unit 时间单位
*/
public IdleStateHandler(boolean observeOutput,
            long readerIdleTime, long writerIdleTime, long allIdleTime,
            TimeUnit unit) {
        if (unit == null) {
            throw new NullPointerException("unit");
        }

        this.observeOutput = observeOutput;

       // 初始化读取空闲时间，最小值为0
        if (readerIdleTime <= 0) {
            readerIdleTimeNanos = 0;
        } else {
            // 定义读取超时时间为自定义设置时间
            readerIdleTimeNanos = Math.max(unit.toNanos(readerIdleTime), MIN_TIMEOUT_NANOS);
        }
        if (writerIdleTime <= 0) {
            writerIdleTimeNanos = 0;
        } else {
            // 设置写超时时间
            writerIdleTimeNanos = Math.max(unit.toNanos(writerIdleTime), MIN_TIMEOUT_NANOS);
        }
        if (allIdleTime <= 0) {
            allIdleTimeNanos = 0;
        } else {
            // 设置总超时时间
            allIdleTimeNanos = Math.max(unit.toNanos(allIdleTime), MIN_TIMEOUT_NANOS);
        }
    }

    • IdleStateHandler和channel通道的关联：通过类图可以得知，idleStateHandler可以重写入站和出站的方法，不过只是通过channelRead和write方法来记录阅读的时间等不做其他操作 

      @Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    // 初始化检测器，开启定时任务
    initialize(ctx);
    super.channelActive(ctx);
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    // 在通道非活跃状态的时候销毁定时任务
    destroy();
    super.channelInactive(ctx);
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    // 如果设置的读超时时间大于0则设置是否读操作为true
    if (readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
        reading = true;
        firstReaderIdleEvent = firstAllIdleEvent = true;
    }
    // 记录时间和标识标志之后就直接fire当前read到下一个ChannelHandler处理类中
    ctx.fireChannelRead(msg);
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    // 当读取完毕时，设置是否正在读取为false，设置最后读取时间为系统当前时间
    if ((readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) && reading) {
        lastReadTime = ticksInNanos();
        reading = false;
    }
    // 同样直接fire掉，跳入到下一个handler中
    ctx.fireChannelReadComplete();
}

// idleStateHandler重写的write方法
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    // Allow writing with void promise if handler is only configured for read timeout events.
    if (writerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
        ctx.write(msg, promise.unvoid()).addListener(writeListener);
    } else {
        ctx.write(msg, promise);
    }
}

// ChannelOutboundHandler提供的write接口方法
/**
    * Called once a write operation is made. The write operation will write the messages through the
     * {@link ChannelPipeline}. Those are then ready to be flushed to the actual {@link Channel} once
     * {@link Channel#flush()} is called
     * 执行一次写操作；写操作通过ChannelPipeline来传输信息；最后通过channel的flush()方法来刷新
     *
     * @param ctx               the {@link ChannelHandlerContext} for which the write operation is made 实际写操作者
     * @param msg               the message to write 写的消息
     * @param promise           the {@link ChannelPromise} to notify once the operation completes 在操作完成时立即通知（类似future异步通知）
     * @throws Exception        thrown if an error occurs
     */
    void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;

    • 定时器初始化/销毁：将使用者自定义的超时时间设置为延迟定时任务 

      初始化：

private void initialize(ChannelHandlerContext ctx) {
        // Avoid the case where destroy() is called before scheduling timeouts.
        // See: https://github.com/netty/netty/issues/143
       // state; // 0 - none, 1 - 初始化, 2 - 销毁
        switch (state) {
        case 1:
        case 2:
            return;
        }

        state = 1;
        initOutputChanged(ctx);
       // 最后读取时间
        lastReadTime = lastWriteTime = ticksInNanos();
       // 如果设置的空闲时间大于0则开启定时任务进行监控
        if (readerIdleTimeNanos > 0) {
            readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
                    readerIdleTimeNanos, TimeUnit.NANOSECONDS);
        }
        // 写超时时间
        if (writerIdleTimeNanos > 0) {
            writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
                    writerIdleTimeNanos, TimeUnit.NANOSECONDS);
        }
        // 总超时时间
        if (allIdleTimeNanos > 0) {
            allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
                    allIdleTimeNanos, TimeUnit.NANOSECONDS);
        }
    }
      销毁：

/**
* @readerIdleTimeout ==> ScheduledFuture类
* @writerIdleTimeout ==> ScheduledFuture类
*/
private void destroy() {
       // 设置销毁状态
        state = 2;
       // 销毁线程
       // ScheduledFuture
        if (readerIdleTimeout != null) {
            readerIdleTimeout.cancel(false);
            readerIdleTimeout = null;
        }
        if (writerIdleTimeout != null) {
            writerIdleTimeout.cancel(false);
            writerIdleTimeout = null;
        }
        if (allIdleTimeout != null) {
            allIdleTimeout.cancel(false);
            allIdleTimeout = null;
        }
    }

    • 开启定时任务

      ScheduledFuture<?> schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit) {
        return ctx.executor().schedule(task, delay, unit);
    }

    • 读/写定时任务：定时任务启动的时候，通过设置的超时时间和上一次触发channelRead的时间进行相减比较来判断是否超时了 

      // 仅分析读取超时时间定时任务，写超时差不多就是触发的时间不一样，比对的变量换成了设置的写超时时间
private final class ReaderIdleTimeoutTask extends AbstractIdleTask {

        ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
            super(ctx);
        }

        @Override
        protected void run(ChannelHandlerContext ctx) {
            // readerIdleTimeNanos：初始化IdleStateHandler设置的读取超时时间
            long nextDelay = readerIdleTimeNanos;
            // 如果没有任何读取操作
            if (!reading) {
                // 判断是否有超时
                // nextDelay = nextDelay-（ticksInNanos() - lastReadTime）
                // 即设置的超时时间减去距离上一次读取的时间
                nextDelay -= ticksInNanos() - lastReadTime;
            }
           // 如果小于等于0 则触发读取超时事件,设置新的延迟时间
            if (nextDelay <= 0) {
                // Reader is idle - set a new timeout and notify the callback.
                readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
               // 标记为第一次
                boolean first = firstReaderIdleEvent;
                // 设置成非第一次
                firstReaderIdleEvent = false;

                try {
                    IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
                    channelIdle(ctx, event);
                } catch (Throwable t) {
                    ctx.fireExceptionCaught(t);
                }
            } else {
                // Read occurred before the timeout - set a new timeout with shorter delay.
                // 超时的时候发生的读取事件，则重新延迟执行
                readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
            }
        }
    }

  6. 项目实战(主要代码，项目以服务端作为心跳监控，也可以在客户端进行心跳监控)

    • 项目结构

      ├─src
│  ├─main
│  │  ├─java
│  │  │  └─com
│  │  │      └─hetangyuese
│  │  │          └─netty
│  │  │              ├─client
│  │  │              │      MyChannelFutureListener.java
│  │  │              │      MyClient05.java
│  │  │              │      MyClientChannelHandler.java
│  │  │              │      MyClientChannelInitializer.java
│  │  │              │
│  │  │              └─server
│  │  │                  │  MyServer05.java
│  │  │                  │  MyServerChannelInitializer.java
│  │  │                  │  MyServerHandler.java
│  │  │                  │
│  │  │                  └─decoder

      ​

    • 在ChannelPipeline中注册IdleStateHandler

      public class MyServerChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().
                addLast(new StringEncoder(Charset.forName("GBK")))
                .addLast(new StringDecoder(Charset.forName("GBK")))
                .addLast(new LoggingHandler(LogLevel.INFO))
               // 设置读取超时时间为5秒，写超时和总超时为0即不做监控
                .addLast(new IdleStateHandler(5, 0, 0))
                .addLast(new MyServerHandler());
    }
}

    • 服务端处理handler(其中userEventTriggered为接收心跳任务触发的事件，这次做了计数三次触发读空闲超时则断开连接)

      public class MyServerHandler extends ChannelInboundHandlerAdapter {

    private AtomicInteger count = new AtomicInteger(1);

    /**
     *  心跳检测机制会进入
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        System.out.println("心跳检测触发了事件, object: , time: " + evt + new Date().toLocaleString());
        super.userEventTriggered(ctx, evt);
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent e = (IdleStateEvent) evt;
            // 客户端连接应该是请求 write
            if (e.state() == IdleState.READER_IDLE) {
                System.out.println("服务端监测到了读取超时");
                count.incrementAndGet();
                if (count.get() > 3) {
                    System.out.println("客户端还在？？ 已经3次检测没有访问了，我要断开了哦！！！");
                    ctx.channel().close();
                }
            } else if (e.state() == IdleState.WRITER_IDLE) {
                // 如果一直有交互则会发送writer_idle
                System.out.println("服务端收到了写入超时");
            } else {
                System.out.println("服务端收到了All_idle");
            }
        } else {
            super.userEventTriggered(ctx,evt);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("myServerHandler is active, time: " + new Date().toLocaleString());
        ctx.writeAndFlush("成功连接服务端, 当前时间：" + new Date().toLocaleString());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("服务端与客户端断开了连接, time: " + new Date().toLocaleString());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("myServerHandler 收到了客户端的信息 msg:" + msg + ", time: " + new Date().toLocaleString());
        ctx.writeAndFlush("您好，客户端，我是服务端");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

    • 运行结果

      myServer is start time: 2019-11-8 14:47:16
myServerHandler is active, time: 2019-11-8 14:47:18
十一月 08, 2019 2:47:18 下午 io.netty.handler.logging.LoggingHandler channelRegistered
信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] REGISTERED
十一月 08, 2019 2:47:18 下午 io.netty.handler.logging.LoggingHandler channelActive
信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] ACTIVE
十一月 08, 2019 2:47:18 下午 io.netty.handler.logging.LoggingHandler write
信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] WRITE: 成功连接服务端, 当前时间：2019-11-8 14:47:18
十一月 08, 2019 2:47:18 下午 io.netty.handler.logging.LoggingHandler flush
信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] FLUSH
心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@a3a0212019-11-8 14:47:23
服务端监测到了读取超时
心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@18692702019-11-8 14:47:28
服务端监测到了读取超时
心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@18692702019-11-8 14:47:33
服务端监测到了读取超时
客户端还在？？ 已经3次检测没有访问了，我要断开了哦！！！
十一月 08, 2019 2:47:33 下午 io.netty.handler.logging.LoggingHandler close
信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] CLOSE
十一月 08, 2019 2:47:33 下午 io.netty.handler.logging.LoggingHandler channelInactive
信息: [id: 0x8d924d06, L:/127.0.0.1:9001 ! R:/127.0.0.1:62195] INACTIVE
服务端与客户端断开了连接, time: 2019-11-8 14:47:33
十一月 08, 2019 2:47:33 下午 io.netty.handler.logging.LoggingHandler channelUnregistered
信息: [id: 0x8d924d06, L:/127.0.0.1:9001 ! R:/127.0.0.1:62195] UNREGISTERED

以上就是心跳监控的所有流程了，合理的利用Netty的心跳机制可以有效的剔除一些无用的连接释放些资源

Netty断线重连：在长连接中有时候出现断开的时候可以重新连接

出现断线重连的情况：

  • 首次连接但是连接不上，通过ChannelFutureListener增加监控进行重连
  • 由于网络原因等、自动断开等 通过在channelInactive中进行重连即可

  1. ChannelFutureListener进行重连

    public class MyChannelFutureListener implements ChannelFutureListener {

    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        if (future.isSuccess()) {
            System.out.println("当前已连接");
            return;
        }
        System.out.println("启动连接客户端失败，开始重连");
        final EventLoop loop = future.channel().eventLoop();
        loop.schedule(new Runnable() {
            @Override
            public void run() {
                try {
                   MyClient05.reConnection();
                   System.out.println("客户端重连成功");
                } catch (Exception e){
                    e.printStackTrace();
                }
            }
        }, 1L, TimeUnit.SECONDS);
    }
}

    client启动不知道为啥想要测试listener的效果时，connect的sync()不能带，是由于同步阻塞的原因？ 

    public void start() {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            bootstrap = getBootstrap();

            bootstrap.group(group)
                    .option(ChannelOption.AUTO_READ, true)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .channel(NioSocketChannel.class)
                    .handler(new MyClientChannelInitializer());

            // ChannelFuture future = bootstrap.connect(new InetSocketAddress(ip, port)).sync();
             ChannelFuture future = bootstrap.connect(new InetSocketAddress(ip, port));      
            // 增加监听 
            future.addListener(new MyChannelFutureListener());
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }

    直接启动客户端，可以看到listener输出：启动连接客户端失败，开始重连

  2. channelInactive进行重连(我是直接new了一个线程去重连)

    @Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    System.out.println("客户端断开了连接, time: " + new Date().toLocaleString());
    new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                new MyClient05("127.0.0.1", 9001).start();
                System.out.println("客户端重新连接了服务端 time：" + new Date().toLocaleString());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }).start();
}

    测试方法：

    1.可以直接通过上面的心跳机制断开连接后，客户端的channelInactive检测到断开会自动执行重连

    测试结果：

    服务端：
------
心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@ab81552019-11-8 16:43:12
服务端监测到了读取超时
心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@15192622019-11-8 16:43:17
服务端监测到了读取超时
心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@15192622019-11-8 16:43:22
服务端监测到了读取超时
客户端还在？？ 已经3次检测没有访问了，我要断开了哦！！！
十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler close
信息: [id: 0x6bfc0d90, L:/192.168.0.118:9001 - R:/192.168.0.118:51031] CLOSE
十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler channelInactive
信息: [id: 0x6bfc0d90, L:/192.168.0.118:9001 ! R:/192.168.0.118:51031] INACTIVE
十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler channelUnregistered
信息: [id: 0x6bfc0d90, L:/192.168.0.118:9001 ! R:/192.168.0.118:51031] UNREGISTERED
服务端与客户端断开了连接, time: 2019-11-8 16:43:22
十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler channelRegistered
信息: [id: 0xbcb2ec62, L:/127.0.0.1:9001 - R:/127.0.0.1:51061] REGISTERED
十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler channelActive
信息: [id: 0xbcb2ec62, L:/127.0.0.1:9001 - R:/127.0.0.1:51061] ACTIVE
myServerHandler is active, time: 2019-11-8 16:43:22
十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler write
信息: [id: 0xbcb2ec62, L:/127.0.0.1:9001 - R:/127.0.0.1:51061] WRITE: 成功连接服务端, 当前时间：2019-11-8 16:43:22
十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler flush
信息: [id: 0xbcb2ec62, L:/127.0.0.1:9001 - R:/127.0.0.1:51061] FLUSH
十一月 08, 2019 4:43:26 下午 io.netty.handler.logging.LoggingHandler channelReadComplete
------------------------------------------------------------------------------------- 

客户端：
-------
当前已连接
客户端与服务端建立了连接 time: 2019-11-8 16:43:07
客户端接收到了服务的响应的数据 msg: 成功连接服务端, 当前时间：2019-11-8 16:43:07, time: 2019-11-8 16:43:07
客户端断开了连接, time: 2019-11-8 16:43:22
当前已连接
客户端与服务端建立了连接 time: 2019-11-8 16:43:22
客户端接收到了服务的响应的数据 msg: 成功连接服务端, 当前时间：2019-11-8 16:43:22, time: 2019-11-8 16:43:22
原文  http://www.cnblogs.com/hetangyuese/p/11821519.html
正文到此结束
所属分类： 编程技术 Java

热门推荐

相关文章

阿里云首购8折
说给你听

本文目录
随机标签
书籍教程
springboot-demo Java入门教程 bootstrap3 CSS Apache基础教程 php ionic 教程 Python mysql教程 eclipse Ubuntu VPS系统配置 AngularJS 教程 MongoDB教程 Struts2教程 Redis教程 Spring教程 Git教程 Jenkins进阶系列 openfire参考指南 Java设计模式 HBase教程 java-demo Maven教程 springcloud-demo hibernate教程 Docker 教程 memcached教程 Quartz指南 Hive教程 ANTLR教程 SpringCloud Ant教程 java实例教程 Hazelcast教程 XStream教程 Elastic-Job-Lite 深入浅出MyBatis ibaties教程 SVN教程 rabittmq教程 Hadoop教程 solr教程 WebService CXF学习 JPA教程 ActiveMQ中文指南 Java内存模型 dubbo教程 python3-demo Linux入门视频教程
近期评论
  1. 1 使用 Docker 打包和运行 Vue 应用
  2. 2 Dependency Check命令行方式扫描jar包的安全漏洞
  3. 3 maven使用Dependency-Check来扫描安全漏洞
  4. 4 使用sonarqube分析代码质量
  5. 5 使用screw来对比数据库表和字段差异
  6. 6 Spring Boot 实现防盗链功能
  7. 7 在Spring Boot中的实现国际化（i18n）
  8. 8 一个用于 SMB/CIFS 网络的 Java 库-JCIFS
  9. 9 Spring Boot 3.x嵌入MongoDB 进行测试
  10. 10 java如何解析和生成yaml文件？
网站信息
  • 文章总数：82,737 篇
  • 文件总数：284,312 个
  • 标签总数：2,416 个
  • 分类总数：85 个
  • 留言数量：2,559 条
  • 在线人数：657 人
  • 运行天数：4,439天
  • 最后更新：2024年12月22日15点
Loading...

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

[HBLOG]公众号 HBLOG HBLOG