Java IO编程由浅入深 - 7 (bio实现C/S聊天室 2 )

﹏ヽ暗。殇╰゛Y 2022-11-10 05:23 142阅读 0赞

### 实现客户端与服务端的心跳机制 ###

*  目录
 *  功能扩展
 *  实现服务端的心跳检测handler
 *  实现客户端心跳机制的handler
 *  测试服务端与客户端之间的心跳通讯
 *  客户端代码
 *  查看服务端控制台
 *  查看客户端控制台

# 目录 #

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_]

# 功能扩展 #

1.  在原有基础上，我定义了一个EventLoopGroup接口，同时实现了一个默认的EventLoopGroup，它的作用是为channel分配一个EventLoop，同时，多个channel可以共享一个EventLoop
2.  EventLoop，类似线程池的功能，内部维护一个线程，同时不断的从task队列里面拿取任务执行
3.  为channel绑定一个方法inEventLoop()，可以通过这个获取与channel绑定的EventLoop对象，通过这个对象可以放一些特殊任务。

# 实现服务端的心跳检测handler #

package com.lhstack.example.handler;
    
    import com.lhstack.bio.channel.ChannelHandlerContext;
    import com.lhstack.bio.channel.SimpleChannelAdapterHandler;
    import com.lhstack.bio.codec.bs.Message;
    import com.lhstack.bio.codec.bs.MessageType;
    
    import java.nio.charset.StandardCharsets;
    import java.util.concurrent.TimeUnit;
    
    /** * @author lhstack * 服务端心跳检测 */
    public class CsServerMessageIdleStateHandler extends SimpleChannelAdapterHandler<Message> { 
    
        /** * 失败次数 */
        private int failCount;
    
        /** * 上次请求时间 */
        private long prevRequestTime;
        
        private static final int MAX_FAIL_COUNT = 3;
        
        private static final int MAX_WAIT_TIME = 30;
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            //调用与channel绑定的eventLoop对象，注册schedule方法，实现心跳检测和发送
            ctx.channel().inEventLoop()
                    .schedule(new Runnable() { 
                        @Override
                        public void run() { 
                            //判断是否超过MAX_WAIT_TIME，如果超过，则判断失败次数，如果大于MAX_FAIL_COUNT，则客户端已经失效了，就关闭客户端连接
                            if(System.currentTimeMillis() / 1000 - prevRequestTime > MAX_WAIT_TIME){ 
                                if(failCount >= MAX_FAIL_COUNT){ 
                                    ctx.close();
                                }else{ 
                                    failCount ++;
                                }
                            }
                            try { 
                                //发送ping消息
                                ctx.writeAndFlush(new Message(MessageType.PING, 0,"ping".getBytes(StandardCharsets.UTF_8),false));
                                //注册下一次执行的任务
                                ctx.channel().inEventLoop().schedule(this,25,TimeUnit.SECONDS);
                            } catch (Exception e) { 
                                e.printStackTrace();
                            }
                        }
                    },25, TimeUnit.SECONDS);
            //连接成功，直接发送一次ping请求
            ctx.writeAndFlush(new Message(MessageType.PING, 0,"ping".getBytes(StandardCharsets.UTF_8),false));
            ctx.fireChannelActive();
        }
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception { 
            //获取到pong，则初始化上面的失败次数，并记录pong的请求时间
            if(msg.getMessageType() == MessageType.PONG){ 
                this.prevRequestTime = System.currentTimeMillis() / 1000;
                this.failCount = 0;
            }else{ 
                ctx.fireChannelRead(msg);
            }
        }
    }

# 实现客户端心跳机制的handler #

# 测试服务端与客户端之间的心跳通讯 #

服务端代码

public static void csMessageServer() 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 CsMessageCodec())
                                    .addLast(new CsServerMessageIdleStateHandler())
                                    .addLast(new SimpleChannelAdapterHandler<Message>() { 
    
                                        @Override
                                        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
                                            //不断的发送测试数据
                                            ctx.channel().inEventLoop()
                                                    .schedule(new Runnable() { 
                                                        @Override
                                                        public void run() { 
                                                            try{ 
                                                                Message message = new Message(MessageType.TEXT, 0, ("当前时间是: " + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))).getBytes(StandardCharsets.UTF_8), true);
                                                                ctx.writeAndFlush(message);
                                                                ctx.channel().inEventLoop().schedule(this,1,TimeUnit.SECONDS);
                                                            }catch (Exception e){ 
                                                                e.printStackTrace();
                                                            }
                                                        }
                                                    }, 1, TimeUnit.SECONDS);
    
                                        }
    
                                        @Override
                                        public void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception { 
                                            System.out.println(msg);
                                            System.out.println(new String(msg.getBody()));
                                        }
    
                                        @Override
                                        public void exceptionCatch(ChannelHandlerContext ctx, Throwable throwable) throws Exception { 
                                            throwable.printStackTrace();
                                        }
                                    });
                        }
                    });
            serverBootStrap.start();
        }

# 客户端代码 #

public static void csMessageClient() throws Exception { 
            BootStrap bootStrap = new BootStrap();
            Channel channel = bootStrap
                    .group(new DefaultEventLoopGroup())
                    .handler(new ChannelInitializeHandler() { 
                @Override
                public void initializer(Channel ch) throws Exception { 
                    ch.pipeline()
                            .addLast(new CsMessageCodec())
                            .addLast(new CsClientMessageIdleStateHandler())
                            .addLast(new SimpleChannelAdapterHandler<Message>() { 
                                @Override
                                public void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception { 
                                    System.out.println(new String(msg.getBody()));
                                }
    
                                @Override
                                public void exceptionCatch(ChannelHandlerContext ctx, Throwable throwable) throws Exception { 
                                    throwable.printStackTrace();
                                }
                            });
                }
            }).connect(new InetSocketAddress("localhost", 8080));
        }

# 查看服务端控制台 #

![在这里插入图片描述][20210316140830895.png]

# 查看客户端控制台 #

![在这里插入图片描述][20210316140903344.png]  
![在这里插入图片描述][20210316140912288.png]  
可以看到，客户端每隔25s收到了服务端的ping请求，实现了心跳功能，如要实现断线重连，则只需要在客户端异常catch回调里面去重新创建客户端去连接服务端即可，这里我就不实现了  
下一期，将开始正式进入客户端与服务端之间的消息聊天通讯

[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
[20210316140830895.png]: /images/20221023/5a0bfcc95a4c422d8e7cd40953699829.png
[20210316140903344.png]: /images/20221023/295a89c021b548bea516bd5ba8a17c88.png
[20210316140912288.png]: /images/20221023/74fd84a7514b4e5a9e81b4f6caff8643.png