Netty之心跳检测

朴灿烈づ我的快乐病毒、 2023-06-19 15:27 32阅读 0赞

# 何为心跳 #

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

> 注：心跳包还有另一个作用，经常被忽略，即：**一个连接如果长时间不用，防火墙或者路由器就会断开该连接**。

## 如何实现 ##

### 核心Handler —— IdleStateHandler ###

在 `Netty` 中, 实现心跳机制的关键是 `IdleStateHandler`, 那么这个 `Handler` 如何使用呢? 先看下它的构造器：

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

这里解释下三个参数的含义：

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

> 注：这三个参数默认的时间单位是**秒**。若需要指定其他时间单位，可以使用另一个构造方法：`IdleStateHandler(boolean observeOutput, long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit)`

### 使用IdleStateHandler实现心跳 ###

下面将使用`IdleStateHandler`来实现心跳，`Client`端连接到`Server`端后，会循环执行一个任务：**随机等待几秒，然后`ping`一下`Server`端，即发送一个心跳包**。当等待的时间超过规定时间，将会发送失败，以为`Server`端在此之前已经主动断开连接了。

服务端

package com.yj.server.service;
    
    import java.util.concurrent.TimeUnit;
    import javax.annotation.PostConstruct;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.handler.timeout.IdleStateHandler;
    
    @Component
    public class HeartBeatServer {
    
    	@Value("${netty.port}")
    	private int port;
    
    	@PostConstruct
    	public void start() throws Exception {
    		EventLoopGroup group = new NioEventLoopGroup();
    		try {
    			ServerBootstrap b = new ServerBootstrap();
    			b.group(group).channel(NioServerSocketChannel.class).localAddress(port)
    					.childHandler(new ChannelInitializer<Channel>() {
    						@Override
    						protected void initChannel(Channel ch) throws Exception {
    							//表示当前channel多长时间没有读写操作时,触发事件,发出心跳包对该channel进行检查
    							//事件触发后,会传递给管道的下一个handleer的userEventTriggered方法进行处理
    							ch.pipeline().addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
                                ch.pipeline().addLast("decoder", new StringDecoder());
                                ch.pipeline().addLast(new AcceptorIdleStateTrigger());
                                ch.pipeline().addLast("encoder", new StringEncoder());
    						}
    					}).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);
    			ChannelFuture f = b.bind().sync();
    			System.out.println("Netty服务端启动成功,开始监听端口：" + port);
    			f.channel().closeFuture().sync();
    		} finally {
    			group.shutdownGracefully().sync();
    		}
    	}
    }

AcceptorIdleStateTrigger

package com.yj.server.service;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    
    public class AcceptorIdleStateTrigger extends ChannelInboundHandlerAdapter {
    
    	@Override
    	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    		if (evt instanceof IdleStateEvent) {
    			IdleState state = ((IdleStateEvent) evt).state();
    			if (state == IdleState.READER_IDLE) {
    				throw new Exception("状态异常");
    			}
    		} else {
    			super.userEventTriggered(ctx, evt);
    		}
    	}
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    		System.out.println("接收到客户端的心跳包["+getDate()+"]");
    	}
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		cause.printStackTrace();
    		ctx.close();
    	}
    	
    	private String getDate() {
    		Date sysDate = new Date();
    		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    		String strDate = format.format(sysDate);
    		return strDate;
    	}
    }

客户端

package com.yj.client.service;
    
    import java.util.concurrent.TimeUnit;
    import javax.annotation.PostConstruct;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    import io.netty.handler.timeout.IdleStateHandler;
    import io.netty.util.HashedWheelTimer;
    
    @Component
    public class HeartBeatClient {
    
    	@Value("${netty.host}")
    	private String host;
    
    	@Value("${netty.port}")
    	private int port;
    	
    	@Value("${netty.reconnect}")
    	private Boolean reconnect;
    	
    	@Value("${netty.attemptsNum}")
    	private int attemptsNum;
    
    	private final HashedWheelTimer timer = new HashedWheelTimer();
    
    	@PostConstruct
    	public void connect() throws Exception {
    		EventLoopGroup group = new NioEventLoopGroup();
    		Bootstrap boot = new Bootstrap();
    		boot.group(group).channel(NioSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO));
    		final ConnectorIdleStateTrigger connectorIdleStateTrigger = new ConnectorIdleStateTrigger(boot, timer, port, host, reconnect,attemptsNum) {
    			public ChannelHandler[] handlers() {
    				return new ChannelHandler[] {new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS),
    						new StringDecoder(),this,new StringEncoder() };
    			}
    		};
    		ChannelFuture future;
    		try {
    			synchronized (boot) {
    				boot.handler(new ChannelInitializer<Channel>() {
    					// 初始化channel
    					@Override
    					protected void initChannel(Channel ch) throws Exception {
    						ch.pipeline().addLast(connectorIdleStateTrigger.handlers());
    					}
    				});
    				future = boot.connect(host, port);
    			}
    			future.sync();
    		} catch (Throwable t) {
    			t.printStackTrace();
    			throw new Exception("连接失败", t);
    		}
    	}
    }

ConnectorIdleStateTrigger

