Netty心跳机制

Posted 2021-04-30 开源java学习

一、何为心跳

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

二、为什么需要心跳

因为网络的不可靠性, 有可能在 TCP 保持长连接的过程中, 由于某些突发情况, 例如网线被拔出, 突然掉电等, 会造成服务器和客户端的连接中断. 在这些突发情况下, 如果恰好服务器和客户端之间没有交互的话, 那么它们是不能在短时间内发现对方已经掉线的. 为了解决这个问题, 我们就需要引入 心跳 机制. 心跳机制的工作原理是: 在服务器和客户端之间一定时间内没有数据交互时, 即处于 idle 状态时, 客户端或服务器会发送一个特殊的数据包给对方, 当接收方收到这个数据报文后, 也立即发送一个特殊的数据报文, 回应发送方, 此即一个 PING-PONG 交互. 自然地, 当某一端收到心跳消息后, 就知道了对方仍然在线, 这就确保 TCP 连接的有效性.

三、心跳实现
使用TCP协议层的Keeplive机制，但是该机制默认的心跳时间是2小时，依赖操作系统实现不够灵活；

心跳机制一般来说都是在逻辑层发送空的包来实现的，比如Netty的IdleStateHandler类实现心跳机制。

心跳机制实现逻辑：每隔几分钟发送一个固定信息给服务端，服务端收到后回复一个固定信息给客户端，如果服务端几分钟内没有收到客户端信息则视客户端断开。

在Netty中IdleStateHandler主要用来检测远端是否存活，如果不存活或活跃则对空闲Socket连接进行处理避免资源的浪费；IdleStateHandler实现对三种心跳的检测，分别是readerIdleTime、writerIdleTime和allIdleTime，参数解释如下： 
1）readerIdleTime：读超时时间
2）writerIdleTime：写超时时间
3）allIdleTime：所有类型的超时时间

所以在channelPipeline中加入IdleStateHandler，我们在handler中提示的是5秒读，所以我们服务端的配置的是：

ph.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));

因为服务端必须5秒接受一次心跳请求，那么客户端的配置:

ph.addLast( new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS));

userEventTriggered是Netty 处理心跳超时事件，在IdleStateHandler设置超时时间，如果达到了，就会直接调用该方法。如果没有超时则不调用。我们重写该方法的话，就可以自行进行相关的业务逻辑处理了。

四、IdleStateHandler心跳检测实例
a、服务端
HeartNettyServer——服务端启动类

package com.dxfx.netty.demo;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioserverSocketChannel;

/**
 * 服务端启动类
 *
 * @author Administrator
 *
 */
public class HeartNettyServer {
 public static void main(String[] args) throws InterruptedException {
 // 首先，netty通过ServerBootstrap启动服务端
         ServerBootstrap server = new ServerBootstrap();
 EventLoopGroup parentGroup = new NioEventLoopGroup();
 EventLoopGroup childGroup =new NioEventLoopGroup();
 //第1步定义两个线程组，用来处理客户端通道的accept和读写事件
 //parentGroup用来处理accept事件，childgroup用来处理通道的读写事件
 //parentGroup获取客户端连接，连接接收到之后再将连接转发给childgroup去处理
 server.group(parentGroup, childGroup);
 
 //用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。
 //用来初始化服务端可连接队列
 //服务端处理客户端连接请求是按顺序处理的，所以同一时间只能处理一个客户端连接，多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小。
         server.option(ChannelOption.SO_BACKLOG, 128);
 
 //第2步绑定服务端通道
         server.channel(NioServerSocketChannel.class);
 
 //第3步绑定handler，处理读写事件，ChannelInitializer是给通道初始化
         server.childHandler(new HeartNettyServerFilter());
 
 //第4步绑定8080端口
         ChannelFuture future = server.bind(8080).sync();
 //当通道关闭了，就继续往下走
 future.channel().closeFuture().sync();
 }
 
}

HeartNettyServerFilter——服务端过滤器，如编解码和心跳的设置

package com.dxfx.netty.demo;

import java.util.concurrent.TimeUnit;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * 服务端过滤器，如编解码和心跳的设置
 *
 * @author Administrator
 *
 */
public class HeartNettyServerFilter extends ChannelInitializer<SocketChannel> {

 @Override
 protected void initChannel(SocketChannel sc) throws Exception {
 ChannelPipeline cp = sc.pipeline();
 cp.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
 
 // 解码和编码，应和客户端一致
        cp.addLast(new StringDecoder());
 cp.addLast(new StringEncoder());
 
 //处理服务端的业务逻辑
        cp.addLast(new HeartNettyServerHandler());
 }
}

HeartNettyServerHandler——处理服务端业务逻辑：心跳超时处理、客服端返回的数据处理

package com.dxfx.netty.demo;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

/**
 * 处理服务端业务逻辑：心跳超时处理、客服端返回的数据处理
 *
 * @author Administrator
 *
 */
public class HeartNettyServerHandler extends ChannelInboundHandlerAdapter {
 /** 空闲次数 */
    private int idle_count = 1;
 /** 发送次数 */
    private int count = 1;

 /**
 * 超时处理，如果5秒没有收到客户端的心跳，就触发; 如果超过两次，则直接关闭;
 */
 @Override
 public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
 if (obj instanceof IdleStateEvent) {
 IdleStateEvent event = (IdleStateEvent) obj;
 if (IdleState.READER_IDLE.equals(event.state())) { // 如果读通道处于空闲状态，说明没有接收到心跳命令
                if (idle_count > 2) {
 System.out.println("超过两次无客户端请求，关闭该channel");
 ctx.channel().close();
 }
 
 System.out.println("已等待5秒还没收到客户端发来的消息");
 idle_count++;
 }
 } else {
 super.userEventTriggered(ctx, obj);
 }
 }

 /**
 * 业务逻辑处理
 */
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 System.out.println("第" + count + "次" + "，服务端收到的消息:" + msg);
 
 String message = (String) msg;
 // 如果是心跳命令，服务端收到命令后回复一个相同的命令给客户端
        if ("hb_request".equals(message)) {
 ctx.write("服务端成功收到心跳信息");
 ctx.flush();
 }
 
 count++;
 }

 /**
 * 异常处理
 */
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 cause.printStackTrace();
 ctx.close();
 }
}

