AI 助理
备案控制台
开发者社区 Java技术栈 文章 正文

Netty 中的心跳机制，还有谁不会？

2021-12-20 326
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 我们知道在TCP长连接或者WebSocket长连接中一般我们都会使用心跳机制–即发送特殊的数据包来通告对方自己的业务还没有办完，不要关闭链接。

我们知道在TCP长连接或者WebSocket长连接中一般我们都会使用心跳机制–即发送特殊的数据包来通告对方自己的业务还没有办完，不要关闭链接。


那么心跳机制可以用来做什么呢？


我们知道网络的传输是不可靠的，当我们发起一个链接请求的过程之中会发生什么事情谁都无法预料，或者断电，服务器重启，断网线之类。


如果有这种情况的发生对方也无法判断你是否还在线。所以这时候我们引入心跳机制，在长链接中双方没有数据交互的时候互相发送数据(可能是空包，也可能是特殊数据)，对方收到该数据之后也回复相应的数据用以确保双方都在线，这样就可以确保当前链接是有效的。


1. 如何实现心跳机制

一般实现心跳机制由两种方式：


TCP协议自带的心跳机制来实现；

在应用层来实现。

但是TCP协议自带的心跳机制系统默认是设置的是2小时的心跳频率。它检查不到机器断电、网线拔出、防火墙这些断线。而且逻辑层处理断线可能也不是那么好处理。另外该心跳机制是与TCP协议绑定的，那如果我们要是使用UDP协议岂不是用不了？所以一般我们都不用。


而一般我们自己实现呢大致的策略是这样的：


Client启动一个定时器，不断发送心跳；

Server收到心跳后，做出回应；

Server启动一个定时器，判断Client是否存在，这里做判断有两种方法：时间差和简单标识。

时间差：


收到一个心跳包之后记录当前时间；

判断定时器到达时间，计算多久没收到心跳时间=当前时间-上次收到心跳时间。如果改时间大于设定值则认为超时。

简单标识：


收到心跳后设置连接标识为true;

判断定时器到达时间，如果未收到心跳则设置连接标识为false;

今天我们来看一下Netty的心跳机制的实现，在Netty中提供了IdleStateHandler类来进行心跳的处理，它可以对一个 Channel 的 读/写设置定时器, 当 Channel 在一定事件间隔内没有数据交互时(即处于 idle 状态), 就会触发指定的事件。


该类可以对三种类型的超时做心跳机制检测：


public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {

   this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);

}

readerIdleTimeSeconds：设置读超时时间；

writerIdleTimeSeconds：设置写超时时间；

allIdleTimeSeconds：同时为读或写设置超时时间；

下面我们还是通过一个例子来讲解IdleStateHandler的使用。


服务端：

public class HeartBeatServer {
    private int port;
    public HeartBeatServer(int port) {
        this.port = port;
    }
    public void start(){
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        ServerBootstrap server = new ServerBootstrap().group(bossGroup,workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new HeartBeatServerChannelInitializer());
        try {
            ChannelFuture future = server.bind(port).sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
    public static void main(String[] args) {
        HeartBeatServer server = new HeartBeatServer(7788);
        server.start();
    }
}

image.png

public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter{
    private int loss_connect_time = 0;
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + "Server :" + msg.toString());
    }
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if(evt instanceof IdleStateEvent){
            //服务端对应着读事件，当为READER_IDLE时触发
                IdleStateEvent event = (IdleStateEvent)evt;
            if(event.state() == IdleState.READER_IDLE){
                loss_connect_time++;
                System.out.println("接收消息超时");
                if(loss_connect_time > 2){
                    System.out.println("关闭不活动的链接");
                    ctx.channel().close();
                }
            }else{
                super.userEventTriggered(ctx,evt);
            }
        }
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

我们看到在handler中调用了userEventTriggered方法，IdleStateEvent的state()方法一个有三个值： READER_IDLE，WRITER_IDLE，ALL_IDLE。正好对应读事件写事件和读写事件。

再来写一下客户端：

public class HeartBeatsClient {
    private  int port;
    private  String address;
    public HeartBeatsClient(int port, String address) {
        this.port = port;
        this.address = address;
    }
    public void start(){
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new HeartBeatsClientChannelInitializer());
        try {
            ChannelFuture future = bootstrap.connect(address,port).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            group.shutdownGracefully();
        }
    }
    public static void main(String[] args) {
        HeartBeatsClient client = new HeartBeatsClient(7788,"127.0.0.1");
        client.start();
    }
}

image.png

public class HeartBeatClientHandler extends ChannelInboundHandlerAdapter {
    private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Heartbeat",
            CharsetUtil.UTF_8));
    private static final int TRY_TIMES = 3;
    private int currentTime = 0;
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("激活时间是："+new Date());
        System.out.println("链接已经激活");
        ctx.fireChannelActive();
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("停止时间是："+new Date());
        System.out.println("关闭链接");
    }
    @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(currentTime <= TRY_TIMES){
                    System.out.println("currentTime:"+currentTime);
                    currentTime++;
                    ctx.channel().writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
                }
            }
        }
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String message = (String) msg;
        System.out.println(message);
        if (message.equals("Heartbeat")) {
            ctx.write("has read message from server");
            ctx.flush();
        }
        ReferenceCountUtil.release(msg);
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

