---

前言

笔记基于黑马的Netty教学讲义加上自己的一些理解，感觉这是看过的视频中挺不错的，基本没有什么废话，视频地址：黑马Netty，这里的聊天室是用的黑马提供的资料，跟着黑马来敲的，补齐了视频中没有写的退出群聊和查看成员方法。

首页这里的代码没有涉及到数据库，对于注册等方法以后进行完善，其次里面的一些方法写得不是很完整，考虑的情况不全，但是只要达到学习的目的就够了，知道聊天室消息的流程就够了

聊天室群聊

1. 思路

• 账号密码存在后台中，登陆完成之后和 Server 建立连接之后，把 channel 和 用户名字关联起来存到一个 map 中
• 对于每个 handler 的处理我们都继承 SimpleChannelInboundHandler 这个类，对于不同的消息，进行不同的处理
• 每个 handler 都是 @Sharable 的，要加上这个注解
• 对于每个人和每个群聊都用 map 进行对 channel 的存储
• 在发送消息的时候，群聊就是通过 map 找出所有的 channel，给出了自己的其他人发送消息
• 对于每一个 client，为了防止异常的连接比如网络卡这些影响，我们需要定时发送心跳包给服务端

2. 代码

1. handler，用于处理消息

1. 好友聊天消息处理器

1. 拿到好友姓名
2. 通过好友姓名查出channel
3. 发送消息
4. 如果对方不在线就响应对方不在线给自己

@ChannelHandler.Sharable
public class ChatRequestHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, ChatRequestMessage msg) throws Exception {//发给谁String to = msg.getTo();//找到channelChannel channel = SessionFactory.getSession().getChannel(to);//对方在线if(channel != null){channel.writeAndFlush(new ChatResponseMessage(msg.getFrom(), msg.getContent()));}//对方不在线else{ctx.writeAndFlush(new ChatResponseMessage(false, "对方用户不在线"));}}
}

2. 群聊消息处理器

1. 获取群聊名字
2. 获取发送的内容
3. 获取这个群里面所有的成员的 channel
4. 给出了自己的所有 channel 发送消息

@ChannelHandler.Sharable
public class GroupChatRequestMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, GroupChatRequestMessage msg) throws Exception {//内容String content = msg.getContent();//获取成员List<Channel> membersChannel = GroupSessionFactory.getGroupSession().getMembersChannel(msg.getGroupName());if(membersChannel == null){ctx.writeAndFlush(new GroupChatResponseMessage(false, "群聊不存在"));}else{for (Channel channel : membersChannel) {if(!ctx.channel().equals(channel)){//不给自己发消息channel.writeAndFlush(new GroupChatResponseMessage(msg.getFrom(), msg.getContent()));}}}}
}

3. 创建群消息处理器

1. 获取群聊名字
2. 获取到要拉进去的人
3. 创建一个 Group，create 方法进行存储到 map 了
4. 给拉进群的人发送消息，通知他们被拉进群了

//不存在状态信息，可共享
@ChannelHandler.Sharable
public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {@Overrideprotected 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){//向用户发送拉入群聊消息List<Channel> channels = groupSession.getMembersChannel(groupName);for (Channel channel : channels) {if(!channel.equals(ctx.channel())){channel.writeAndFlush(new GroupCreateResponseMessage(true, "您已被拉入群聊" + groupName));}}//发送成功消息ctx.writeAndFlush(new GroupCreateResponseMessage(true, "创建群聊成功"));}else{ctx.writeAndFlush(new GroupCreateResponseMessage(false, "群已经存在了"));}}
}

4. 加入群聊消息处理器

1. 获取自己的channel
2. 把自己添加到 group 的 map 中
3. 拿到这个群里面的人的 channel
4. 给除了自己的所有人发通知有人加群了

@ChannelHandler.Sharable
public class GroupJoinRequestMessageHandler extends SimpleChannelInboundHandler<GroupJoinRequestMessage> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, GroupJoinRequestMessage msg) throws Exception {Channel channel = ctx.channel();//加入groupGroupSessionFactory.getGroupSession().joinMember(msg.getGroupName(), msg.getUsername());List<Channel> membersChannel = GroupSessionFactory.getGroupSession().getMembersChannel(msg.getGroupName());//发送消息for (Channel users : membersChannel) {if(!users.equals(channel)){users.writeAndFlush(new GroupJoinResponseMessage(true, msg.getUsername() + "加入了聊天室"));}}}
}

