由浅入深Netty聊天室案例

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

目录

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

1 聊天室业务介绍

在这里插入图片描述

  1. /**
  2.  * 用户管理接口
  3.  */
  4. public interface UserService {
  7.     /**
  8.      * 登录
  9.      * @param username 用户名
  10.      * @param password 密码
  11.      * @return 登录成功返回 true, 否则返回 false
  12.      */
  13.     boolean login(String username, String password);
  14. }
  16. /**
  17.  * 会话管理接口
  18.  */
  19. public interface Session {
  22.     /**
  23.      * 绑定会话
  24.      * @param channel 哪个 channel 要绑定会话
  25.      * @param username 会话绑定用户
  26.      */
  27.     void bind(Channel channel, String username);
  29.     /**
  30.      * 解绑会话
  31.      * @param channel 哪个 channel 要解绑会话
  32.      */
  33.     void unbind(Channel channel);
  35.     /**
  36.      * 获取属性
  37.      * @param channel 哪个 channel
  38.      * @param name 属性名
  39.      * @return 属性值
  40.      */
  41.     Object getAttribute(Channel channel, String name);
  43.     /**
  44.      * 设置属性
  45.      * @param channel 哪个 channel
  46.      * @param name 属性名
  47.      * @param value 属性值
  48.      */
  49.     void setAttribute(Channel channel, String name, Object value);
  51.     /**
  52.      * 根据用户名获取 channel
  53.      * @param username 用户名
  54.      * @return channel
  55.      */
  56.     Channel getChannel(String username);
  57. }
  59. /**
  60.  * 聊天组会话管理接口
  61.  */
  62. public interface GroupSession {
  65.     /**
  66.      * 创建一个聊天组, 如果不存在才能创建成功, 否则返回 null
  67.      * @param name 组名
  68.      * @param members 成员
  69.      * @return 成功时返回组对象, 失败返回 null
  70.      */
  71.     Group createGroup(String name, Set<String> members);
  73.     /**
  74.      * 加入聊天组
  75.      * @param name 组名
  76.      * @param member 成员名
  77.      * @return 如果组不存在返回 null, 否则返回组对象
  78.      */
  79.     Group joinMember(String name, String member);
  81.     /**
  82.      * 移除组成员
  83.      * @param name 组名
  84.      * @param member 成员名
  85.      * @return 如果组不存在返回 null, 否则返回组对象
  86.      */
  87.     Group removeMember(String name, String member);
  89.     /**
  90.      * 移除聊天组
  91.      * @param name 组名
  92.      * @return 如果组不存在返回 null, 否则返回组对象
  93.      */
  94.     Group removeGroup(String name);
  96.     /**
  97.      * 获取组成员
  98.      * @param name 组名
  99.      * @return 成员集合, 没有成员会返回 empty set
  100.      */
  101.     Set<String> getMembers(String name);
  103.     /**
  104.      * 获取组成员的 channel 集合, 只有在线的 channel 才会返回
  105.      * @param name 组名
  106.      * @return 成员 channel 集合
  107.      */
  108.     List<Channel> getMembersChannel(String name);
  109. }

