【Netty】利用Netty实现心跳检测和重连机制

时间 2019-12-11 标签 Netty 利用 netty 实现心跳检测机制

1、前言

心跳机制是定时发送一个自定义的结构体(心跳包)，让对方知道本身还活着，以确保链接的有效性的机制。
咱们用到的不少框架都用到了心跳检测，好比服务注册到 Eureka Server 以后会维护一个心跳链接，告诉 Eureka Server 本身还活着。本文就是利用 Netty 来实现心跳检测，以及客户端重连。html

2、设计思路

分为客户端和服务端
创建链接后，客户端先发送一个消息询问服务端是否能够进行通讯了。
客户端收到服务端 Yes 的应答后，主动发送心跳消息，服务端接收到心跳消息后，返回心跳应答，周而复始。
心跳超时利用 Netty 的 ReadTimeOutHandler 机制，当必定周期内(默认值50s)没有读取到对方任何消息时，须要主动关闭链路。若是是客户端，从新发起链接。
为了不出现粘/拆包问题，使用 DelimiterBasedFrameDecoder 和 StringDecoder 来处理消息。

3、编码

先编写客户端 NettyClient

 
     public class NettyClient {  
       
 private static final String HOST = "127.0.0.1";  
       
 private static final int PORT = 9911;  
       
 private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);  
       
 EventLoopGroup group = new NioEventLoopGroup();  
       
 private void connect(String host,int port){  
       
 try {  
       
 Bootstrap b = new Bootstrap();  
       
 b.group(group)  
       
 .channel(NioSocketChannel.class)  
       
 .option(ChannelOption.TCP_NODELAY,true)  
       
 .remoteAddress(new InetSocketAddress(host,port))  
       
 .handler(new ChannelInitializer<SocketChannel>() {  
       
 @Override  
       
 protected void initChannel(SocketChannel ch) throws Exception {  
       
 ByteBuf delimiter = Unpooled.copiedBuffer("$_", CharsetUtil.UTF_8);  
       
 ch.pipeline()  
       
 .addLast(new DelimiterBasedFrameDecoder(1024,delimiter))  
       
 .addLast(new StringDecoder())  
       
 // 当必定周期内(默认50s)没有收到对方任何消息时，须要主动关闭连接  
       
 .addLast("readTimeOutHandler",new ReadTimeoutHandler(50))  
       
 .addLast("heartBeatHandler",new HeartBeatReqHandler());  
       
 }  
       
 });  
       
 // 发起异步链接操做  
       
 ChannelFuture future = b.connect().sync();  
       
 future.channel().closeFuture().sync();  
       
 }catch (Exception e){  
       
 e.printStackTrace();  
       
 }finally {  
       
 // 全部资源释放完以后，清空资源，再次发起重连操做  
       
 executor.execute(()->{  
       
 try {  
       
 TimeUnit.SECONDS.sleep(5);  
       
 //发起重连操做  
       
 connect(NettyClient.HOST,NettyClient.PORT);  
       
 } catch (InterruptedException e) {  
       
 e.printStackTrace();  
       
 }  
       
 });  
       
 }  
       
 }  
       
 public static void main(String[] args) {  
       
 new NettyClient().connect(NettyClient.HOST,NettyClient.PORT);  
       
 }  
       
}

这里稍微复杂点的就是38行开始的重连部分。
2. 心跳消息发送类 HeartBeatReqHandlerjava

 
     package cn.sp.heartbeat;  
       
import io.netty.buffer.Unpooled;  
       
import io.netty.channel.ChannelHandler;  
       
import io.netty.channel.ChannelHandlerContext;  
       
import io.netty.channel.SimpleChannelInboundHandler;  
       
import java.util.concurrent.ScheduledFuture;  
       
import java.util.concurrent.TimeUnit;  
       
/**  
       
 * Created by 2YSP on 2019/5/23.  
       
 */  
       
@ChannelHandler.Sharable  
       
public class HeartBeatReqHandler extends SimpleChannelInboundHandler<String> {  
       
 private volatile ScheduledFuture<?> heartBeat;  
       
 private static final String hello = "start notify with server$_";  
       
 @Override  
       
 public void channelActive(ChannelHandlerContext ctx) throws Exception {  
       
 ctx.writeAndFlush(Unpooled.copiedBuffer(hello.getBytes()));  
       
 System.out.println("================");  
       
 }  
       
 @Override  
       
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {  
       
 if (heartBeat != null){  
       
 heartBeat.cancel(true);  
       
 heartBeat = null;  
       
 }  
       
 ctx.fireExceptionCaught(cause);  
       
 }  
       