5. 查看群成员消息处理器

1. 获取群聊名字
2. 获取到所有的成员的 channel
3. 通过 channel 找到对应的名字

@ChannelHandler.Sharable
public class GroupMembersRequestMessageHandler extends SimpleChannelInboundHandler<GroupMembersRequestMessage> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, GroupMembersRequestMessage msg) throws Exception {String groupName = msg.getGroupName();//获取List<String> users = GroupSessionFactory.getGroupSession().getUsers(groupName);ctx.channel().writeAndFlush("组 " + groupName + " 的成员是：" + users.toString());}
}

6. 退出群消息处理器

1. 拿到群聊名字
2. 获取群里面的人的 channel
3. 发送消息通知

@ChannelHandler.Sharable
public class GroupQuitRequestMessageHandler extends SimpleChannelInboundHandler<GroupQuitRequestMessage> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, GroupQuitRequestMessage msg) throws Exception {String username = msg.getUsername();String groupName = msg.getGroupName();//移除sessionGroupSessionFactory.getGroupSession().removeMember(groupName, username);for (Channel channel : GroupSessionFactory.getGroupSession().getMembersChannel(groupName)) {channel.writeAndFlush(new GroupQuitResponseMessage(true, "用户" + username + "退出了群聊"));}}
}

7. 登陆消息处理器

1. 获取用户名密码，进行检测
2. 把 channel 和 名字相对应起来存入 map 中
3. 发通知