2 聊天室业务-登录

  1. @Slf4j
  2. public class ChatServer {
  4.     public static void main(String[] args) {
  6.         NioEventLoopGroup boss = new NioEventLoopGroup();
  7.         NioEventLoopGroup worker = new NioEventLoopGroup();
  8.         LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
  9.         MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  10.         try {
  12.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  13.             serverBootstrap.channel(NioServerSocketChannel.class);
  14.             serverBootstrap.group(boss, worker);
  15.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  17.                 @Override
  18.                 protected void initChannel(SocketChannel ch) throws Exception {
  20.                     ch.pipeline().addLast(new ProcotolFrameDecoder());
  21.                     ch.pipeline().addLast(LOGGING_HANDLER);
  22.                     ch.pipeline().addLast(MESSAGE_CODEC);
  23.                     ch.pipeline().addLast(new SimpleChannelInboundHandler<LoginRequestMessage>() {
  25.                         @Override
  26.                         protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
  28.                             String username = msg.getUsername();
  29.                             String password = msg.getPassword();
  30.                             boolean login = UserServiceFactory.getUserService().login(username, password);
  31.                             LoginResponseMessage message;
  32.                             if(login) {
  34.                                 message = new LoginResponseMessage(true, "登录成功");
  35.                             } else {
  37.                                 message = new LoginResponseMessage(false, "用户名或密码不正确");
  38.                             }
  39.                             ctx.writeAndFlush(message);
  40.                         }
  41.                     });
  42.                 }
  43.             });
  44.             Channel channel = serverBootstrap.bind(8080).sync().channel();
  45.             channel.closeFuture().sync();
  46.         } catch (InterruptedException e) {
  48.             log.error("server error", e);
  49.         } finally {
  51.             boss.shutdownGracefully();
  52.             worker.shutdownGracefully();
  53.         }
  54.     }
  55. }
  57. @Slf4j
  58. public class ChatClient {
  60.     public static void main(String[] args) {
  62.         NioEventLoopGroup group = new NioEventLoopGroup();
  63.         LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
  64.         MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  65.         CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);
  66.         AtomicBoolean LOGIN = new AtomicBoolean(false);
  67.         try {
  69.             Bootstrap bootstrap = new Bootstrap();
  70.             bootstrap.channel(NioSocketChannel.class);
  71.             bootstrap.group(group);
  72.             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  74.                 @Override
  75.                 protected void initChannel(SocketChannel ch) throws Exception {
  77.                     ch.pipeline().addLast(new ProcotolFrameDecoder());
  78. //                    ch.pipeline().addLast(LOGGING_HANDLER);
  79.                     ch.pipeline().addLast(MESSAGE_CODEC);
  80.                     ch.pipeline().addLast("client handler", new ChannelInboundHandlerAdapter() {
  82.                         // 接收响应消息
  83.                         @Override
  84.                         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  86.                             log.debug("msg: {}", msg);
  87.                             if ((msg instanceof LoginResponseMessage)) {
  89.                                 LoginResponseMessage response = (LoginResponseMessage) msg;
  90.                                 if (response.isSuccess()) {
  92.                                     // 如果登录成功
  93.                                     LOGIN.set(true);
  94.                                 }
  95.                                 // 唤醒 system in 线程
  96.                                 WAIT_FOR_LOGIN.countDown();
  97.                             }
  98.                         }
  100.                         // 在连接建立后触发 active 事件
  101.                         @Override
  102.                         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  104.                             // 负责接收用户在控制台的输入，负责向服务器发送各种消息
  105.                             new Thread(() -> {
  107.                                 Scanner scanner = new Scanner(System.in);
  108.                                 System.out.println("请输入用户名:");
  109.                                 String username = scanner.nextLine();
  110.                                 System.out.println("请输入密码:");
  111.                                 String password = scanner.nextLine();
  112.                                 // 构造消息对象
  113.                                 LoginRequestMessage message = new LoginRequestMessage(username, password);
  114.                                 // 发送消息
  115.                                 ctx.writeAndFlush(message);
  116.                                 System.out.println("等待后续操作...");
  117.                                 try {
  119.                                     WAIT_FOR_LOGIN.await();
  120.                                 } catch (InterruptedException e) {
  122.                                     e.printStackTrace();
  123.                                 }
  124.                                 // 如果登录失败
  125.                                 if (!LOGIN.get()) {
  127.                                     ctx.channel().close();
  128.                                     return;
  129.                                 }
  130.                                 while (true) {
  132.                                     System.out.println("==================================");
  133.                                     System.out.println("send [username] [content]");
  134.                                     System.out.println("gsend [group name] [content]");
  135.                                     System.out.println("gcreate [group name] [m1,m2,m3...]");
  136.                                     System.out.println("gmembers [group name]");
  137.                                     System.out.println("gjoin [group name]");
  138.                                     System.out.println("gquit [group name]");
  139.                                     System.out.println("quit");
  140.                                     System.out.println("==================================");
  141.                                     String command = scanner.nextLine();
  142.                                     String[] s = command.split(" ");
  143.                                     switch (s[0]){
  145.                                         case "send":
  146.                                             ctx.writeAndFlush(new ChatRequestMessage(username, s[1], s[2]));
  147.                                             break;
  148.                                         case "gsend":
  149.                                             ctx.writeAndFlush(new GroupChatRequestMessage(username, s[1], s[2]));
  150.                                             break;
  151.                                         case "gcreate":
  152.                                             Set<String> set = new HashSet<>(Arrays.asList(s[2].split(",")));
  153.                                             set.add(username); // 加入自己
  154.                                             ctx.writeAndFlush(new GroupCreateRequestMessage(s[1], set));
  155.                                             break;
  156.                                         case "gmembers":
  157.                                             ctx.writeAndFlush(new GroupMembersRequestMessage(s[1]));
  158.                                             break;
  159.                                         case "gjoin":
  160.                                             ctx.writeAndFlush(new GroupJoinRequestMessage(username, s[1]));
  161.                                             break;
  162.                                         case "gquit":
  163.                                             ctx.writeAndFlush(new GroupQuitRequestMessage(username, s[1]));
  164.                                             break;
  165.                                         case "quit":
  166.                                             ctx.channel().close();
  167.                                             return;
  168.                                     }
  169.                                 }
  170.                             }, "system in").start();
  171.                         }
  172.                     });
  173.                 }
  174.             });
  175.             Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
  176.             channel.closeFuture().sync();
  177.         } catch (Exception e) {
  179.             log.error("client error", e);
  180.         } finally {
  182.             group.shutdownGracefully();
  183.         }
  184.     }
  185. }

3 聊天室业务-单聊

服务器端将 handler 独立出来