 @Override  
       
 protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {  
       
 if ("ok".equalsIgnoreCase(msg)){  
       
 //服务端返回ok开始心跳  
       
 heartBeat = ctx.executor().scheduleAtFixedRate(new HeartBeatTask(ctx),0,5000, TimeUnit.MILLISECONDS);  
       
 }else {  
       
 System.out.println("Client receive server heart beat message : --->"+msg);  
       
 }  
       
 }  
       
 private class HeartBeatTask implements Runnable{  
       
 private final ChannelHandlerContext ctx;  
       
 public HeartBeatTask(ChannelHandlerContext ctx){  
       
 this.ctx = ctx;  
       
 }  
       
 @Override  
       
 public void run() {  
       
 String heartBeat = "I am ok";  
       
 System.out.println("Client send heart beat message to server: ----->"+heartBeat);  
       
 ctx.writeAndFlush(Unpooled.copiedBuffer((heartBeat+"$_").getBytes()));  
       
 }  
       
 }  
       
}

channelActive()方法在首次创建链接后向服务端问好，若是服务端返回了 "ok" 就建立一个线程每隔5秒发送一次心跳消息。若是发生了异常，就取消定时任务并将其设置为 null，等待 GC 回收。
3. 服务端 NettyServergit

 
     public class NettyServer {  
       
 public static void main(String[] args) {  
       
 new NettyServer().bind(9911);  
       
 }  
       
 private void bind(int port){  
       
 EventLoopGroup group = new NioEventLoopGroup();  
       
 try {  
       
 ServerBootstrap b = new ServerBootstrap();  
       
 b.group(group)  
       
 .channel(NioServerSocketChannel.class)  
       
 .childHandler(new ChannelInitializer<SocketChannel>() {  
       
 @Override  
       
 protected void initChannel(SocketChannel ch) throws Exception {  
       
 ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());  
       
 ch.pipeline()  
       
 .addLast(new DelimiterBasedFrameDecoder(1024,delimiter))  
       
 .addLast(new StringDecoder())  
       
 .addLast("readTimeOutHandler",new ReadTimeoutHandler(50))  
       
 .addLast("HeartBeatHandler",new HeartBeatRespHandler());  
       
 }  
       
 });  
       
 // 绑定端口，同步等待成功  
       
 b.bind(port).sync();  
       
 System.out.println("Netty Server start ok ....");  
       
 }catch (Exception e){  
       
 e.printStackTrace();  
       
 }  
       
 }  
       
}

心跳响应类 HeartBeatRespHandler

 
     package cn.sp.heartbeat;  
       
import io.netty.buffer.Unpooled;  
       
import io.netty.channel.ChannelHandler;  
       
import io.netty.channel.ChannelHandlerContext;  
       
import io.netty.channel.SimpleChannelInboundHandler;  
       
/**  
       
 * Created by 2YSP on 2019/5/23.  
       
 */  
       
@ChannelHandler.Sharable  
       
public class HeartBeatRespHandler extends SimpleChannelInboundHandler<String> {  
       
 private static final String resp = "I have received successfully$_";  
       
 @Override  
       
 protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {  
       
 if (msg.equals("start notify with server")){  
       
 ctx.writeAndFlush(Unpooled.copiedBuffer("ok$_".getBytes()));  
       
 }else {  
       
 //返回心跳应答信息  
       
 System.out.println("Receive client heart beat message: ---->"+ msg);  
       
 ctx.writeAndFlush(Unpooled.copiedBuffer(resp.getBytes()));  
       
 }  
       
 }  
       
}

第一次告诉客户端我已经准备好了，后面打印客户端发过来的信息并告诉客户端我已经收到你的消息了。github

4、测试

启动服务端再启动客户端，能够看到心跳检测正常，以下图。框架

服务端控制台

客户端控制台

如今让服务端宕机一段时间，看客户端可否重连并开始正常工做。

关闭服务端后，客户端周期性的链接失败，控制台输出如图：异步

链接失败

从新启动服务端，过一下子发现重连成功了。

成功重连

5、总结

总得来讲，使用 Netty 实现心跳检测仍是比较简单的，这里比较懒没有使用其余序列化协议（如 ProtoBuf 等），若是感兴趣的话你们能够本身试试。
代码地址，点击这里。
有篇SpringBoot 整合长链接心跳机制的文章写的也很不错，地址https://crossoverjie.top/2018/05/24/netty/Netty(1)TCP-Heartbeat/ide