@ChannelHandler.Sharable
public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<LoginRequestMessage> {@Overrideprotected 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 = null;if (login) {message = new LoginResponseMessage(true, "登陆成功");//保存channelSessionFactory.getSession().bind(ctx.channel(), username);} else {message = new LoginResponseMessage(false, "用户名或者密码不正确");}ctx.writeAndFlush(message);}
}

8. 退出程序处理器

1. 两种情况，一种是连接断开，一种是异常
2. 都要对应处理，把 channel 移除掉

@Slf4j
@ChannelHandler.Sharable
//我们独立写一个类，因为不是和聊天相关的了
public class QuitHandler extends ChannelInboundHandlerAdapter {//连接断开的时候会触发 Inactive@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {//从 SessionFactory 中移除channel，断开了SessionFactory.getSession().unbind(ctx.channel());log.debug("{} 已经断开", ctx.channel());}//异常是会触发@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {SessionFactory.getSession().unbind(ctx.channel());log.debug("{} 已经异常断开，异常是：{}", ctx.channel(), cause.getMessage());}
}

2. 消息类，用于发送返回

1. AbstractResponseMessage

@Data
@ToString(callSuper = true)
public abstract class AbstractResponseMessage extends Message {private boolean success;private String reason;public AbstractResponseMessage() {}public AbstractResponseMessage(boolean success, String reason) {this.success = success;this.reason = reason;}
}

2. ChatRequestMessage

@Data
@ToString(callSuper = true)
public class ChatRequestMessage extends Message {private String content;private String to;private String from;public ChatRequestMessage() {}public ChatRequestMessage(String from, String to, String content) {this.from = from;this.to = to;this.content = content;}@Overridepublic int getMessageType() {return ChatRequestMessage;}
}

3. ChatResponseMessage

@Data
@ToString(callSuper = true)
public class ChatResponseMessage extends AbstractResponseMessage {private String from;private String content;public ChatResponseMessage(boolean success, String reason) {super(success, reason);}public ChatResponseMessage(String from, String content) {this.from = from;this.content = content;}@Overridepublic int getMessageType() {return ChatResponseMessage;}
}

4. GroupChatRequestMessage

@Data
@ToString(callSuper = true)
public class GroupChatRequestMessage extends Message {private String content;private String groupName;private String from;public GroupChatRequestMessage(String from, String groupName, String content) {this.content = content;this.groupName = groupName;this.from = from;}@Overridepublic int getMessageType() {return GroupChatRequestMessage;}
}

5. GroupChatResponseMessage

@Data
@ToString(callSuper = true)
public class GroupChatResponseMessage extends AbstractResponseMessage {private String from;private String content;public GroupChatResponseMessage(boolean success, String reason) {super(success, reason);}public GroupChatResponseMessage(String from, String content) {this.from = from;this.content = content;}@Overridepublic int getMessageType() {return GroupChatResponseMessage;}
}

6. GroupCreateRequestMessage

@Data
@ToString(callSuper = true)
public class GroupCreateRequestMessage extends Message {private String groupName;private Set<String> members;public GroupCreateRequestMessage(String groupName, Set<String> members) {this.groupName = groupName;this.members = members;}@Overridepublic int getMessageType() {return GroupCreateRequestMessage;}
}

7. GroupCreateResponseMessage

@Data
@ToString(callSuper = true)
public class GroupCreateResponseMessage extends AbstractResponseMessage {public GroupCreateResponseMessage(boolean success, String reason) {super(success, reason);}@Overridepublic int getMessageType() {return GroupCreateResponseMessage;}
}

8. GroupJoinRequestMessage

@Data
@ToString(callSuper = true)
public class GroupJoinRequestMessage extends Message {private String groupName;private String username;public GroupJoinRequestMessage(String username, String groupName) {this.groupName = groupName;this.username = username;}@Overridepublic int getMessageType() {return GroupJoinRequestMessage;}
}

9. GroupJoinResponseMessage

@Data
@ToString(callSuper = true)
public class GroupJoinResponseMessage extends AbstractResponseMessage {public GroupJoinResponseMessage(boolean success, String reason) {super(success, reason);}@Overridepublic int getMessageType() {return GroupJoinResponseMessage;}
}

10. GroupMembersRequestMessage

@Data
@ToString(callSuper = true)
public class GroupMembersRequestMessage extends Message {private String groupName;public GroupMembersRequestMessage(String groupName) {this.groupName = groupName;}@Overridepublic int getMessageType() {return GroupMembersRequestMessage;}
}

11. GroupMembersResponseMessage

@Data
@ToString(callSuper = true)
public class GroupMembersResponseMessage extends Message {private Set<String> members;public GroupMembersResponseMessage(Set<String> members) {this.members = members;}@Overridepublic int getMessageType() {return GroupMembersResponseMessage;}
}

12. GroupQuitRequestMessage

@Data
@ToString(callSuper = true)
public class GroupQuitRequestMessage extends Message {private String groupName;private String username;public GroupQuitRequestMessage(String username, String groupName) {this.groupName = groupName;this.username = username;}@Overridepublic int getMessageType() {return GroupQuitRequestMessage;}
}

13. GroupQuitResponseMessage

@Data
@ToString(callSuper = true)
public class GroupQuitResponseMessage extends AbstractResponseMessage {public GroupQuitResponseMessage(boolean success, String reason) {super(success, reason);}@Overridepublic int getMessageType() {return GroupQuitResponseMessage;}
}

14. LoginRequestMessage

@Data
@ToString(callSuper = true)
public class LoginRequestMessage extends Message {private String username;private String password;public LoginRequestMessage() {}public LoginRequestMessage(String username, String password) {this.username = username;this.password = password;}@Overridepublic int getMessageType() {return LoginRequestMessage;}
}

15. LoginResponseMessage

@Data
@ToString(callSuper = true)
public class LoginResponseMessage extends AbstractResponseMessage {public LoginResponseMessage(boolean success, String reason) {super(success, reason);}@Overridepublic int getMessageType() {return LoginResponseMessage;}
}

15. Message

@Data
public abstract class Message implements Serializable {public static Class<?> getMessageClass(int messageType) {return messageClasses.get(messageType);}private int sequenceId;private int messageType;public abstract int getMessageType();public static final int LoginRequestMessage = 0;public static final int LoginResponseMessage = 1;public static final int ChatRequestMessage = 2;public static final int ChatResponseMessage = 3;public static final int GroupCreateRequestMessage = 4;public static final int GroupCreateResponseMessage = 5;public static final int GroupJoinRequestMessage = 6;public static final int GroupJoinResponseMessage = 7;public static final int GroupQuitRequestMessage = 8;public static final int GroupQuitResponseMessage = 9;public static final int GroupChatRequestMessage = 10;public static final int GroupChatResponseMessage = 11;public static final int GroupMembersRequestMessage = 12;public static final int GroupMembersResponseMessage = 13;public static final int PingMessage = 14;public static final int PongMessage = 15;private static final Map<Integer, Class<?>> messageClasses = new HashMap<>();static {messageClasses.put(LoginRequestMessage, LoginRequestMessage.class);messageClasses.put(LoginResponseMessage, LoginResponseMessage.class);messageClasses.put(ChatRequestMessage, ChatRequestMessage.class);messageClasses.put(ChatResponseMessage, ChatResponseMessage.class);messageClasses.put(GroupCreateRequestMessage, GroupCreateRequestMessage.class);messageClasses.put(GroupCreateResponseMessage, GroupCreateResponseMessage.class);messageClasses.put(GroupJoinRequestMessage, GroupJoinRequestMessage.class);messageClasses.put(GroupJoinResponseMessage, GroupJoinResponseMessage.class);messageClasses.put(GroupQuitRequestMessage, GroupQuitRequestMessage.class);messageClasses.put(GroupQuitResponseMessage, GroupQuitResponseMessage.class);messageClasses.put(GroupChatRequestMessage, GroupChatRequestMessage.class);messageClasses.put(GroupChatResponseMessage, GroupChatResponseMessage.class);messageClasses.put(GroupMembersRequestMessage, GroupMembersRequestMessage.class);messageClasses.put(GroupMembersResponseMessage, GroupMembersResponseMessage.class);messageClasses.put(PingMessage, GroupMembersResponseMessage.class);messageClasses.put(PongMessage, GroupMembersResponseMessage.class);}
}

16. PingMessage

public class PingMessage extends Message{@Overridepublic int getMessageType() {return PingMessage;}
}

17. PongMessage

public class PongMessage extends Message{@Overridepublic int getMessageType() {return PongMessage;}
}

3. Session类，用于存储 channel

1. Group:聊天室

@Data
/*** 聊天组，即聊天室*/
public class Group {// 聊天室名称private String name;// 聊天室成员private Set<String> members;public static final Group EMPTY_GROUP = new Group("empty", Collections.emptySet());public Group(String name, Set<String> members) {this.name = name;this.members = members;}
}

2. GroupSession：聊天室接口

/*** 聊天组会话管理接口*/
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);/*** 获取组成员的名字* @param name 组名*/List<String> getUsers(String name);
}

