当前位置:   article > 正文

基于netty的网络聊天室(二)——心跳检测及断线重连_netty websocket 断线通知

netty websocket 断线通知

前面介绍了Netty服务端客户端基本通信框架的搭建过程。下面将介绍Netty如何进行心跳检测以及处理客户端的断线重连。

为了适应恶劣的网络环境,比如网络超时、闪断,客户端进程僵死,需要机制来保证双方的通信能正常工作或者自动恢复。对于服务端来说,当客户端由于某些原因导致无法与服务端通信的,服务端需要主动注销与客户端的连接,减少无效链接的资源消耗。对于客户端来说,当服务进程宕机后进行重启,客户端应该自动能发起重连操作。

(一)心跳监测客户端

采用心跳机制,来确保服务端能及时发现无效的客户端链接。这里有个问题,心跳机制的发起方应该由服务端还是客户端。假设服务端出现宕机,客户端唯一能做的就是保证能及时发现服务端重启后能进行重连。因此,可以只由服务端来发起心跳检测。一旦服务端发现客户端连接超时多次,则 立即关闭链路。

心跳检测具体的设计思路如下:

1.服务端定时查看客户端链路是否空闲,一旦持续时间T没有收到客户端的请求包,则主动发送Ping包给客户端,同时心跳超时次数加1。

2.客户端收到服务的Ping请求,则立即发送一个Pong应答包。

3.服务端每次收到客户端的数据包,则重置超时次数。若连续N次未收到心跳应答包,则关闭链接。

心跳检测示例代码如下:

1.服务端NettyChatServer类的ChannelPipeline增加空闲状态处理器(IdleStateHandler)。该类用于检测通信Channel的读写状态超时,以此来实现心跳检测。IdleStateHandler的构造函数有三个参数,依次为读超时秒数,写超时秒数,读写超时秒数。我们只需要用到第一个参数。


2.ChatServerHandler类必须覆写userEventTriggered()方法处理超时逻辑。当超时次数少于指定次数时,向客户端发送Ping包;当超时次数大于指定次数时,注销客户端链接。


