Spring Boot 3 + Netty 构建高并发即时通讯服务

// 1. WebSocket连接全局配置
globalData: {
    // WebSocket服务器连接地址
    chatServerUrl: "ws://127.0.0.1:875/ws",

    // 全局WebSocket连接对象
    CHAT: null,

    // 标记WebSocket连接状态
    chatSocketOpen: false,
},
// 2. 应用启动时初始化WebSocket连接
onLaunch: function() {
    // 程序启动时连接聊天服务器
    this.doConnect(false);
},
// 3. 核心方法：建立WebSocket连接
doConnect(isFirst) {
    // 重连时显示提示
    if (isFirst) {
        uni.showToast({
            icon: "loading",
            title: "断线重连中...",
            duration: 2000
        });
    }

    var me = this;
    // 仅当用户已登录时才连接WebSocket
    if (me.getUserInfoSession() != null && me.getUserInfoSession() != "" && me.getUserInfoSession() != undefined) {
        // 创建WebSocket连接
        me.globalData.CHAT = uni.connectSocket({
            url: me.globalData.chatServerUrl,
            complete: ()=> {}
        });

        // 4. 连接成功事件处理
        me.globalData.CHAT.onOpen(function(){
            // 更新连接状态标记
            me.globalData.chatSocketOpen = true;
            console.log("ws连接已打开，socketOpen = " + me.globalData.chatSocketOpen);

            // 构建初始化消息(消息类型0表示连接初始化)
            var chatMsg = {
                senderId: me.getUserInfoSession().id,
                msgType: 0
            } 
            var dataContent = { 
                chatMsg: chatMsg
            }
            var msgPending = JSON.stringify(dataContent);

            // 发送初始化消息，通知服务器用户身份
            me.globalData.CHAT.send({
                data: msgPending
            }); 
        });

        // 5. 连接关闭事件处理
        me.globalData.CHAT.onClose(function(){
            me.globalData.chatSocketOpen = false;
            console.log("ws连接已关闭，socketOpen = " + me.globalData.chatSocketOpen);
        }); 

        // 6. 接收消息事件处理
        me.globalData.CHAT.onMessage(function(res){
            console.log('App.vue 收到服务器内容：' + res.data);

            // 处理接收到的消息
            me.dealReceiveLastestMsg(JSON.parse(res.data));
        });

        // 7. 连接错误事件处理
        me.globalData.CHAT.onError(function(){
            me.globalData.chatSocketOpen = false;
            console.log('WebSocket连接打开失败，请检查！');
        });
    }
},
// 8. 发送WebSocket消息的通用方法
sendSocketMessage(msg) {
    // 检查连接状态，只有在连接开启时才发送
    if (this.globalData.chatSocketOpen) {
        uni.sendSocketMessage({
            data: msg
        });
    } else {
        uni.showToast({
            icon: "none",
            title: "您已断开聊天服务器的连接"
        })
    }
},
// 9. 处理接收到的消息
dealReceiveLastestMsg(msgJSON) {
    console.log(msgJSON);
    var chatMsg = msgJSON.chatMsg;
    var chatTime = msgJSON.chatTime;
    var senderId = chatMsg.senderId;

    var receiverType = chatMsg.receiverType;
    console.log('chatMsg.receiverType = ' + receiverType);

    var me = this;
    // 获取发送者的用户信息
    var userId = me.getUserInfoSession().id;
    var userToken = me.getUserSessionToken();
    var serverUrl = me.globalData.serverUrl;
    // 请求发送者详细信息
    uni.request({
        method: "POST",
        header: {
            headerUserId: userId,
            headerUserToken: userToken
        },
        url: serverUrl + "/userinfo/get?userId=" + senderId,
        success(result) {
            if (result.data.status == 200) {
                var currentSourceUserInfo = result.data.data;
                me.currentSourceUserInfo = currentSourceUserInfo;

                // 根据消息类型设置显示内容
                var msgShow = chatMsg.msg;
                if (chatMsg.msgType == 2) {
                    msgShow = "[图片]"
                } elseif (chatMsg.msgType == 4) {
                    msgShow = "[视频]"
                } elseif (chatMsg.msgType == 3) {
                    msgShow = "[语音]"
                } 
                // 保存最新消息到本地存储
                me.saveLastestMsgToLocal(senderId, currentSourceUserInfo, msgShow, chatTime, msgJSON);
            }
        }
    })
},
// 10. 将最新消息保存到本地存储
saveLastestMsgToLocal(sourceUserId, sourceUser, msgContent, chatTime, msgJSON) {
    // 构造最新消息对象
    var lastMsg = {
        sourceUserId: sourceUserId,     // 源头用户，聊天对象
        name: sourceUser.nickname,
        face: sourceUser.face,
        msgContent: msgContent,
        chatTime: chatTime,
        unReadCounts: 0,
        communicationType: 1,   // 1:单聊，2:群聊
    }

    // 获取本地存储的聊天列表
    var lastestUserChatList = uni.getStorageSync("lastestUserChatList");
    if (lastestUserChatList == null || lastestUserChatList == undefined || lastestUserChatList == "") {
        lastestUserChatList = [];
    }

    // 更新或新增消息记录
    var dealMsg = false;
    for (var i = 0; i < lastestUserChatList.length; i++) {
        var tmp = lastestUserChatList[i];
        if (tmp.sourceUserId == lastMsg.sourceUserId) {
            // 已存在聊天记录，更新最新消息
            lastestUserChatList.splice(i, 1, lastMsg);
            dealMsg = true;
            break;
        }
    }
    if (!dealMsg) {
        // 新的聊天对象，添加到列表开头
        lastestUserChatList.unshift(lastMsg);
    }
    // 保存更新后的聊天列表
    uni.setStorageSync("lastestUserChatList", lastestUserChatList);
    // 通知UI更新
    uni.$emit('reRenderReceiveMsgInMsgVue', "domeafavor");
    uni.$emit('receiveMsgInMsgListVue', msgJSON);
},
// 11. 关闭WebSocket连接
closeWSConnect() {
    this.globalData.CHAT.close();
}