3. GroupSessionFactory

public abstract class GroupSessionFactory {private static GroupSession session = new GroupSessionMemoryImpl();public static GroupSession getGroupSession() {return session;}
}

4. GroupSessionFactoryImpl：群聊方法实现类

public class GroupSessionMemoryImpl implements GroupSession {private final Map<String, Group> groupMap = new ConcurrentHashMap<>();@Overridepublic Group createGroup(String name, Set<String> members) {Group group = new Group(name, members);return groupMap.putIfAbsent(name, group);}@Overridepublic Group joinMember(String name, String member) {return groupMap.computeIfPresent(name, (key, value) -> {value.getMembers().add(member);return value;});}@Overridepublic Group removeMember(String name, String member) {return groupMap.computeIfPresent(name, (key, value) -> {value.getMembers().remove(member);return value;});}@Overridepublic Group removeGroup(String name) {return groupMap.remove(name);}@Overridepublic Set<String> getMembers(String name) {return groupMap.getOrDefault(name, Group.EMPTY_GROUP).getMembers();}@Overridepublic List<Channel> getMembersChannel(String name) {//判断群聊存不存在if(groupMap.get(name) == null){return null;}return getMembers(name).stream().map(member -> SessionFactory.getSession().getChannel(member)).filter(Objects::nonNull).collect(Collectors.toList());}@Overridepublic List<String> getUsers(String name) {if(groupMap.get(name) == null){return null;}return new ArrayList<>(groupMap.get(name).getMembers());}
}

5. Session：会话管理接口

/*** 会话管理接口*/
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);
}

6. SessionFactory ：会话工厂类

public abstract class SessionFactory {private static Session session = new SessionMemoryImpl();public static Session getSession() {return session;}
}

7. SessionMemoryImpl：会话管理接口实现类

里面的方法就是绑定 channel 和 name 的

