由浅入深Netty聊天室案例

曾经终败给现在 2024-03-22 19:46 62阅读 0赞

#### 目录 ####

*  1 聊天室业务介绍
 *  2 聊天室业务-登录
 *  3 聊天室业务-单聊
 *  4 聊天室业务-群聊
 *  5 聊天室业务-退出
 *  6 聊天室业务-空闲检测
 *   *  6.1 连接假死

--------------------

## 1 聊天室业务介绍 ##

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

/**
     * 用户管理接口
     */
    public interface UserService {
        
    
        /**
         * 登录
         * @param username 用户名
         * @param password 密码
         * @return 登录成功返回 true, 否则返回 false
         */
        boolean login(String username, String password);
    }

/**
     * 会话管理接口
     */
    public interface Session {
        
    
        /**
         * 绑定会话
         * @param channel 哪个 channel 要绑定会话
         * @param username 会话绑定用户
         */
        void bind(Channel channel, String username);
    
        /**
         * 解绑会话
         * @param channel 哪个 channel 要解绑会话
         */
        void unbind(Channel channel);
    
        /**
         * 获取属性
         * @param channel 哪个 channel
         * @param name 属性名
         * @return 属性值
         */
        Object getAttribute(Channel channel, String name);
    
        /**
         * 设置属性
         * @param channel 哪个 channel
         * @param name 属性名
         * @param value 属性值
         */
        void setAttribute(Channel channel, String name, Object value);
    
        /**
         * 根据用户名获取 channel
         * @param username 用户名
         * @return channel
         */
        Channel getChannel(String username);
    }

/**
     * 聊天组会话管理接口
     */
    public interface GroupSession {
        
    
        /**
         * 创建一个聊天组, 如果不存在才能创建成功, 否则返回 null
         * @param name 组名
         * @param members 成员
         * @return 成功时返回组对象, 失败返回 null
         */
        Group createGroup(String name, Set<String> members);
    
        /**
         * 加入聊天组
         * @param name 组名
         * @param member 成员名
         * @return 如果组不存在返回 null, 否则返回组对象
         */
        Group joinMember(String name, String member);
    
        /**
         * 移除组成员
         * @param name 组名
         * @param member 成员名
         * @return 如果组不存在返回 null, 否则返回组对象
         */
        Group removeMember(String name, String member);
    
        /**
         * 移除聊天组
         * @param name 组名
         * @return 如果组不存在返回 null, 否则返回组对象
         */
        Group removeGroup(String name);
    
        /**
         * 获取组成员
         * @param name 组名
         * @return 成员集合, 没有成员会返回 empty set
         */
        Set<String> getMembers(String name);
    
        /**
         * 获取组成员的 channel 集合, 只有在线的 channel 才会返回
         * @param name 组名
         * @return 成员 channel 集合
         */
        List<Channel> getMembersChannel(String name);
    }

## 2 聊天室业务-登录 ##

@Slf4j
    public class ChatServer {
        
        public static void main(String[] args) {
        
            NioEventLoopGroup boss = new NioEventLoopGroup();
            NioEventLoopGroup worker = new NioEventLoopGroup();
            LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
            MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
            try {
        
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.channel(NioServerSocketChannel.class);
                serverBootstrap.group(boss, worker);
                serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
        
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
        
                        ch.pipeline().addLast(new ProcotolFrameDecoder());
                        ch.pipeline().addLast(LOGGING_HANDLER);
                        ch.pipeline().addLast(MESSAGE_CODEC);
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<LoginRequestMessage>() {
        
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
        
                                String username = msg.getUsername();
                                String password = msg.getPassword();
                                boolean login = UserServiceFactory.getUserService().login(username, password);
                                LoginResponseMessage message;
                                if(login) {
        
                                    message = new LoginResponseMessage(true, "登录成功");
                                } else {
        
                                    message = new LoginResponseMessage(false, "用户名或密码不正确");
                                }
                                ctx.writeAndFlush(message);
                            }
                        });
                    }
                });
                Channel channel = serverBootstrap.bind(8080).sync().channel();
                channel.closeFuture().sync();
            } catch (InterruptedException e) {
        
                log.error("server error", e);
            } finally {
        
                boss.shutdownGracefully();
                worker.shutdownGracefully();
            }
        }
    }