登录 handler

  1. @ChannelHandler.Sharable
  2. public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<LoginRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
  7.         String username = msg.getUsername();
  8.         String password = msg.getPassword();
  9.         boolean login = UserServiceFactory.getUserService().login(username, password);
  10.         LoginResponseMessage message;
  11.         if(login) {
  13.             SessionFactory.getSession().bind(ctx.channel(), username);
  14.             message = new LoginResponseMessage(true, "登录成功");
  15.         } else {
  17.             message = new LoginResponseMessage(false, "用户名或密码不正确");
  18.         }
  19.         ctx.writeAndFlush(message);
  20.     }
  21. }

单聊 handler

  1. @ChannelHandler.Sharable
  2. public class ChatRequestMessageHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, ChatRequestMessage msg) throws Exception {
  7.         String to = msg.getTo();
  8.         Channel channel = SessionFactory.getSession().getChannel(to);
  9.         // 在线
  10.         if(channel != null) {
  12.             channel.writeAndFlush(new ChatResponseMessage(msg.getFrom(), msg.getContent()));
  13.         }
  14.         // 不在线
  15.         else {
  17.             ctx.writeAndFlush(new ChatResponseMessage(false, "对方用户不存在或者不在线"));
  18.         }
  19.     }
  20. }

4 聊天室业务-群聊

创建群聊

  1. @ChannelHandler.Sharable
  2. public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, GroupCreateRequestMessage msg) throws Exception {
  7.         String groupName = msg.getGroupName();
  8.         Set<String> members = msg.getMembers();
  9.         // 群管理器
  10.         GroupSession groupSession = GroupSessionFactory.getGroupSession();
  11.         Group group = groupSession.createGroup(groupName, members);
  12.         if (group == null) {
  14.             // 发生成功消息
  15.             ctx.writeAndFlush(new GroupCreateResponseMessage(true, groupName + "创建成功"));
  16.             // 发送拉群消息
  17.             List<Channel> channels = groupSession.getMembersChannel(groupName);
  18.             for (Channel channel : channels) {
  20.                 channel.writeAndFlush(new GroupCreateResponseMessage(true, "您已被拉入" + groupName));
  21.             }
  22.         } else {
  24.             ctx.writeAndFlush(new GroupCreateResponseMessage(false, groupName + "已经存在"));
  25.         }
  26.     }
  27. }

群聊

  1. @ChannelHandler.Sharable
  2. public class GroupChatRequestMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, GroupChatRequestMessage msg) throws Exception {
  7.         List<Channel> channels = GroupSessionFactory.getGroupSession()
  8.                 .getMembersChannel(msg.getGroupName());
  10.         for (Channel channel : channels) {
  12.             channel.writeAndFlush(new GroupChatResponseMessage(msg.getFrom(), msg.getContent()));
  13.         }
  14.     }
  15. }

加入群聊

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

退出群聊

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

查看成员

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

5 聊天室业务-退出

  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. public class QuitHandler extends ChannelInboundHandlerAdapter {
  5.     // 当连接断开时触发 inactive 事件
  6.     @Override
  7.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  8.         SessionFactory.getSession().unbind(ctx.channel());
  9.         log.debug("{} 已经断开", ctx.channel());
  10.     }
  12.     // 当出现异常时触发
  13.     @Override
  14.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  15.         SessionFactory.getSession().unbind(ctx.channel());
  16.         log.debug("{} 已经异常断开 异常是{}", ctx.channel(), cause.getMessage());
  17.     }
  18. }

6 聊天室业务-空闲检测

6.1 连接假死

原因

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

问题

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

服务器端解决

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

    // 用来判断是不是 读空闲时间过长，或 写空闲时间过长
    // 5s 内如果没有收到 channel 的数据，会触发一个 IdleState#READER_IDLE 事件
    ch.pipeline().addLast(new IdleStateHandler(5, 0, 0));
    // ChannelDuplexHandler 可以同时作为入站和出站处理器
    ch.pipeline().addLast(new ChannelDuplexHandler() {

    1. // 用来触发特殊事件
    2. @Override
    3. public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception{
    5.     IdleStateEvent event = (IdleStateEvent) evt;
    6.     // 触发了读空闲事件
    7.     if (event.state() == IdleState.READER_IDLE) {
    9.         log.debug("已经 5s 没有读到数据了");
    10.         ctx.channel().close();
    11.     }
    12. }

    });

客户端定时心跳

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

    // 用来判断是不是 读空闲时间过长，或 写空闲时间过长
    // 3s 内如果没有向服务器写数据，会触发一个 IdleState#WRITER_IDLE 事件
    ch.pipeline().addLast(new IdleStateHandler(0, 3, 0));
    // ChannelDuplexHandler 可以同时作为入站和出站处理器
    ch.pipeline().addLast(new ChannelDuplexHandler() {

    1. // 用来触发特殊事件
    2. @Override
    3. public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception{
    5.     IdleStateEvent event = (IdleStateEvent) evt;
    6.     // 触发了写空闲事件
    7.     if (event.state() == IdleState.WRITER_IDLE) {
    9.         //                                log.debug("3s 没有写数据了，发送一个心跳包");
    10.         ctx.writeAndFlush(new PingMessage());
    11.     }
    12. }

    });

发表评论

表情：
评论列表 （有 0 条评论，120人围观）

还没有评论，来说两句吧...

相关阅读