1 聊天室业务介绍
/** * 用户管理接口 */ 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 独立出来
登录 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()); } } });