public class SessionMemoryImpl implements Session {private final Map<String, Channel> usernameChannelMap = new ConcurrentHashMap<>();private final Map<Channel, String> channelUsernameMap = new ConcurrentHashMap<>();private final Map<Channel,Map<String,Object>> channelAttributesMap = new ConcurrentHashMap<>();@Overridepublic void bind(Channel channel, String username) {usernameChannelMap.put(username, channel);channelUsernameMap.put(channel, username);channelAttributesMap.put(channel, new ConcurrentHashMap<>());}@Overridepublic void unbind(Channel channel) {String username = channelUsernameMap.remove(channel);usernameChannelMap.remove(username);channelAttributesMap.remove(channel);}@Overridepublic Object getAttribute(Channel channel, String name) {return channelAttributesMap.get(channel).get(name);}@Overridepublic void setAttribute(Channel channel, String name, Object value) {channelAttributesMap.get(channel).put(name, value);}@Overridepublic Channel getChannel(String username) {return usernameChannelMap.get(username);}@Overridepublic String toString() {return usernameChannelMap.toString();}
}

4. 登陆类

1. UserService：登陆管理接口

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

2. UserServiceFactory：返回userService的实现类

public abstract class UserServiceFactory {private static UserService userService = new UserServiceMemoryImpl();public static UserService getUserService() {return userService;}
}

3. UserServiceMemoryImpl：用户管理接口的实现类

public class UserServiceMemoryImpl implements UserService {private Map<String, String> allUserMap = new ConcurrentHashMap<>();{allUserMap.put("zhangsan", "123");allUserMap.put("lisi", "123");allUserMap.put("wangwu", "123");allUserMap.put("zhaoliu", "123");allUserMap.put("qianqi", "123");}@Overridepublic boolean login(String username, String password) {String pass = allUserMap.get(username);if (pass == null) {return false;}return pass.equals(password);}
}

5. 协议

1. MessageCodecSharable: 协议累，消息按这种格式传递和解析

@Slf4j
@ChannelHandler.Sharable
/*** 必须和 LengthFieldBasedFrameDecoder 一起使用，确保接到的 ByteBuf 消息是完整的*/
public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {@Overrideprotected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {ByteBuf out = ctx.alloc().buffer();// 1. 4字节的魔数，最好和java那些不一样，自己独特的out.writeBytes(new byte[]{1, 2, 3, 4});// 2. 1字节的版本,out.writeByte(1);// 3. 1字节的序列化方式，我们约定jdk是0, json是1out.writeByte(0);// 4. 1字节的指令类型，聊天消息还是登陆消息等等，加以区分out.writeByte(msg.getMessageType());// 5. 4个字节的指令请求序号，目前还用不上out.writeInt(msg.getSequenceId());// 无意义，对齐填充out.writeByte(0xff);// 6. 获取内容的字节数组，使用序列化的方式，也可以使用转为JSON的方式//    这里选择了序列化的方式，因为上面序列化方式我们选择了JDKByteArrayOutputStream bos = new ByteArrayOutputStream();ObjectOutputStream oos = new ObjectOutputStream(bos);oos.writeObject(msg);//oos把msg序列化结果写到ObjectOutputStream，ObjectOutputStream再把结果写到ByteArrayOutputStreambyte[] bytes = bos.toByteArray();// 7. 字节数组的长度out.writeInt(bytes.length);// 8. 写入内容out.writeBytes(bytes);//出了内容可变，其他都是固定的，字节大小：// 4+1+1+1+4+4 = 15个字节，我们使用out.writeByte(0xff) +1个字节凑够16个，仅仅是对齐用的，有个规定是2的n次方倍才叫专业outList.add(out);}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {int magicNum = in.readInt();byte version = in.readByte();byte serializerType = in.readByte();byte messageType = in.readByte();int sequenceId = in.readInt();in.readByte();int length = in.readInt();byte[] bytes = new byte[length];in.readBytes(bytes, 0, length);ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));Message message = (Message) ois.readObject();//结果放到out里面传给下一个入站处理器就行了out.add(message);}
}

2. 协议解码处理类, 配合上面的协议类一起用

public class ProcotolFrameDecoder extends LengthFieldBasedFrameDecoder {public ProcotolFrameDecoder() {this(1024, 12, 4, 0, 0);}public ProcotolFrameDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);}
}

6. 服务端

• 添加各种 handler
• 注意心跳机制的处理
• 其他就是很常规的代码了