[java] view plain copy



  1. package com.kingston.netty;  
  2.   
  3. import io.netty.channel.Channel;  
  4. import io.netty.channel.ChannelHandlerAdapter;  
  5. import io.netty.channel.ChannelHandlerContext;  
  6. import io.netty.channel.ChannelPromise;  
  7. import io.netty.handler.timeout.IdleState;  
  8. import io.netty.handler.timeout.IdleStateEvent;  
  9.   
  10. import java.io.IOException;  
  11. import java.util.Map;  
  12. import java.util.concurrent.ConcurrentHashMap;  
  13.   
  14. import com.kingston.base.ServerManager;  
  15. import com.kingston.net.Packet;  
  16. import com.kingston.net.PacketManager;  
  17. import com.kingston.net.PacketType;  
  18. import com.kingston.service.login.ClientHeartBeat;  
  19. import com.kingston.service.login.LoginManagerProxy;  
  20. import com.kingston.service.login.ServerLogin;  
  21.   
  22. public class ChatServerHandler extends ChannelHandlerAdapter{  
  23.       
  24.     //客户端超时次数  
  25.     private Map<ChannelHandlerContext,Integer> clientOvertimeMap = new ConcurrentHashMap<>();  
  26.     private final int MAX_OVERTIME  = 3;  //超时次数超过该值则注销连接  
  27.       
  28.     @Override  
  29.     public void channelRead(ChannelHandlerContext context,Object msg)  
  30.             throws Exception{  
  31.         Packet  packet = (Packet)msg;  
  32.         if(packet.getPacketType() == PacketType.ServerLogin ){  
  33.             ServerLogin loginPact = (ServerLogin)packet;  
  34.             LoginManagerProxy.getManager().validateLogin(context,loginPact.getUserId(), loginPact.getUserPwd());  
  35.             return ;  
  36.         }else{  
  37.             if(validateSession(packet)){  
  38.                 PacketManager.execPacket(packet);  
  39.             }  
  40.         }  
  41.           
  42.         clientOvertimeMap.remove(context);//只要接受到数据包,则清空超时次数  
  43.           
  44.     }  
  45.       
  46.     private  boolean validateSession(Packet loginPact){  
  47.         return true;  
  48.     }  
  49.   
  50.     @Override  
  51.     public void close(ChannelHandlerContext ctx,ChannelPromise promise){  
  52.         System.err.println("TCP closed...");  
  53.         ctx.close(promise);  
  54.     }  
  55.   
  56.     @Override  
  57.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {  
  58.         System.err.println("客户端关闭1");  
  59.     }  
  60.       
  61.         @Override  
  62.         public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {  
  63.             ctx.disconnect(promise);  
  64.             System.err.println("客户端关闭2");  
  65.         }  
  66.           
  67.         @Override  
  68.         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {  
  69.             System.err.println("业务逻辑出错");  
  70.             cause.printStackTrace();  
  71.             //          ctx.fireExceptionCaught(cause);  
  72.             Channel channel = ctx.channel();  
  73.             if(cause instanceof  IOException && channel.isActive()){  
  74.                 System.err.println("simpleclient"+channel.remoteAddress()+"异常");  
  75.                 ctx.close();  
  76.             }  
  77.         }  
  78.           
  79.         @Override  
  80.         public void userEventTriggered(ChannelHandlerContext ctx, Object evt)  
  81.                 throws Exception {  
  82.             //心跳包检测读超时  
  83.             if (evt instanceof IdleStateEvent) {  
  84.                 IdleStateEvent e = (IdleStateEvent) evt;  
  85.                 if (e.state() == IdleState.READER_IDLE) {  
  86.                     System.err.println("客户端读超时");  
  87.                     int overtimeTimes = clientOvertimeMap.getOrDefault(ctx, 0);  
  88.                     if(overtimeTimes < MAX_OVERTIME){  
  89.                         ServerManager.sendPacketTo(new ClientHeartBeat(), ctx);  
  90.                         addUserOvertime(ctx);  
  91.                     }else{  
  92.                         ServerManager.ungisterUserContext(ctx);  
  93.                     }  
  94.                 }   
  95.             }  
  96.         }  
  97.           
  98.         private void addUserOvertime(ChannelHandlerContext ctx){  
  99.             int oldTimes = 0;  
  100.             if(clientOvertimeMap.containsKey(ctx)){  
  101.                 oldTimes = clientOvertimeMap.get(ctx);  
  102.             }  
  103.             clientOvertimeMap.put(ctx, (int)(oldTimes+1));  
  104.         }  
  105. }  

3.增加下发包ClientHeartBeat类定义。客户端在收到该包的时候,需要向服务端发送一个应答包。



[java] view plain copy



  1. package com.kingston.service.login;  
  2.   
  3. import io.netty.buffer.ByteBuf;  
  4.   
  5. import com.kingston.base.ServerManager;  
  6. import com.kingston.net.Packet;  
  7. import com.kingston.net.PacketType;  
  8.   
  9. public class ClientHeartBeat extends Packet{  
  10.   
  11.     @Override  
  12.     public void writePacketMsg(ByteBuf buf) {  
  13.         // TODO Auto-generated method stub  
  14.           
  15.     }  
  16.   
  17.     @Override  
  18.     public void readFromBuff(ByteBuf buf) {  
  19.         // TODO Auto-generated method stub  
  20.           
  21.     }  
  22.   
  23.     @Override  
  24.     public PacketType getPacketType() {  
  25.         return PacketType.ClientHeartBeat;  
  26.     }  
  27.   
  28.     @Override  
  29.     public void execPacket() {  
  30.         System.err.println("收到服务端的ping请求后,回复一个pong响应");  
  31.         ServerManager.sendServerRequest(new ServerHeartBeat());  
  32.           
  33.     }  
  34.   
  35. }  