b、客户端
HeartNettyClient——客户端启动类

package com.dxfx.netty.demo;

import java.io.IOException;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * 客户端启动类
 *
 * @author Administrator
 *
 */
public class HeartNettyClient {

 public static void main(String[] args) throws InterruptedException, IOException {
 // 首先，netty通过Bootstrap启动客户端
        Bootstrap client = new Bootstrap();

 // 第1步 定义线程组，处理读写和链接事件，没有了accept事件
        EventLoopGroup group = new NioEventLoopGroup();
 client.group(group);

 // 第2步 绑定客户端通道
        client.channel(NioSocketChannel.class);

 // 第3步 给NIoSocketChannel初始化handler， 处理读写事件
        client.handler(new HeartNettyClientFilter());

 // 连接服务端
        Channel future = client.connect("localhost", 8080).sync().channel();
 
 //给服务端发送数据
        String str = "Hello Netty";
 future.writeAndFlush(str);
 System.out.println("客户端发送数据:" + str);
 }

}

HeartNettyClientFilter——客户端过滤器，如编解码和心跳的设置

package com.dxfx.netty.demo;

import java.util.concurrent.TimeUnit;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * 客户端过滤器，如编解码和心跳的设置
 *
 * @author Administrator
 *
 */
public class HeartNettyClientFilter  extends ChannelInitializer<SocketChannel> {

 @Override
 protected void initChannel(SocketChannel ch) throws Exception {
 ChannelPipeline ph = ch.pipeline();
 //因为服务端设置的超时时间是5秒，所以客户端设置4秒
        ph.addLast(new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS));
 
 // 解码和编码，应和服务端一致
        ph.addLast(new StringDecoder());
 ph.addLast(new StringEncoder());
 
 //处理客户端的业务逻辑
        ph.addLast(new HeartNettyClientHandler());
 }
}

 

HeartNettyClientHandler——处理客户端业务逻辑：心跳超时处理、服务端返回的数据处理

package com.dxfx.netty.demo;

import java.text.SimpleDateFormat;
import java.util.Date;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;

/**
 * 处理客户端业务逻辑：心跳超时处理、服务端返回的数据处理
 *
 * @author Administrator
 *
 */
public class HeartNettyClientHandler extends ChannelInboundHandlerAdapter {
 /** 客户端请求的心跳命令 */
    private static final ByteBuf HEARTBEAT_SEQUENCE = 
 Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("hb_request", CharsetUtil.UTF_8));

 /** 空闲次数 */
    private int idle_count = 1;

 /** 发送次数 */
    private int count = 1;

 /** 循环次数 */
    private int fcount = 1;

 /**
 * 建立连接时
 */
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 System.out.println("建立连接时：" + date());
 ctx.fireChannelActive();
 }

 /**
 * 关闭连接时
 */
 @Override
 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 System.out.println("关闭连接时：" + date());
 }

 /**
 * 心跳请求处理，每4秒发送一次心跳请求;
 *
 */
 @Override
 public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
 System.out.println("\r\n循环请求的时间：" + date() + "，次数" + fcount);
 
 if (obj instanceof IdleStateEvent) {
 IdleStateEvent event = (IdleStateEvent) obj;
 if (IdleState.WRITER_IDLE.equals(event.state())) { // 如果写通道处于空闲状态就发送心跳命令
 // 设置发送次数，允许发送3次心跳包
                if (idle_count <= 3) {
 idle_count++;
 ctx.channel().writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
 } else {
 System.out.println("心跳包发送结束，不再发送心跳请求！！！");
 }
 }
 }
 
 fcount++;
 }

 /**
 * 业务逻辑处理
 */
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 System.out.println("第" + count + "次" + "，客户端收到的消息:" + msg);
 count++;
 }
 
 private String date(){
 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 return sdf.format(new Date());
 }
}

 

c、客户端输出信息

客户端发送数据:Hello Netty
建立连接时：2018-12-14 22:18:05

循环请求的时间：2018-12-14 22:18:09，次数1
第1次，客户端收到的消息:服务端成功收到心跳信息

循环请求的时间：2018-12-14 22:18:13，次数2
第2次，客户端收到的消息:服务端成功收到心跳信息

循环请求的时间：2018-12-14 22:18:17，次数3
第3次，客户端收到的消息:服务端成功收到心跳信息

循环请求的时间：2018-12-14 22:18:21，次数4
心跳包发送结束，不再发送心跳请求！！！

循环请求的时间：2018-12-14 22:18:25，次数5
心跳包发送结束，不再发送心跳请求！！！

循环请求的时间：2018-12-14 22:18:29，次数6
心跳包发送结束，不再发送心跳请求！！！
关闭连接时：2018-12-14 22:18:32 

d、服务端输出信息

第1次，服务端收到的消息:Hello Netty
第2次，服务端收到的消息:hb_request
第3次，服务端收到的消息:hb_request
第4次，服务端收到的消息:hb_request
已等待5秒还没收到客户端发来的消息
已等待5秒还没收到客户端发来的消息
超过两次无客户端请求，关闭该channel
已等待5秒还没收到客户端发来的消息