即时通信|自定义基于 Netty 的二进制协议（应用层协议）+心跳检测

基于IM仿微信聊天的场景TCP传输层负责把字节流可靠地从A送到B自定义协议应用层负责规定字节流的含义┌──────────┬──────────┬──────────────────┐│ 4字节长度 │ 1字节命令 │ N字节业务数据 │└──────────┴──────────┴──────────────────┘↑ ↑解决粘包 区分是登录还是发消息TCP 只管送到不管你发的是登录请求还是聊天消息还是什么其他消息那是应用层的事。1导入在BIO通信入门中提到原生 Java 写的网络通信不是可以通过线程池来解决线程数量过多的问题吗那我们为什么要引入Netty呢?// 用线程池限制线程数量 ExecutorService pool Executors.newFixedThreadPool(200); // 最多200个线程 ServerSocket server new ServerSocket(7397); while(true) { Socket client server.accept(); pool.submit(() - { // 处理这个客户端 InputStream in client.getInputStream(); in.read(buf); // ← 问题在这里线程会卡在这里等数据 }); }线程池解决了线程数量爆炸但引入了新问题线程在等待数据时是阻塞的。1.1核心问题阻塞 IO 下线程在空等线程池200个线程1000个用户同时在线线程1等待用户1发消息... 阻塞中用户1在慢慢打字线程2等待用户2发消息... 阻塞中用户2去上厕所了线程3等待用户3发消息... 阻塞中...线程200等待用户200发消息... 阻塞中用户201来了对不起线程池满了排队等着❌200个线程全部卡在 in.read() 傻等什么都没干却占着资源不放。1.2 Netty NIO 的本质区别Netty 用的是非阻塞 IO线程不等待有数据才处理Netty 只用几个线程管理1000个连接选择器Selector同时监视1000个连接↓用户1有数据来了→ 线程处理用户1用户1处理完了 → 线程去处理用户99用户99处理完了 → 线程去处理用户400...线程一直在干活不阻塞等待 ✅线程池只解决了线程数量的问题没解决线程阻塞空等的问题。Netty 用 NIO让线程不再空等一个线程可以处理成千上万个连接这才是本质提升。1.3 BIO开销崩溃的成因单核CPU 同一时刻只能运行一个线程要让多个线程同时运行操作系统就不停地在线程之间来回切换。每次切换时必须把当前线程的运行状态完整保存再把下一个线程的状态完整恢复线程1 运行中...↓ 保存线程1的寄存器、程序计数器、栈指针...上下文CPU 切换到线程2↓ 恢复线程2的寄存器、程序计数器、栈指针...线程2 运行中...↓ 保存线程2的上下文...CPU 切换到线程3...这个保存恢复的过程就叫上下文切换每次大约耗时 1~10 微秒。线程少的时候没问题10个线程每秒切换 10次 × 10微秒 0.1毫秒CPU 几乎感觉不到线程多了就崩了假设5000个线程每个线程都要分到CPU时间每秒切换 5000次 × 10微秒 50毫秒CPU 每秒有 50毫秒在做切换这件事本身而不是在处理你的业务逻辑更严重的是5000个线程大部分都在 in.read() 阻塞等待CPU 辛苦切换到线程2 → 线程2我在等数据没事干CPU 辛苦切换到线程3 → 线程3我也在等数据没事干CPU 辛苦切换到线程4 → 线程4我也在等数据没事干...切换了5000次真正干活的只有少数线程BIO 线程池下同一时刻真正有数据处理的线程数远少于总线程数大量线程在空等导致上下文切换都是无效开销。2 基于Netty的自定义二进制协议2.1选择基于Netty 的原因第一层原因性能Netty 本身的价值IM 聊天软件的特点- 连接数多大量用户同时在线- 但大部分时间没有消息连接空闲- 需要实时推送服务端主动发消息这个特点天然适合 NIO用 Netty 可以用少量线程支撑大量连接性能更好。第二层Netty 让网络编程变简单就算不考虑高并发原生 Java NIO 也不太易用// 原生 NIO 代码极其繁琐 Selector selector Selector.open(); ServerSocketChannel serverChannel ServerSocketChannel.open(); serverChannel.configureBlocking(false); serverChannel.register(selector, SelectionKey.OP_ACCEPT); while(true) { selector.select(); SetSelectionKey keys selector.selectedKeys(); IteratorSelectionKey iter keys.iterator(); while(iter.hasNext()) { SelectionKey key iter.next(); if(key.isAcceptable()) { ... } if(key.isReadable()) { ... } iter.remove(); } }Netty 把这些封装好了你只需要// Netty 代码清晰简洁 serverBootstrap .group(parentGroup, childGroup) .channel(NioServerSocketChannel.class) .childHandler(new MyChannelInitializer());2.2处理过程自定义协议写了编解码逻辑报文格式、序列化但粘包拆包、线程调度、内存管理这些更底层更复杂的部分仍然是Netty 在背后托底站在 Netty 的肩膀上写的自定义编解码。第一步发送方编码ObjEncoder// ObjEncoder.java // 继承 Netty 的 MessageToByteEncoderNetty 负责 // 1. 在正确的线程里调用 encode() // 2. 管理 ByteBuf 的内存分配和释放 public class ObjEncoder extends MessageToByteEncoderPacket { Override protected void encode(ChannelHandlerContext ctx, Packet in, ByteBuf out) { // ① 用 Protostuff 把 Java 对象序列化成字节数组 // 例如 LoginRequest{userId001, password123} → byte[] byte[] data SerializationUtil.serialize(in); // ② 写4字节长度data长度 1字节命令字 // Netty 的 ByteBuf 帮你管理写指针不用手动计算偏移量 out.writeInt(data.length 1); // ③ 写1字节命令字标识这是什么类型的包 // LoginRequest1, MsgRequest3, ReconnectRequest14... out.writeByte(in.getCommand()); // ④ 写实际业务数据 out.writeBytes(data); // 最终报文格式 // ┌──────────┬──────────┬──────────────────┐ // │ 4字节长度 │ 1字节命令 │ N字节业务数据 │ // └──────────┴──────────┴──────────────────┘ } }第二步数据在网络上传输TCP 层Netty 托底发送方 接收方│ ││ ┌─────────────────────────┐ ││ │ Netty 帮你做的事 │ ││ │ • 把 ByteBuf 交给 TCP │ ││ │ • 处理 TCP 分片 │ ││ │ • 保证数据可靠到达 │ ││ └─────────────────────────┘ ││ ─────────── 网络 ──────────── │完全不需要写这部分代码Netty 的 NioSocketChannel 帮你处理了所有底层 TCP 操作第三步接收方解码ObjDecoder—— 粘包拆包在这里体现// ObjDecoder.java // 继承 ByteToMessageDecoder这是 Netty 解决粘包拆包的关键 // Netty 保证 // 1. 每次有新数据到达时自动调用 decode() // 2. 上次没读完的数据会自动保留下次继续 // 3. 在正确的线程里调用线程安全 public class ObjDecoder extends ByteToMessageDecoder { Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, ListObject out) { // ① 检查是否够读4字节长度字段 // 【粘包拆包场景1】数据刚到一半连长度都没收完 // 例如发送了100字节但只收到了2字节 // Netty 会保留这2字节等下次数据到来再调用 decode() if (in.readableBytes() 4) { return; // 数据不够直接返回等待 } // ② 标记当前读指针位置 // 目的如果后续数据不够可以回退到这里重新等待 in.markReaderIndex(); // ③ 读取4字节长度 int dataLength in.readInt(); // ④ 检查剩余数据是否够一个完整包体 // 【粘包拆包场景2】知道了长度但数据还没全部到达 // 例如长度字段说有1000字节但目前只收到500字节 if (in.readableBytes() dataLength) { // 重置读指针回到 markReaderIndex() 的位置 // 下次 decode() 被调用时从头再来 in.resetReaderIndex(); return; // 数据不完整等待更多数据 } // ⑤ 数据完整读取1字节命令字 // 根据命令字知道这是什么类型的包 byte command in.readByte(); // ⑥ 读取剩余业务数据总长度 - 1字节命令字 byte[] data new byte[dataLength - 1]; in.readBytes(data); // ⑦ 根据命令字找到对应的 Class 类型反序列化成 Java 对象 // Packet.get(command) 从注册表查找 // command1 → LoginRequest.class // command3 → MsgRequest.class // command14 → ReconnectRequest.class Class? extends Packet packetClass Packet.get(command); Packet packet SerializationUtil.deserialize(data, packetClass); // ⑧ 把解码好的对象加入列表 // Netty 会自动把它传递给 Pipeline 中下一个 Handler out.add(packet); // 【粘包场景3】一次收到多个包的数据 // 例如同时收到了3条消息的数据粘在一起 // Netty 的 ByteToMessageDecoder 会循环调用 decode() // 直到 in 里没有足够数据为止自动拆分出每一个包 } }第四步解码后交给业务 Handler 处理// MsgHandler.java以消息处理为例 // Netty 的 Pipeline 自动把解码好的 Packet 传递到这里 // 你完全不用关心数据是怎么来的 public class MsgHandler extends MyBizHandlerMsgRequest { Override public void channelRead(Channel channel, MsgRequest msg) { // 拿到的已经是完整的 Java 对象直接用就行 String fromUserId msg.getUserId(); // 发送者 String toUserId msg.getFriendId(); // 接收者 String content msg.getMsgText(); // 消息内容 // 找到接收者的 Channel转发消息 Channel toChannel SocketChannelUtil.getChannel(toUserId); if (toChannel ! null) { // Netty 的 writeAndFlush 会触发 ObjEncoder 编码后发出 toChannel.writeAndFlush(new MsgResponse( fromUserId, content, msg.getMsgType(), msg.getMsgDate() )); } } }2.3序列化数据长什么样 —— 报文格式设计2.3.1协议格式定义// ObjEncoder.java 中定义了报文格式 out.writeInt(data.length 1); // 4字节包体总长度 out.writeByte(in.getCommand()); // 1字节命令字 out.writeBytes(data); // N字节业务数据 // 最终格式 // ┌───────────────┬─────────────┬──────────────────────┐ // │ 4字节 Length │ 1字节 Command│ N字节 Data │ // └───────────────┴─────────────┴──────────────────────┘2.3.2命令字如何定义// Command.java —— 用接口常量定义所有命令字 public interface Command { Byte LoginRequest 1; // 登录请求 Byte LoginResponse 2; // 登录响应 Byte MsgRequest 3; // 发消息 Byte MsgResponse 4; // 收消息 Byte TalkNoticeRequest 5; Byte TalkNoticeResponse 6; Byte SearchFriendRequest 7; Byte SearchFriendResponse 8; Byte AddFriendRequest 9; Byte AddFriendResponse 10; Byte DelTalkRequest 11; Byte MsgGroupRequest 12; Byte MsgGroupResponse 13; Byte ReconnectRequest 14; // 断线重连 Byte RegisterRequest 15; Byte RegisterResponse 16; }2.3.3Packet 基类 —— 命令字注册表// Packet.java public abstract class Packet { // 命令字 → Class 的注册表 // 解码时根据命令字找到对应的类进行反序列化 private static final MapByte, Class? extends Packet packetType new ConcurrentHashMap(); static { packetType.put(Command.LoginRequest, LoginRequest.class); packetType.put(Command.LoginResponse, LoginResponse.class); packetType.put(Command.MsgRequest, MsgRequest.class); packetType.put(Command.MsgResponse, MsgResponse.class); packetType.put(Command.ReconnectRequest, ReconnectRequest.class); // ...其余命令字 } // 解码器调用这个方法根据命令字拿到对应 Class public static Class? extends Packet get(Byte command) { return packetType.get(command); } // 每个子类必须实现返回自己的命令字 public abstract Byte getCommand(); }2.3.4具体包的字段定义// MsgRequest.java —— 定义发消息包长什么样 public class MsgRequest extends Packet { private String userId; // 发送者ID private String friendId; // 接收者ID private String msgText; // 消息内容 private Integer msgType; // 消息类型0文字 1表情 private Date msgDate; // 发送时间 Override public Byte getCommand() { return Command.MsgRequest; // 命令字 3 } } // LoginRequest.java —— 定义登录包长什么样 public class LoginRequest extends Packet { private String userId; private String userPassword; Override public Byte getCommand() { return Command.LoginRequest; // 命令字 1 } } // ReconnectRequest.java —— 定义重连包长什么样 public class ReconnectRequest extends Packet { private String userId; // 携带userId告诉服务端我是谁 Override public Byte getCommand() { return Command.ReconnectRequest; // 命令字 14 } }2.3.5序列化// SerializationUtil.java public class SerializationUtil { // ① Schema 缓存避免重复创建提升性能 // Schema 是 Protostuff 的序列化模板描述一个类如何序列化 private static MapClass?, Schema? cachedSchema new ConcurrentHashMap(); // ② 用于创建对象实例不调用构造方法直接分配内存 private static Objenesis objenesis new ObjenesisStd(); // ③ 序列化Java对象 → byte[] public static T byte[] serialize(T obj) { ClassT cls (ClassT) obj.getClass(); // 分配缓冲区 LinkedBuffer buffer LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE); try { // 获取或创建该类的序列化模板 SchemaT schema getSchema(cls); // Protostuff 执行序列化把对象所有字段写入字节数组 return ProtostuffIOUtil.toByteArray(obj, schema, buffer); } finally { buffer.clear(); // 清空缓冲区避免内存泄漏 } } // ④ 反序列化byte[] → Java对象 public static T T deserialize(byte[] data, ClassT cls) { // 不走构造方法直接创建空对象更快 T message objenesis.newInstance(cls); // 获取该类的序列化模板 SchemaT schema getSchema(cls); // Protostuff 把字节数组的数据填充到对象字段里 ProtostuffIOUtil.mergeFrom(data, message, schema); return message; } // ⑤ 获取 Schema优先从缓存取 private static T SchemaT getSchema(ClassT cls) { SchemaT schema (SchemaT) cachedSchema.get(cls); if (schema null) { // 第一次使用该类时创建之后复用 schema RuntimeSchema.createFrom(cls); cachedSchema.put(cls, schema); } return schema; } }MsgRequest 对象userId 001friendId 002msgText 你好msgType 0msgDate 2026-04-06↓ SerializationUtil.serialize()↓ Protostuff 把所有字段压缩成二进制byte[] data [0x0A, 0x03, 0x30, 0x30, 0x31, ...] // 紧凑的二进制↓ ObjEncoder 拼装报文最终字节流[0x00, 0x00, 0x00, 0x1F] // 4字节长度[0x03] // 1字节命令字MsgRequest3[0x0A, 0x03, 0x30, ...] // N字节 Protostuff序列化数据2.3.5.1为什么用 Protostuff 而不用 JSON同一个 MsgRequest 对象JSON 序列化{userId:001,friendId:002,msgText:你好,msgType:0}→ 约 60 字节字段名也要传输冗余大Protostuff 序列化→ 约 15 字节只传字段值按字段编号区分极度紧凑2.3.5.2为什么要创建需要反射扫描的字段先理解反射是什么正常情况下你操作一个对象是这样的// 编译时就知道 MsgRequest 有哪些字段直接访问MsgRequest msg new MsgRequest();msg.userId 001; // 直接赋值msg.msgText 你好;反射是另一种方式在运行时动态获取类的信息// 不提前知道类的结构运行时动态获取Class? cls MsgRequest.class;// 获取这个类所有字段Field[] fields cls.getDeclaredFields();for (Field field : fields) {System.out.println(field.getName()); // 打印字段名System.out.println(field.getType()); // 打印字段类型}// 输出// userId → String// friendId → String// msgText → String// msgType → Integer// msgDate → Date---Protostuff 创建 Schema 的过程SchemaMsgRequest schema RuntimeSchema.createFrom(MsgRequest.class);// 这行代码背后发生了什么//// 第一步反射获取 MsgRequest 所有字段// Field[0] userId (String)// Field[1] friendId (String)// Field[2] msgText (String)// Field[3] msgType (Integer)// Field[4] msgDate (Date)//// 第二步分析每个字段的序列化方式// String → 怎么转成字节// Integer → 怎么转成字节// Date → 怎么转成字节//// 第三步建立字段编号映射// 1号 → userId// 2号 → friendId// 3号 → msgText// ...//// 把以上扫描结果保存成 Schema 对象---为什么反射慢// 直接访问字段编译时确定极快msg.userId 001;// CPU执行直接写内存地址1纳秒级别// 反射访问字段运行时查找慢field.set(msg, 001);// CPU执行// 1. 查找字段名称// 2. 检查访问权限// 3. 找到内存地址// 4. 写入值// 比直接访问慢几十倍缓存的价值不缓存每次序列化都反射扫描发送消息 → 反射扫描5个字段 → 建立映射 → 序列化 → 发送发送消息 → 反射扫描5个字段 → 建立映射 → 序列化 → 发送发送消息 → 反射扫描5个字段 → 建立映射 → 序列化 → 发送...重复扫描缓存只有第一次扫描发送消息 → 反射扫描5个字段 → 建立映射 → 存入缓存 → 序列化 → 发送发送消息 → 从缓存取映射 → 序列化 → 发送发送消息 → 从缓存取映射 → 序列化 → 发送一句话反射是 Java 在运行时动态读取类结构的机制Protostuff用反射扫描类有哪些字段、每个字段是什么类型这个扫描过程比较耗时所以把扫描结果缓存起来下次直接用不用重复扫描。3 断线重连机制“心跳检测”定时巡检 应用层重连包方案每隔5秒检查一次 Channel 是否还活着→ 活着什么都不做→ 死了重新建立连接发送重连包告诉服务端我回来了3.1 第一步客户端启动连接ClientApplication.java// ClientApplication.java public class ClientApplication extends Application { // 单线程池专门负责建立网络连接 private static ExecutorService executorService Executors.newFixedThreadPool(1); // 定时任务线程池专门负责定时巡检 private static ScheduledExecutorService scheduledExecutorService Executors.newScheduledThreadPool(1); Override public void start(Stage primaryStage) throws Exception { // ① 第一次启动建立连接 // submit() 把连接任务提交给线程池 // .get() 阻塞等待直到连接建立成功才继续 Channel channel executorService .submit(nettyClient) // 执行 NettyClient.call() .get(); // 等待连接完成 // ② 启动定时巡检任务 scheduledExecutorService.scheduleAtFixedRate( () - { // 巡检逻辑见第二步 }, 3, // 延迟3秒后开始第一次巡检 5, // 之后每隔5秒巡检一次 TimeUnit.SECONDS ); } }第二步定时巡检逻辑ClientApplication.javascheduledExecutorService.scheduleAtFixedRate(() - { // 用 while 而不是 if // 万一重连一次没成功继续尝试直到成功为止 while (!nettyClient.isActive()) { System.out.println(通信管道巡检通信管道状态 nettyClient.isActive()); // 打印当前状态false try { System.out.println(通信管道巡检断线重连[Begin]); // ① 重新建立 TCP 连接 // 和启动时一样提交连接任务阻塞等待建立成功 Channel freshChannel executorService .submit(nettyClient) .get(); // ② 检查是否已登录有 userId 才需要重连 // 如果还没登录过不需要发重连包continue 重新检查 if (null CacheUtil.userId) continue; // ③ 向服务端发送重连包 // 告诉服务端userId 对应的用户重连了更新 Channel 映射 freshChannel.writeAndFlush( new ReconnectRequest(CacheUtil.userId) ); } catch (InterruptedException | ExecutionException e) { System.out.println(通信管道巡检断线重连[Error]); } } // isActive() true 时while 不进入巡检结束 }, 3, 5, TimeUnit.SECONDS);第三步NettyClient 建立连接NettyClient.java// NettyClient.java // 实现了 CallableChannel可以被线程池 submit() public class NettyClient implements CallableChannel { private final String inetHost 127.0.0.1; // 服务端地址 private final Integer inetPort 7397; // 服务端端口 private Channel channel; // 当前连接的 Channel Override public Channel call() throws Exception { try { Bootstrap b new Bootstrap(); b.group(workerGroup) .channel(NioSocketChannel.class) .option(ChannelOption.AUTO_READ, true) // 自动读取数据 .handler(new MyChannelInitializer(uiService)); // 发起连接syncUninterruptibly() 阻塞直到连接成功 ChannelFuture channelFuture b .connect(inetHost, inetPort) .syncUninterruptibly(); // 保存新建立的 Channel this.channel channelFuture.channel(); // 存入全局 Bean 容器其他地方可以直接取用 BeanUtil.addBean(channel, channel); } catch (Exception e) { logger.error(socket client start error, e.getMessage()); } return channel; } // 巡检时调用这个方法判断连接是否存活 public boolean isActive() { return channel.isActive(); // channel.isActive() 返回 false 的情况 // • 网络断开 // • 服务端主动关闭连接 // • 长时间无数据导致 TCP 超时 } }第四步服务端收到重连包ReconnectHandler.java// ReconnectHandler.java public class ReconnectHandler extends MyBizHandlerReconnectRequest { Override public void channelRead(Channel channel, ReconnectRequest msg) { logger.info(客户端断线重连处理。userId{}, msg.getUserId()); // ① 删除旧的 Channel 映射 // 断线前的旧 Channel 已经失效必须先清除 // 否则服务端还会往旧 Channel 发消息发送失败 SocketChannelUtil.removeUserChannelByUserId(msg.getUserId()); // ② 注册新的 Channel 映射 // 把 userId 和新建立的 Channel 绑定起来 // 之后发给这个用户的消息都通过新 Channel 发送 SocketChannelUtil.addChannel(msg.getUserId(), channel); // ③ 重新加入所有群组 // 断线重连后群组的 ChannelGroup 里没有新 Channel // 需要重新注册否则收不到群消息 ListString groupsIdList userService .queryTalkBoxGroupsIdList(msg.getUserId()); for (String groupsId : groupsIdList) { SocketChannelUtil.addChannelGroup(groupsId, channel); } } }第五步服务端 Channel 生命周期管理MyBizHandler.java// MyBizHandler.java // 所有 Handler 的基类处理连接的生命周期事件 public abstract class MyBizHandlerT extends SimpleChannelInboundHandlerT { // 连接建立时触发 Override public void channelActive(ChannelHandlerContext ctx) throws Exception { super.channelActive(ctx); logger.info(客户端连接通知{}, ctx.channel()); // 只做日志记录Channel 注册在 LoginHandler 里做 } // 连接断开时触发正常断开 Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { super.channelInactive(ctx); // 从用户 Channel 映射中删除 // 防止服务端继续往已断开的 Channel 发消息 SocketChannelUtil.removeChannel( ctx.channel().id().toString() ); // 从所有群组的 ChannelGroup 中删除 // 防止群消息广播时发给已断开的 Channel SocketChannelUtil.removeChannelGroupByChannel(ctx.channel()); } // 发生异常时触发异常断开 Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { logger.error(服务端异常断开, cause); // 和 channelInactive 做同样的清理工作 // 保证无论正常断开还是异常断开Channel 都能被清理 SocketChannelUtil.removeChannel( ctx.channel().id().toString() ); SocketChannelUtil.removeChannelGroupByChannel(ctx.channel()); } }第六步Channel 注册表SocketChannelUtil.java// SocketChannelUtil.java public class SocketChannelUtil { // userId → Channel 映射发单聊消息用 private static MapString, Channel userChannel new ConcurrentHashMap(); // channelId → userId 映射断线清理时用 private static MapString, String userChannelId new ConcurrentHashMap(); // groupId → ChannelGroup 映射发群消息用 private static MapString, ChannelGroup channelGroupMap new ConcurrentHashMap(); // 注册新 Channel public static void addChannel(String userId, Channel channel) { userChannel.put(userId, channel); userChannelId.put(channel.id().toString(), userId); } // 通过 channelId 删除断线时用只知道 channelId public static void removeChannel(String channelId) { String userId userChannelId.get(channelId); if (null userId) return; userChannel.remove(userId); userChannelId.remove(channelId); } // 通过 userId 删除重连时用知道 userId public static void removeUserChannelByUserId(String userId) { userChannel.remove(userId); } // 获取用户 Channel发消息时用 public static Channel getChannel(String userId) { return userChannel.get(userId); } // 注册群组 Channel public static synchronized void addChannelGroup( String talkId, Channel channel) { ChannelGroup channelGroup channelGroupMap.get(talkId); if (null channelGroup) { // 第一次创建这个群组的 ChannelGroup channelGroup new DefaultChannelGroup( GlobalEventExecutor.INSTANCE ); channelGroupMap.put(talkId, channelGroup); } channelGroup.add(channel); } // 断线时从所有群组移除该 Channel public static void removeChannelGroupByChannel(Channel channel) { for (ChannelGroup group : channelGroupMap.values()) { group.remove(channel); // 不在这个群组里也没关系remove 自动忽略 } } }正常连接阶段─────────────────────────────────────────────客户端启动→ NettyClient.call() 建立 TCP 连接→ 登录成功CacheUtil.userId 001→ SocketChannelUtil.addChannel(001, channel)每5秒巡检→ isActive() true → 什么都不做 ✅断线阶段─────────────────────────────────────────────网络断开→ MyBizHandler.channelInactive() 触发→ SocketChannelUtil.removeChannel() 服务端清理→ isActive() false 客户端检测到重连阶段─────────────────────────────────────────────巡检线程检测到 isActive() false→ while 循环进入→ NettyClient.call() 重新建立 TCP 连接→ freshChannel.writeAndFlush(new ReconnectRequest(001))服务端收到重连包→ ReconnectHandler.channelRead()→ removeUserChannelByUserId(001) 删旧映射→ addChannel(001, newChannel) 建新映射→ 重新加入所有群组重连成功→ isActive() true→ while 循环退出 ✅→ 恢复正常收发消息---即客户端用定时任务每5秒检查 Channel 是否存活断线则重新建立 TCP 连接并发送 ReconnectRequest 包服务端收到后更新Channel 注册表重新绑定用户和新 Channel 的关系保证消息能继续正常收发。三句话讲清楚原理第一句客户端每隔5秒检查一次网络连接是否还活着第二句发现断了就立刻重新连接服务器第三句连上之后发一个重连包告诉服务器我回来了服务器收到后更新我的连接记录断线重连分客户端和服务端两部分。▎ 客户端启动一个定时任务每隔5秒检查一次Channel是否存活。Channel可以理解为客户端和服务器之间的通信管道。如果管道断了就立刻重新连接服务器连上之后发送一个ReconnectRequest包把自己的userId带过去。▎ 服务端收到ReconnectRequest后先把用户旧的管道记录删掉再把新管道和这个用户绑定起来同时把用户重新加入他所在的群组这样群消息也能正常收到。▎ 另外服务端还监听了连接断开的事件无论是正常断开还是异常断开都会自动清理掉失效的管道记录防止往已经断开的连接发消息。问为什么用定时任务检测不用心跳包▎ 心跳包需要客户端定时发、服务端定时收实现更复杂。这个项目直接检查Channel状态更简单直接适合学习项目。问Channel是什么▎ 就是客户端和服务端之间的网络连接通道可以理解为一根管子数据从这里流过去。isActive()就是检查这根管子是否还通着。问为什么重连后要重新加入群组▎ 服务端用ChannelGroup管理群成员断线后旧Channel已经从群组里移除了新Channel进来必须重新注册不然收不到群消息。断线↓客户端每5秒发现管道不通↓重新建立连接新管道↓发送ReconnectRequest(我的userId)↓服务端删旧管道 → 存新管道 → 重新加群组↓恢复正常