启动服务端和客户端我们看到输出为：


image.png

我们再来屡一下思路：


首先客户端激活channel，因为客户端中并没有发送消息所以会触发客户端的IdleStateHandler，它设置的写超时时间为3s；

然后触发客户端的事件机制进入userEventTriggered方法，在触发器中计数并向客户端发送消息；

服务端接收消息；

客户端触发器继续轮询发送消息，直到计数器满不再向服务端发送消息；

服务端在IdleStateHandler设置的读消息超时时间5s内未收到消息，触发了服务端中handler的userEventTriggered方法，于是关闭客户端的链接。

大体我们的简单心跳机制就是这样的思路，通过事件触发机制以及计数器的方式来实现，上面我们的案例中最后客户端没有发送消息的时候我们是强制断开了客户端的链接，那么既然可以关闭，我们是不是也可是重新链接客户端呢？因为万一客户端本身并不想关闭而是由于别的原因导致他无法与服务端通信。下面我们来说一下重连机制。


当我们的服务端在未读到客户端消息超时而关闭客户端的时候我们一般在客户端的finally块中方的是关闭客户端的代码，这时我们可以做一下修改的，finally是一定会被执行新的，所以我们可以在finally块中重新调用一下启动客户端的代码，这样就又重新启动了客户端了，上客户端代码：

/**
 * 本Client为测试netty重连机制
 * Server端代码都一样，所以不做修改
 * 只用在client端中做一下判断即可
 */
public class HeartBeatsClient2 {
    private  int port;
    private  String address;
    ChannelFuture future;
    public HeartBeatsClient2(int port, String address) {
        this.port = port;
        this.address = address;
    }
    public void start(){
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new HeartBeatsClientChannelInitializer());
        try {
            future = bootstrap.connect(address,port).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            //group.shutdownGracefully();
            if (null != future) {
                if (future.channel() != null && future.channel().isOpen()) {
                    future.channel().close();
                }
            }
            System.out.println("准备重连");
            start();
            System.out.println("重连成功");
        }
    }
    public static void main(String[] args) {
        HeartBeatsClient2 client = new HeartBeatsClient2(7788,"127.0.0.1");
        client.start();
    }
}

其余部分的代码与上面的实例并无异同，只需改造客户端即可，我们再运行服务端和客户端会看到客户端虽然被关闭了，但是立马又被重启：

image.png

文章标签：
网络安全
网络协议
关键词：
Netty心跳机制
Netty机制
Netty心跳
技术君
目录
相关文章
竹子爱熊猫
|
移动开发 网络协议 算法
(十)Netty进阶篇：漫谈网络粘包、半包问题、解码器与长连接、心跳机制实战
在前面关于《Netty入门篇》的文章中，咱们已经初步对Netty这个著名的网络框架有了认知，本章的目的则是承接上文，再对Netty中的一些进阶知识进行阐述，毕竟前面的内容中，仅阐述了一些Netty的核心组件，想要真正掌握Netty框架，对于它我们应该具备更为全面的认知。
竹子爱熊猫
824 2
一只牛博
|
存储 缓存 运维
时间轮奇妙旅程：深度解析Netty中的时间轮机制
时间轮奇妙旅程：深度解析Netty中的时间轮机制
一只牛博
596 1
得过且过的勇者y
|
网络协议 调度
Netty心跳检测
客户端的心跳检测对于任何长连接的应用来说，都是一个非常基础的功能。要理解心跳的重要性，首先需要从网络连接假死的现象说起。
得过且过的勇者y
308 1
小小工匠
|
编解码 开发者
Netty Review - 深入理解Netty: ChannelHandler的生命周期与事件处理机制
Netty Review - 深入理解Netty: ChannelHandler的生命周期与事件处理机制
小小工匠
357 0
小小工匠
|
前端开发 UED
Netty Review - Netty自动重连机制揭秘：原理与最佳实践
Netty Review - Netty自动重连机制揭秘：原理与最佳实践
小小工匠
404 0
小小工匠
|
监控 网络协议 调度
Netty Review - 深入探讨Netty的心跳检测机制：原理、实战、IdleStateHandler源码分析
Netty Review - 深入探讨Netty的心跳检测机制：原理、实战、IdleStateHandler源码分析
小小工匠
807 0
杨宸杨
|
Nacos
Netty自定义消息协议的实现逻辑处理粘包拆包、心跳机制
Netty自定义消息协议的实现逻辑处理粘包拆包、心跳机制
杨宸杨
349 0
17863961702
Netty（五）之心跳机制与重连
文章目标 1）实现客户端和服务端的心跳 2）心跳多少次没有应答断开处理 3）客户端宕机通知服务端 4）服务端宕机客户端重连
17863961702
303 0
Netty（五）之心跳机制与重连
kunfire
|
缓存 移动开发 网络协议
第 8 章 Netty 编解码器和 Handler 调用机制
第 8 章 Netty 编解码器和 Handler 调用机制
kunfire
352 0
六脉神剑.
|
网络协议 算法
小六六学Netty系列之编解码器和handler的调用机制（下）
前言 文本已收录至我的GitHub仓库，欢迎Star：github.com/bin39232820… 种一棵树最好的时间是十年前，其次是现在
六脉神剑.
224 0