Java IO编程由浅入深 - 10 (bio 基于websocket的心跳检测实现 )

Myth丶恋晨 2022-10-22 15:53 165阅读 0赞

### 实现基于websocket的心跳检测 ###

*  目录
 *  在之前的handler上面，添加userTriggerEvent方法
 *  websocket心跳检测handler实现
 *  WebSocketEvent对象
 *  Server端实现
 *  使用客户端连接服务端，看是否收到心跳包

# 目录 #

1.  [Java IO编程由浅入深 - 1 (bio c/s架构实现)][Java IO_ - 1 _bio c_s]
2.  [Java IO编程由浅入深 - 2(bio 基于字符串的消息编解码器)][Java IO_ - 2_bio]
3.  [Java IO编程由浅入深 - 3 (bio 基于消息长度的解编码器)][Java IO_ - 3 _bio]
4.  [Java IO编程由浅入深 - 4 (bio http协议解编码器，实现http服务器)][Java IO_ - 4 _bio http_http]
5.  [Java IO编程由浅入深 - 5 (项目架构重构)][Java IO_ - 5]
6.  [Java IO编程由浅入深 - 6 (bio实现C/S聊天室 1 )][Java IO_ - 6 _bio_C_S_ 1]
7.  [Java IO编程由浅入深 - 7 (bio实现C/S聊天室 2 )][Java IO_ - 7 _bio_C_S_ 2]
8.  [Java IO编程由浅入深 - 8 (bio实现C/S聊天室 3 )][Java IO_ - 8 _bio_C_S_ 3]
9.  [Java IO编程由浅入深 - 9 (bio实现http升级到websocket )][Java IO_ - 9 _bio_http_websocket]
10. [Java IO编程由浅入深 - 10 (bio 基于websocket的心跳检测实现 )][Java IO_ - 10 _bio _websocket_]

# 在之前的handler上面，添加userTriggerEvent方法 #

此方法可以让使用者像使用netty那样触发事件回调  
因为websocket握手成功之后，需要发送一个事件到管道里面，然后handler从上到下的方式触发userTriggerEvent方法，通过订阅此方法，实现心跳逻辑

# websocket心跳检测handler实现 #