4.服务端在收到应答包后,重置超时次数为0


心跳调试技巧:如果需要演示心跳超时,只需在客户端启动后在任意代码里加个断点,这样服务端就会检测到客户端读超时。
(二)客户端断线重连
当服务端宕机后,客户端需要定时检测服务端开启状态,重新连接。实现逻辑也比较简单,只要在NettyChatClient类断开链接的逻辑后加上重连逻辑即可(reConnectServer()方法)。每次重连检测不必过于频繁,可以让线程休眠一段时间。


[java] view plain copy



  1. package com.kingston.netty;  
  2.   
  3. import java.net.InetSocketAddress;  
  4.   
  5. import io.netty.bootstrap.Bootstrap;  
  6. import io.netty.channel.ChannelFuture;  
  7. import io.netty.channel.ChannelInitializer;  
  8. import io.netty.channel.ChannelPipeline;  
  9. import io.netty.channel.EventLoopGroup;  
  10. import io.netty.channel.nio.NioEventLoopGroup;  
  11. import io.netty.channel.socket.SocketChannel;  
  12. import io.netty.channel.socket.nio.NioSocketChannel;  
  13. import io.netty.handler.codec.LengthFieldPrepender;  
  14.   
  15. import com.kingston.net.codec.PacketDecoder;  
  16. import com.kingston.net.codec.PacketEncoder;  
  17.   
  18.   
  19. public class NettyChatClient {  
  20.   
  21.     public void connect(String host,int port) throws Exception{  
  22.         EventLoopGroup group = new NioEventLoopGroup();  
  23.         try{  
  24.             Bootstrap b  = new Bootstrap();  
  25.             b.group(group).channel(NioSocketChannel.class)  
  26.             .handler(new ChannelInitializer<SocketChannel>(){  
  27.   
  28.                 @Override  
  29.                 protected void initChannel(SocketChannel arg0)  
  30.                         throws Exception {  
  31.                     ChannelPipeline pipeline = arg0.pipeline();  
  32.                     pipeline.addLast(new PacketDecoder(1024*10,2,0,2));  
  33.                     pipeline.addLast(new LengthFieldPrepender(2));  
  34.                     pipeline.addLast(new PacketEncoder());  
  35. //                  pipeline.addLast(new HeartBeatReqHandler());   
  36.                     pipeline.addLast(new NettyClientHandler());  
  37.                 }  
  38.                   
  39.             });  
  40.               
  41.             ChannelFuture f = b.connect(new InetSocketAddress(host, port),  
  42.                     new InetSocketAddress(NettyContants.LOCAL_SERVER_IP, NettyContants.LOCAL_SERVER_PORT))  
  43.                     .sync();  
  44.             f.channel().closeFuture().sync();  
  45.         }catch(Exception e){  
  46.             e.printStackTrace();  
  47.         }finally{  
  48. //          group.shutdownGracefully();  //这里不再是优雅关闭了  
  49.             reConnectServer();  
  50.         }  
  51.     }  
  52.       
  53.     /** 
  54.      * 断线重连 
  55.      */  
  56.     private void reConnectServer(){  
  57.           
  58.         try {  
  59.             Thread.sleep(5000);  
  60.             System.err.println("客户端进行断线重连");  
  61.             connect(NettyContants.REMOTE_SERVER_IP,  
  62.                     NettyContants.REMOTE_SERVER_PORT);  
  63.         } catch (Exception e) {  
  64.             e.printStackTrace();  
  65.         }  
  66.     }  
  67.       
  68.       
  69. }  




调试技巧:启动服务端与客户端后,单方面关闭服务端,即可看见客户端定时重连了。需要保证客户端重连成功后,能够与服务端收发数据,同时客户端也不无须继续检测重连。



全部代码已在github上托管(代码经过多次重构,与博客上的代码略有不同)


(服务端git地址,go->)

(客户端git地址,go->)

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/242046
推荐阅读
相关标签
  

闽ICP备14008679号