上一篇【第48篇】Netty整合Spring——生产级Netty服务的最佳实践下一篇【第50篇】Netty MQTT实战——开发物联网消息服务器一、IM系统架构----- ----- ----- ------------------ | 客户端 | | 客户端 | | 客户端 | | IM接入服务器 | ----- ----- ----- ------------------ | | | | -------------------------------- | ------------------ | 消息路由层 | ------------------ | -------------------------- | | ------------- ------------- | 用户管理 | | 消息存储 | ------------- -------------二、核心Handler实现ChannelHandler.SharablepublicclassIMServerHandlerextendsChannelInboundHandlerAdapter{// userId → Channel映射privatestaticfinalConcurrentHashMapString,ChanneluserChannelsnewConcurrentHashMap();OverridepublicvoidchannelRead(ChannelHandlerContextctx,Objectmsg){IMMessageimMsg(IMMessage)msg;switch(imMsg.getType()){caseLOGIN:handleLogin(ctx,imMsg);break;caseSINGLE_CHAT:handleSingleChat(ctx,imMsg);break;caseGROUP_CHAT:handleGroupChat(ctx,imMsg);break;caseHEARTBEAT:handleHeartbeat(ctx);break;}}privatevoidhandleLogin(ChannelHandlerContextctx,IMMessagemsg){userChannels.put(msg.getFromUser(),ctx.channel());ctx.channel().attr(AttributeKey.valueOf(userId)).set(msg.getFromUser());ctx.writeAndFlush(IMMessage.of(LOGIN_RESP,OK));System.out.println(用户登录: msg.getFromUser());}privatevoidhandleSingleChat(ChannelHandlerContextctx,IMMessagemsg){ChanneltargetuserChannels.get(msg.getToUser());if(target!nulltarget.isActive()){target.writeAndFlush(msg);}else{ctx.writeAndFlush(IMMessage.of(ERROR,用户不在线));}}OverridepublicvoidchannelInactive(ChannelHandlerContextctx){StringuserId(String)ctx.channel().attr(AttributeKey.valueOf(userId)).get();if(userId!null){userChannels.remove(userId);System.out.println(用户下线: userId);}}}三、消息协议publicclassIMMessage{publicenumType{LOGIN,LOGIN_RESP,SINGLE_CHAT,GROUP_CHAT,HEARTBEAT,ERROR}privateTypetype;privateStringfromUser;privateStringtoUser;privateStringcontent;privatelongtimestamp;// getters/setters...}// 编码器publicclassIMMessageEncoderextendsMessageToByteEncoderIMMessage{protectedvoidencode(ChannelHandlerContextctx,IMMessagemsg,ByteBufout){// type(1B) fromLen(1B) fromUser toLen(1B) toUser contentLen(4B) contentout.writeByte(msg.getType().ordinal());writeString(out,msg.getFromUser());writeString(out,msg.getToUser());out.writeLong(msg.getTimestamp());byte[]contentmsg.getContent().getBytes();out.writeInt(content.length);out.writeBytes(content);}}四、群聊广播// 群聊实现遍历channel列表广播privatestaticfinalMapString,SetChannelgroupChannelsnewConcurrentHashMap();privatevoidhandleGroupChat(ChannelHandlerContextctx,IMMessagemsg){SetChannelmembersgroupChannels.get(msg.getToUser());// toUser群IDif(members!null){for(Channelmember:members){if(member.isActive()member!ctx.channel()){member.writeAndFlush(msg);}}}}五、总结功能实现方式用户登录userId → Channel映射单聊根据userId获取Channel直接发送群聊遍历群成员Channel列表广播心跳IdleStateHandler检测下线channelInactive移除映射上一篇【第48篇】Netty整合Spring——生产级Netty服务的最佳实践下一篇【第50篇】Netty MQTT实战——开发物联网消息服务器