package com.lhstack.bio.handler.websocket;
    
    import com.lhstack.bio.channel.ChannelHandlerContext;
    import com.lhstack.bio.channel.SimpleChannelAdapterHandler;
    import com.lhstack.bio.codec.websocket.frame.PingWebSocketFrame;
    import com.lhstack.bio.codec.websocket.frame.PongWebSocketFrame;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.time.LocalDateTime;
    import java.time.format.DateTimeFormatter;
    import java.util.concurrent.TimeUnit;
    
    /** * websocket心跳检测程序 * @author lhstack */
    public class WebsocketMessageIdleStateHandler extends SimpleChannelAdapterHandler<PongWebSocketFrame> { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(WebsocketMessageIdleStateHandler.class);
    
        private final int maxFailCount;
    
        private final int maxWaitTime;
    
        private int failCount;
    
        private long preRequestTime;
    
        public WebsocketMessageIdleStateHandler(){ 
            this(3,30);
        }
    
        public WebsocketMessageIdleStateHandler(int maxFailCount,int maxWaitTime){ 
            this.maxFailCount = maxFailCount;
            this.maxWaitTime = maxWaitTime;
        }
    
        @Override
        public void userTriggerEvent(ChannelHandlerContext ctx, Object evt) throws Exception { 
            if(evt == WebSocketEvent.ON_OPEN){ 
                ctx.channel().inEventLoop()
                        .schedule(new Runnable() { 
                            @Override
                            public void run() { 
                                try{ 
                                    if(failCount >= maxFailCount){ 
                                        ctx.close();
                                        return ;
                                    }else if(System.currentTimeMillis() / 1000 - preRequestTime >= maxWaitTime){ 
                                        this.failCount ++;
                                        return ;
                                    }
                                    LOGGER.debug("websocket server ping client {},date {}",ctx.channel().id(), LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                                    ctx.writeAndFlush(new PingWebSocketFrame());
                                    ctx.channel().inEventLoop().schedule(this,25,TimeUnit.SECONDS);
                                }catch (Exception e){ 
    
                                }
                            }
                        },25, TimeUnit.SECONDS);
                LOGGER.debug("websocket server ping client {},date {}",ctx.channel().id(), LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                ctx.writeAndFlush(new PingWebSocketFrame());
            }
        }
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx, PongWebSocketFrame msg) throws Exception { 
            LOGGER.debug("client {} pong server,date {}",ctx.channel().id(), LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            this.preRequestTime = System.currentTimeMillis() / 1000;
            this.failCount = 0;
        }
    }

# WebSocketEvent对象 #

package com.lhstack.bio.handler.websocket;
    
    /** * @author lhstack * websocket 事件 */
    public enum WebSocketEvent { 
       /** * websocket事件 */
       ON_OPEN,ON_MESSAGE,ON_CLOSE
    }

# Server端实现 #

public static void websocketServer() throws Exception { 
            ServerBootStrap serverBootStrap = new ServerBootStrap(8080);
            serverBootStrap.group(Executors.newFixedThreadPool(200),new DefaultEventLoopGroup())
                    .handler(new ChannelInitializeHandler() { 
                        @Override
                        public void initializer(Channel ch) throws Exception { 
                            ch.pipeline()
                                    .addLast(new HttpServerMessageCodec())
                                    .addLast(new WebsocketMessageHandler("/test", new SimpleChannelAdapterHandler<TextWebSocketFrame>() { 
                                        @Override
                                        public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { 
                                            System.out.println(msg.text());
                                            ctx.writeAndFlush(new TextWebSocketFrame("this is test message"));
                                        }
    
                                        @Override
                                        public void channelRemove(ChannelHandlerContext ctx) throws Exception { 
                                            System.out.println(ctx);
                                        }
    
                                        @Override
                                        public void exceptionCatch(ChannelHandlerContext ctx, Throwable throwable) throws Exception { 
                                            System.out.println(ctx);
                                            throwable.printStackTrace();
                                        }
                                    }))
                                    .addLast(new WebsocketMessageHandler("/hello", new SimpleChannelAdapterHandler<TextWebSocketFrame>() { 
                                        @Override
                                        public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { 
                                            System.out.println(msg.text());
                                            ctx.writeAndFlush(new TextWebSocketFrame("this is hello world message"));
                                        }
    
                                        @Override
                                        public void channelRemove(ChannelHandlerContext ctx) throws Exception { 
                                            System.out.println(ctx);
                                        }
    
                                        @Override
                                        public void exceptionCatch(ChannelHandlerContext ctx, Throwable throwable) throws Exception { 
                                            System.out.println(ctx);
                                            throwable.printStackTrace();
                                        }
                                    }))
                                    .addLast(new WebsocketMessageIdleStateHandler())
                                    .addLast(new SimpleChannelAdapterHandler<HttpRequest>() { 
                                        @Override
                                        public void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception { 
                                            System.out.println(msg);
                                            ctx.writeAndFlush(new HttpResponse("HTTP/1.1",200,Unpooled.wrappedBuffer("hello world".getBytes(StandardCharsets.UTF_8))));
                                        }
                                    });
                        }
                    }).start();

心跳检测的handler放到websocket握手handler之后即可

# 使用客户端连接服务端，看是否收到心跳包 #

第一次握手成功之后，会触发一次心跳检测  
![在这里插入图片描述][20210317201705741.png]

![在这里插入图片描述][20210317201659339.png]  
然后每隔25s触发一次心跳检测  
每次发送心跳包时，会判断上一次请求的时间，如果大于maxWaitTime，则失败次数加一  
当心跳失败次数大于等于maxFailCount，则会关闭客户端  
![在这里插入图片描述][20210317201736649.png]  
下一期实现多客户端群聊功能

[Java IO_ - 1 _bio c_s]: https://blog.csdn.net/qq_42413011/article/details/114553235
[Java IO_ - 2_bio]: https://blog.csdn.net/qq_42413011/article/details/114579502
[Java IO_ - 3 _bio]: https://blog.csdn.net/qq_42413011/article/details/114636827
[Java IO_ - 4 _bio http_http]: https://blog.csdn.net/qq_42413011/article/details/114646159
[Java IO_ - 5]: https://blog.csdn.net/qq_42413011/article/details/114802496
[Java IO_ - 6 _bio_C_S_ 1]: https://blog.csdn.net/qq_42413011/article/details/114852932
[Java IO_ - 7 _bio_C_S_ 2]: https://blog.csdn.net/qq_42413011/article/details/114878895
[Java IO_ - 8 _bio_C_S_ 3]: https://blog.csdn.net/qq_42413011/article/details/114896029
[Java IO_ - 9 _bio_http_websocket]: https://blog.csdn.net/qq_42413011/article/details/114925184
[Java IO_ - 10 _bio _websocket_]: https://blog.csdn.net/qq_42413011/article/details/114947311
[20210317201705741.png]: /images/20221022/f7f181ad62af447a8d220ebf33e2418b.png
[20210317201659339.png]: /images/20221022/a7fae1d2ecc847eeb2654aa90148233b.png
[20210317201736649.png]: /images/20221022/7535fcd71c2a47cd953981844f1a1cb3.png