package com.yj.client.service;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.concurrent.TimeUnit;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandler.Sharable;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.channel.ChannelInitializer;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    import io.netty.util.CharsetUtil;
    import io.netty.util.Timeout;
    import io.netty.util.Timer;
    import io.netty.util.TimerTask;
    
    /**
     * 重连检测狗，当发现当前的链路不稳定关闭之后，进行重连
     */
    @Sharable
    public abstract class ConnectorIdleStateTrigger extends ChannelInboundHandlerAdapter
    		implements TimerTask, ChannelHandlerHolder {
    
    	private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled
    			.unreleasableBuffer(Unpooled.copiedBuffer("Heartbeat", CharsetUtil.UTF_8));
    
    	private int attemptsNum;
    
    	private final Bootstrap bootstrap;
    
    	private final Timer timer;
    
    	private final int port;
    
    	private final String host;
    
    	private volatile boolean reconnect = true;
    
    	private int attempts;
    
    	public ConnectorIdleStateTrigger(Bootstrap bootstrap, Timer timer, int port, String host, boolean reconnect,
    			int attemptsNum) {
    		this.bootstrap = bootstrap;
    		this.timer = timer;
    		this.port = port;
    		this.host = host;
    		this.reconnect = reconnect;
    		this.attemptsNum = attemptsNum;
    	}
    
    	/**
    	 * channel链路每次active的时候，将其连接的次数重置为 0
    	 */
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) throws Exception {
    		System.out.println("当前链路已经激活了,将重连尝试次数重置为0");
    		attempts = 0;
    		ctx.fireChannelActive();
    	}
    
    	@Override
    	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    		if (reconnect) {
    			if (attempts < attemptsNum) {
    				attempts++;
    				System.out.println("进行第" + attempts + "次重连["+getDate()+"]");
    				// 重连的间隔时间会越来越长
    				int timeout = 2 << attempts;
    				timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
    			}
    		}
    		ctx.fireChannelInactive();
    	}
    
    	@Override
    	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    		if (evt instanceof IdleStateEvent) {
    			IdleState state = ((IdleStateEvent) evt).state();
    			if (state == IdleState.WRITER_IDLE) {
    				ctx.writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
    			}
    		} else {
    			super.userEventTriggered(ctx, evt);
    		}
    	}
    
    	public void run(Timeout timeout) throws Exception {
    		ChannelFuture future;
    		// bootstrap已经初始化好了，只需要将handler填入就可以了
    		synchronized (bootstrap) {
    			bootstrap.handler(new ChannelInitializer<Channel>() {
    				@Override
    				protected void initChannel(Channel ch) throws Exception {
    					ch.pipeline().addLast(handlers());
    				}
    			});
    			future = bootstrap.connect(host, port);
    		}
    		// future对象
    		future.addListener(new ChannelFutureListener() {
    
    			public void operationComplete(ChannelFuture f) throws Exception {
    				boolean succeed = f.isSuccess();
    
    				// 如果重连失败，则调用ChannelInactive方法，再次出发重连事件，一直尝试12次，如果失败则不再重连
    				if (!succeed) {
    					System.out.println("重连失败");
    					f.channel().pipeline().fireChannelInactive();
    				} else {
    					System.out.println("重连成功");
    				}
    			}
    		});
    	}
    
    	private String getDate() {
    		Date sysDate = new Date();
    		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    		String strDate = format.format(sysDate);
    		return strDate;
    	}
    }

启动服务端，客户端，当客户端没有 写操作 时，每隔4秒会发送一个心跳包到服务端。

服务端输出如下：

Netty服务端启动成功,开始监听端口：8899
    接收到客户端的心跳包[2019-12-07 12:53:45]
    接收到客户端的心跳包[2019-12-07 12:53:49]
    接收到客户端的心跳包[2019-12-07 12:53:53]
    接收到客户端的心跳包[2019-12-07 12:53:57]
    接收到客户端的心跳包[2019-12-07 12:54:01]
    接收到客户端的心跳包[2019-12-07 12:54:05]
    接收到客户端的心跳包[2019-12-07 12:54:09]
    接收到客户端的心跳包[2019-12-07 12:54:13]
    接收到客户端的心跳包[2019-12-07 12:54:17]
    接收到客户端的心跳包[2019-12-07 12:54:21]
    接收到客户端的心跳包[2019-12-07 12:54:25]
    接收到客户端的心跳包[2019-12-07 12:54:29]
    接收到客户端的心跳包[2019-12-07 12:54:33]
    接收到客户端的心跳包[2019-12-07 12:54:37]
    接收到客户端的心跳包[2019-12-07 12:54:41]
    接收到客户端的心跳包[2019-12-07 12:54:45]
    接收到客户端的心跳包[2019-12-07 12:54:49]
    接收到客户端的心跳包[2019-12-07 12:54:53]

关闭服务端后，客户端会尝试重连12次，再次启动服务端，客户端重连成功

进行第1次重连[2019-12-07 12:59:27]
    重连失败
    进行第2次重连[2019-12-07 12:59:29]
    重连失败
    进行第3次重连[2019-12-07 12:59:31]
    重连失败
    进行第4次重连[2019-12-07 12:59:33]
    重连失败
    进行第5次重连[2019-12-07 12:59:35]
    重连失败
    进行第6次重连[2019-12-07 12:59:37]
    重连成功
    当前链路已经激活了,将重连尝试次数重置为0