publicclassUserChannelSession {
privatestatic Map<String, List<Channel>> multiSession = newHashMap<>();
privatestatic Map<String, String> userChannelIdRelation = newHashMap<>();
publicstaticvoidputUserChannelIdRelation(String userId, String channelId) {
        userChannelIdRelation.put(channelId, userId);
    }
publicstatic String getUserIdByChannelId(String channelId) {
return userChannelIdRelation.get(channelId);
    }
publicstaticvoidputMultiChannels(String userId, Channel channel) {
        List<Channel> channels = getMultiChannels(userId);
if (channels == null || channels.size() == 0) {
            channels = newArrayList<>();
        }
        channels.add(channel);
        multiSession.put(userId, channels);
    }
publicstaticvoidremoveUserChannels(String userId, String channelId) {
        List<Channel> channels = getMultiChannels(userId);
if (channels == null || channels.size() == 0) {
return;
        }
for (Channel channel : channels) {
if (channel.id().asLongText().equals(channelId)) {
                channels.remove(channel);
            }
        }
        multiSession.put(userId, channels);
    }
publicstatic List<Channel> getMultiChannels(String userId) {
return multiSession.get(userId);
    }
}

publicclassChatHandlerextendsSimpleChannelInboundHandler<TextWebSocketFrame> {
publicstaticChannelGroupclients=newDefaultChannelGroup(GlobalEventExecutor.INSTANCE);
@Override
protectedvoidchannelRead0(ChannelHandlerContext channelHandlerContext,
                                TextWebSocketFrame textWebSocketFrame) throws Exception {
Stringcontent= textWebSocketFrame.text();
DataContentdataContent= JsonUtils.jsonToPojo(content, DataContent.class);
ChatMsgchatMsg= dataContent.getChatMsg();
StringmsgText= chatMsg.getMsg();
StringreceiverId= chatMsg.getReceiverId();
StringsenderId= chatMsg.getSenderId();
        chatMsg.setChatTime(LocalDateTime.now());
IntegermsgType= chatMsg.getMsgType();
ChannelcurrentChannel= channelHandlerContext.channel();
StringcurrentChannelId= currentChannel.id().asLongText();
if (msgType == MsgTypeEnum.CONNECT_INIT.type) {
            UserChannelSession.putMultiChannels(senderId, currentChannel);
            UserChannelSession.putUserChannelIdRelation(currentChannelId, senderId);
        } elseif (msgType == MsgTypeEnum.WORDS.type) {
            List<Channel> receiverChannels = UserChannelSession.getMultiChannels(receiverId);
if (receiverChannels == null || receiverChannels.size() == 0 || receiverChannels.isEmpty()) {
                chatMsg.setIsReceiverOnLine(false);
            } else {
                chatMsg.setIsReceiverOnLine(true);
for (Channel receiverChannel : receiverChannels) {
ChannelfindChannel= clients.find(receiverChannel.id());
if (findChannel != null) {
                        dataContent.setChatMsg(chatMsg);
StringchatTimeFormat=
                                LocalDateUtils.format(chatMsg.getChatTime(), LocalDateUtils.DATETIME_PATTERN_2);
                        dataContent.setChatTime(chatTimeFormat);
                        findChannel.writeAndFlush(
newTextWebSocketFrame(
                                        JsonUtils.objectToJson(dataContent)));
                    }
                }
            }
        }
        currentChannel.writeAndFlush(newTextWebSocketFrame(currentChannelId));
    }
@Override
publicvoidhandlerAdded(ChannelHandlerContext ctx)throws Exception {
ChannelcurrentChannel= ctx.channel();
        clients.add(currentChannel);
    }
@Override
publicvoidhandlerRemoved(ChannelHandlerContext ctx)throws Exception {
ChannelcurrentChannel= ctx.channel();
StringuserId= UserChannelSession.getUserIdByChannelId(currentChannel.id().asLongText());
        UserChannelSession.removeUserChannels(userId, currentChannel.id().asLongText());
        clients.remove(currentChannel);
    }
@Override
publicvoidexceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {
Channelchannel= ctx.channel();
        channel.close();
        clients.remove(channel);
StringuserId= UserChannelSession.getUserIdByChannelId(channel.id().asLongText());
        UserChannelSession.removeUserChannels(userId, channel.id().asLongText());
    }
}