@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();LoginRequestMessageHandler LOGIN_HANDLER = new LoginRequestMessageHandler();ChatRequestHandler CHAT_HANDLER = new ChatRequestHandler();GroupCreateRequestMessageHandler GROUP_CREATE_HANDLER = new GroupCreateRequestMessageHandler();GroupJoinRequestMessageHandler GROUP_JOIN_HANDLER = new GroupJoinRequestMessageHandler();GroupMembersRequestMessageHandler GROUP_MEMBERT_HANDLER = new GroupMembersRequestMessageHandler();GroupQuitRequestMessageHandler GROUP_QUIT_HANDLER = new GroupQuitRequestMessageHandler();GroupChatRequestMessageHandler CROUP_CHAR_HANDLER = new GroupChatRequestMessageHandler();QuitHandler QUIT_HANDLER = new QuitHandler();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.channel(NioServerSocketChannel.class);serverBootstrap.group(boss, worker);serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {//空闲状态检测器：心跳检测，假死的连接占用的资源不能自动释放//用来判断是不是读空闲时间过程或者写空闲时间过长// 参数1：读    参数2：写   参数3：读写// 5秒内如果没有收到channel读入的数据，就会触发一个事件 IdleState.READER_IDLEch.pipeline().addLast(new IdleStateHandler(5, 8, 0));//心跳检测的处理器 ChannelDuplexHandler可以同时作为入站和出站处理器ch.pipeline().addLast(new ChannelDuplexHandler(){//用来触发特殊事件@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {IdleStateEvent event = (IdleStateEvent) evt;if(event.state() == IdleState.READER_IDLE ){//触发了读空闲事件，//log.debug("已经5秒了没有读到数据了");ctx.channel().close();}}});//ProcotolFrameDecoder() 提取出来作为一个单独的类，channel独有的，不能共享ch.pipeline().addLast(new ProcotolFrameDecoder());//ch.pipeline().addLast(LOGGING_HANDLER);ch.pipeline().addLast(MESSAGE_CODEC);//我们用SimpleChannelInboundHandler对指定消息做操作//登陆处理器ch.pipeline().addLast(LOGIN_HANDLER);//好友聊天消息处理器ch.pipeline().addLast(CHAT_HANDLER);//群聊创建处理器ch.pipeline().addLast(GROUP_CREATE_HANDLER);//加入群聊处理器ch.pipeline().addLast(GROUP_JOIN_HANDLER);//群成员处理器ch.pipeline().addLast(GROUP_MEMBERT_HANDLER);//退出群聊处理器ch.pipeline().addLast(GROUP_QUIT_HANDLER);//群聊天处理器ch.pipeline().addLast(CROUP_CHAR_HANDLER);//正常异常退出客户端ch.pipeline().addLast(QUIT_HANDLER);}});Channel channel = serverBootstrap.bind(8080).sync().channel();channel.closeFuture().sync();} catch (InterruptedException e) {log.error("server error", e);} finally {boss.shutdownGracefully();worker.shutdownGracefully();}}}

7. 客户端

• 通过菜单对不同的消息进行不同的处理
• 使用 CountDownLatch 来完成线程间的通信

@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>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new ProcotolFrameDecoder());//ch.pipeline().addLast(LOGGING_HANDLER);ch.pipeline().addLast(MESSAGE_CODEC);// 3秒内如果没有发数据，就触发一个IdleState.，就会触发一个事件 IdleState.WRITER_IDLE写空闲ch.pipeline().addLast(new IdleStateHandler(0, 3, 0));//心跳检测的处理器 ChannelDuplexHandler可以同时作为入站和出站处理器ch.pipeline().addLast(new ChannelDuplexHandler(){//用来触发特殊事件@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {IdleStateEvent event = (IdleStateEvent) evt;if(event.state() == IdleState.WRITER_IDLE ){//触发了写空闲//发一个包证明还活着//log.debug("发一个心跳包，证明我还活着");ctx.writeAndFlush(new PingMessage());}}});ch.pipeline().addLast("clien handler", new ChannelInboundHandlerAdapter(){//接收服务器响应的消息,Nio线程调用的，而下面channelActive中输入是其他线程的，那如何让两个线程通信呢？//使用@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.debug("msg：{}", msg);//计数器 -1，为0，通知下面线程该运行了if(msg instanceof LoginResponseMessage){LoginResponseMessage response = (LoginResponseMessage) msg;if (response.isSuccess()) {LOGIN.set(true);}}//唤醒 System in线程WAIT_FOR_LOGIN.countDown();}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//连接建立后触发active事件,新弄一个线程接收用户在控制台的输入，向服务器发送各种线程//防止使用nio线程导致阻塞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;}default:throw new IllegalStateException("Unexpected value: " + s[0]);}}}, "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();}}
}

如有错误，欢迎指出！！！