@Slf4j
    public class ChatClient {
        
        public static void main(String[] args) {
        
            NioEventLoopGroup group = new NioEventLoopGroup();
            LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
            MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
            CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);
            AtomicBoolean LOGIN = new AtomicBoolean(false);
            try {
        
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.channel(NioSocketChannel.class);
                bootstrap.group(group);
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
        
                        ch.pipeline().addLast(new ProcotolFrameDecoder());
    //                    ch.pipeline().addLast(LOGGING_HANDLER);
                        ch.pipeline().addLast(MESSAGE_CODEC);
                        ch.pipeline().addLast("client handler", new ChannelInboundHandlerAdapter() {
        
                            // 接收响应消息
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
                                log.debug("msg: {}", msg);
                                if ((msg instanceof LoginResponseMessage)) {
        
                                    LoginResponseMessage response = (LoginResponseMessage) msg;
                                    if (response.isSuccess()) {
        
                                        // 如果登录成功
                                        LOGIN.set(true);
                                    }
                                    // 唤醒 system in 线程
                                    WAIT_FOR_LOGIN.countDown();
                                }
                            }
    
                            // 在连接建立后触发 active 事件
                            @Override
                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
                                // 负责接收用户在控制台的输入，负责向服务器发送各种消息
                                new Thread(() -> {
        
                                    Scanner scanner = new Scanner(System.in);
                                    System.out.println("请输入用户名:");
                                    String username = scanner.nextLine();
                                    System.out.println("请输入密码:");
                                    String password = scanner.nextLine();
                                    // 构造消息对象
                                    LoginRequestMessage message = new LoginRequestMessage(username, password);
                                    // 发送消息
                                    ctx.writeAndFlush(message);
                                    System.out.println("等待后续操作...");
                                    try {
        
                                        WAIT_FOR_LOGIN.await();
                                    } catch (InterruptedException e) {
        
                                        e.printStackTrace();
                                    }
                                    // 如果登录失败
                                    if (!LOGIN.get()) {
        
                                        ctx.channel().close();
                                        return;
                                    }
                                    while (true) {
        
                                        System.out.println("==================================");
                                        System.out.println("send [username] [content]");
                                        System.out.println("gsend [group name] [content]");
                                        System.out.println("gcreate [group name] [m1,m2,m3...]");
                                        System.out.println("gmembers [group name]");
                                        System.out.println("gjoin [group name]");
                                        System.out.println("gquit [group name]");
                                        System.out.println("quit");
                                        System.out.println("==================================");
                                        String command = scanner.nextLine();
                                        String[] s = command.split(" ");
                                        switch (s[0]){
        
                                            case "send":
                                                ctx.writeAndFlush(new ChatRequestMessage(username, s[1], s[2]));
                                                break;
                                            case "gsend":
                                                ctx.writeAndFlush(new GroupChatRequestMessage(username, s[1], s[2]));
                                                break;
                                            case "gcreate":
                                                Set<String> set = new HashSet<>(Arrays.asList(s[2].split(",")));
                                                set.add(username); // 加入自己
                                                ctx.writeAndFlush(new GroupCreateRequestMessage(s[1], set));
                                                break;
                                            case "gmembers":
                                                ctx.writeAndFlush(new GroupMembersRequestMessage(s[1]));
                                                break;
                                            case "gjoin":
                                                ctx.writeAndFlush(new GroupJoinRequestMessage(username, s[1]));
                                                break;
                                            case "gquit":
                                                ctx.writeAndFlush(new GroupQuitRequestMessage(username, s[1]));
                                                break;
                                            case "quit":
                                                ctx.channel().close();
                                                return;
                                        }
                                    }
                                }, "system in").start();
                            }
                        });
                    }
                });
                Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
                channel.closeFuture().sync();
            } catch (Exception e) {
        
                log.error("client error", e);
            } finally {
        
                group.shutdownGracefully();
            }
        }
    }

## 3 聊天室业务-单聊 ##

服务器端将 handler 独立出来

@ChannelHandler.Sharable
    public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<LoginRequestMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
        
            String username = msg.getUsername();
            String password = msg.getPassword();
            boolean login = UserServiceFactory.getUserService().login(username, password);
            LoginResponseMessage message;
            if(login) {
        
                SessionFactory.getSession().bind(ctx.channel(), username);
                message = new LoginResponseMessage(true, "登录成功");
            } else {
        
                message = new LoginResponseMessage(false, "用户名或密码不正确");
            }
            ctx.writeAndFlush(message);
        }
    }

单聊 handler

@ChannelHandler.Sharable
    public class ChatRequestMessageHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ChatRequestMessage msg) throws Exception {
        
            String to = msg.getTo();
            Channel channel = SessionFactory.getSession().getChannel(to);
            // 在线
            if(channel != null) {
        
                channel.writeAndFlush(new ChatResponseMessage(msg.getFrom(), msg.getContent()));
            }
            // 不在线
            else {
        
                ctx.writeAndFlush(new ChatResponseMessage(false, "对方用户不存在或者不在线"));
            }
        }
    }

## 4 聊天室业务-群聊 ##

创建群聊

@ChannelHandler.Sharable
    public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, GroupCreateRequestMessage msg) throws Exception {
        
            String groupName = msg.getGroupName();
            Set<String> members = msg.getMembers();
            // 群管理器
            GroupSession groupSession = GroupSessionFactory.getGroupSession();
            Group group = groupSession.createGroup(groupName, members);
            if (group == null) {
        
                // 发生成功消息
                ctx.writeAndFlush(new GroupCreateResponseMessage(true, groupName + "创建成功"));
                // 发送拉群消息
                List<Channel> channels = groupSession.getMembersChannel(groupName);
                for (Channel channel : channels) {
        
                    channel.writeAndFlush(new GroupCreateResponseMessage(true, "您已被拉入" + groupName));
                }
            } else {
        
                ctx.writeAndFlush(new GroupCreateResponseMessage(false, groupName + "已经存在"));
            }
        }
    }

群聊

@ChannelHandler.Sharable
    public class GroupChatRequestMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, GroupChatRequestMessage msg) throws Exception {
        
            List<Channel> channels = GroupSessionFactory.getGroupSession()
                    .getMembersChannel(msg.getGroupName());
    
            for (Channel channel : channels) {
        
                channel.writeAndFlush(new GroupChatResponseMessage(msg.getFrom(), msg.getContent()));
            }
        }
    }

加入群聊

@ChannelHandler.Sharable
    public class GroupJoinRequestMessageHandler extends SimpleChannelInboundHandler<GroupJoinRequestMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, GroupJoinRequestMessage msg) throws Exception {
        
            Group group = GroupSessionFactory.getGroupSession().joinMember(msg.getGroupName(), msg.getUsername());
            if (group != null) {
        
                ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群加入成功"));
            } else {
        
                ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群不存在"));
            }
        }
    }

退出群聊

@ChannelHandler.Sharable
    public class GroupQuitRequestMessageHandler extends SimpleChannelInboundHandler<GroupQuitRequestMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, GroupQuitRequestMessage msg) throws Exception {
        
            Group group = GroupSessionFactory.getGroupSession().removeMember(msg.getGroupName(), msg.getUsername());
            if (group != null) {
        
                ctx.writeAndFlush(new GroupJoinResponseMessage(true, "已退出群" + msg.getGroupName()));
            } else {
        
                ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群不存在"));
            }
        }
    }

查看成员

@ChannelHandler.Sharable
    public class GroupMembersRequestMessageHandler extends SimpleChannelInboundHandler<GroupMembersRequestMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, GroupMembersRequestMessage msg) throws Exception {
        
            Set<String> members = GroupSessionFactory.getGroupSession()
                    .getMembers(msg.getGroupName());
            ctx.writeAndFlush(new GroupMembersResponseMessage(members));
        }
    }

## 5 聊天室业务-退出 ##

@Slf4j
    @ChannelHandler.Sharable
    public class QuitHandler extends ChannelInboundHandlerAdapter {
    
        // 当连接断开时触发 inactive 事件
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            SessionFactory.getSession().unbind(ctx.channel());
            log.debug("{} 已经断开", ctx.channel());
        }
    
    	// 当出现异常时触发
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            SessionFactory.getSession().unbind(ctx.channel());
            log.debug("{} 已经异常断开 异常是{}", ctx.channel(), cause.getMessage());
        }
    }

## 6 聊天室业务-空闲检测 ##

### 6.1 连接假死 ###

原因

*  网络设备出现故障，例如网卡，机房等，底层的 TCP 连接已经断开了，但应用程序没有感知到，仍然占用着资源。
 *  公网网络不稳定，出现丢包。如果连续出现丢包，这时现象就是客户端数据发不出去，服务端也一直收不到数据，就这么一直耗着
 *  应用程序线程阻塞，无法进行数据读写

问题

*  假死的连接占用的资源不能自动释放
 *  向假死的连接发送数据，得到的反馈是发送超时

服务器端解决

*  怎么判断客户端连接是否假死呢？如果能收到客户端数据，说明没有假死。因此策略就可以定为，每隔一段时间就检查这段时间内是否接收到客户端数据，没有就可以判定为连接假死

// 用来判断是不是 读空闲时间过长，或 写空闲时间过长
    // 5s 内如果没有收到 channel 的数据，会触发一个 IdleState#READER_IDLE 事件
    ch.pipeline().addLast(new IdleStateHandler(5, 0, 0));
    // ChannelDuplexHandler 可以同时作为入站和出站处理器
    ch.pipeline().addLast(new ChannelDuplexHandler() {
        
        // 用来触发特殊事件
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception{
        
            IdleStateEvent event = (IdleStateEvent) evt;
            // 触发了读空闲事件
            if (event.state() == IdleState.READER_IDLE) {
        
                log.debug("已经 5s 没有读到数据了");
                ctx.channel().close();
            }
        }
    });

客户端定时心跳

*  客户端可以定时向服务器端发送数据，只要这个时间间隔小于服务器定义的空闲检测的时间间隔，那么就能防止前面提到的误判，客户端可以定义如下心跳处理器

// 用来判断是不是 读空闲时间过长，或 写空闲时间过长
    // 3s 内如果没有向服务器写数据，会触发一个 IdleState#WRITER_IDLE 事件
    ch.pipeline().addLast(new IdleStateHandler(0, 3, 0));
    // ChannelDuplexHandler 可以同时作为入站和出站处理器
    ch.pipeline().addLast(new ChannelDuplexHandler() {
        
        // 用来触发特殊事件
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception{
        
            IdleStateEvent event = (IdleStateEvent) evt;
            // 触发了写空闲事件
            if (event.state() == IdleState.WRITER_IDLE) {
        
                //                                log.debug("3s 没有写数据了，发送一个心跳包");
                ctx.writeAndFlush(new PingMessage());
            }
        }
    });

[b7b1b929a7974e67bbe75522e52bd62c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/f43fedb77378422482ad6d0d815bcc04.png