一、前言近期基于若依(RuoYi-Vue)后台框架自研一套站内即时聊天模块完整覆盖会话创建、会话列表、历史消息、未读标记、实时消息推送、离线消息补发、心跳保活、Token鉴权等核心能力。整体技术架构分层清晰ControllerHTTP 接口提供前端会话、历史消息、已读标记等查询操作Service核心业务逻辑会话防重复创建、消息入库、未读数统计、离线消息处理Mapper MyBatis复杂联表查询实时计算用户专属未读消息、关联用户昵称头像WebSocket长连接实时推送、JWT鉴权、心跳PING/PONG、在线用户管理、离线消息推送MySQL数据表chat_session会话表 chat_message消息表配套唯一索引、联合查询索引解决并发与查询性能问题本文完整贴出建表SQL、实体DTO、Controller、Service、MapperXML、WebSocket全套源码并附带设计思路、踩坑解决方案可直接集成到现有SpringBoot后台项目。二、整体设计思路会话唯一设计解决重复创建会话并发问题两人聊天仅存在唯一一条会话记录规则不区分发起人与接收人统一 minId小用户ID、maxId大用户ID 存储数据库对 from_user_id, to_user_id 建立联合唯一索引杜绝重复会话并发同时创建会话时捕获唯一索引冲突二次查询返回已有会话避免重复数据未读数实时计算方案放弃会话表冗余未读字段单一计数采用实时子查询统计联表 chat_message 统计当前登录用户所有未读消息支持多端登录、精准区分“发给我的未读”避免不同用户查看同一会话未读数错乱事务与解耦优化消息入库、会话更新、未读数自增放入事务WebSocket消息推送放到事务外部执行防止长连接推送阻塞数据库事务持有推送失败不影响消息持久化用户可通过会话列表轮询拉取离线消息兜底WebSocket 安全设计握手阶段校验URL路径userId URL携带JWT Token双重校验本地解析JWT密钥和后台登录体系统一校验过期、非法Token直接关闭连接单用户仅保留一条在线连接新连接自动剔除旧连接PING/PONG心跳保活处理前端断线重连场景发送消息加同步锁规避多线程并发发送报通道异常离线消息补发用户WebSocket上线时自动查询所有未读消息逐条推送推送完成批量标记已读避免重复推送。三、MySQL 数据库建表语句3.1 chat_session 聊天会话表存储两个用户一对一会话基础信息联合唯一索引保证双人仅一条会话/*Navicat Premium Data TransferSource Server : 46Source Server Type : MySQLSource Server Version : 80045Source Host : 192.168.1.46:3306Source Schema : shan_guo_devTarget Server Type : MySQLTarget Server Version : 80045File Encoding : 65001Date: 10/07/2026 09:45:30*/SET NAMES utf8mb4;SET FOREIGN_KEY_CHECKS 0;– Table structure for chat_sessionDROP TABLE IF EXISTSchat_session;CREATE TABLEchat_session(idbigint NOT NULL AUTO_INCREMENT COMMENT ‘会话主键ID’,from_user_idbigint NOT NULL COMMENT ‘小ID用户’,to_user_idbigint NOT NULL COMMENT ‘大ID用户’,last_contentvarchar(500) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL COMMENT ‘会话最后一条消息内容(列表预览)’,last_msg_timedatetime NULL DEFAULT NULL COMMENT ‘会话最后一条消息发送时间(排序依据)’,unread_countint NULL DEFAULT 0 COMMENT ‘当前会话未读消息数量(冗余字段)’,create_timedatetime NULL DEFAULT CURRENT_TIMESTAMP COMMENT ‘会话创建时间’,update_timedatetime NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT ‘会话更新时间’,PRIMARY KEY (id) USING BTREE,– 双人会话唯一约束防止重复创建会话UNIQUE INDEXuk_user_pair(from_user_id,to_user_id) USING BTREE,INDEXidx_from(from_user_id) USING BTREE,INDEXidx_to(to_user_id) USING BTREE) ENGINE InnoDB AUTO_INCREMENT 3 CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci COMMENT ‘聊天会话表’ ROW_FORMAT Dynamic;SET FOREIGN_KEY_CHECKS 1;优化说明原SQL存在重复唯一索引 uk_from_to、uk_user_pair删除重复索引仅保留 uk_user_pair减少索引占用。3.2 chat_message 聊天消息表存储每条聊天记录联合索引加速未读消息查询/*Navicat Premium Data TransferSource Server : 46Source Server Type : MySQLSource Server Version : 80045Source Host : 192.168.1.46:3306Source Schema : shan_guo_devTarget Server Type : MySQLTarget Server Version : 80045File Encoding : 65001Date: 10/07/2026 09:45:38*/SET NAMES utf8mb4;SET FOREIGN_KEY_CHECKS 0;– Table structure for chat_messageDROP TABLE IF EXISTSchat_message;CREATE TABLEchat_message(idbigint NOT NULL AUTO_INCREMENT COMMENT ‘消息主键ID’,session_idbigint NOT NULL COMMENT ‘所属会话ID’,sender_idbigint NOT NULL COMMENT ‘消息发送人用户ID’,receiver_idbigint NOT NULL COMMENT ‘消息接收人用户ID’,contentvarchar(1000) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT ‘消息内容’,msg_typechar(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT ‘1’ COMMENT ‘消息类型1-文本 2-图片’,read_statuschar(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT ‘0’ COMMENT ‘读取状态0-未读 1-已读’,create_timedatetime NULL DEFAULT CURRENT_TIMESTAMP COMMENT ‘消息发送时间’,PRIMARY KEY (id) USING BTREE,– 根据会话查询历史消息索引INDEXidx_session(session_id) USING BTREE,– 用户未读消息查询联合索引优化离线消息加载INDEXidx_receiver_read(receiver_id,read_status) USING BTREE) ENGINE InnoDB AUTO_INCREMENT 11 CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci COMMENT ‘聊天消息表’ ROW_FORMAT Dynamic;SET FOREIGN_KEY_CHECKS 1;四、实体类 DTO 完整代码4.1 ChatSession 会话实体package com.para.system.chat.domain;import lombok.Data;import java.util.Date;/**聊天会话实体/Datapublic class ChatSession {/* 会话主键/private Long id;/* 较小用户ID/private Long fromUserId;/* 较大用户ID/private Long toUserId;/* 最后一条消息内容/private String lastContent;/* 最后消息时间/private Date lastMsgTime;/* 冗余未读数/private Integer unreadCount;/* 当前用户实时未读数查询计算字段/private Integer realSelfUnread;/* 创建时间/private Date createTime;/* 更新时间 */private Date updateTime;// 联表sys_user查询字段/** 对方昵称/private String targetNickName;/* 对方头像 */private String targetAvatar;}4.2 ChatMessage 消息实体package com.para.system.chat.domain;import lombok.Data;import java.util.Date;/**聊天消息实体/Datapublic class ChatMessage {/* 消息主键/private Long id;/* 会话ID/private Long sessionId;/* 发送人ID/private Long senderId;/* 接收人ID/private Long receiverId;/* 消息内容/private String content;/* 消息类型 1文本 2图片/private String msgType;/* 读取状态 0未读 1已读/private String readStatus;/* 发送时间 */private Date createTime;}4.3 ChatMsgDTO WebSocket传输DTO前后端WebSocket交互统一载体package com.para.system.chat.domain;import lombok.Data;/**WebSocket消息传输DTO/Datapublic class ChatMsgDTO {/* 会话ID/private Long sessionId;/* 接收人ID/private Long receiverId;/* 消息内容/private String content;/* 消息类型 1文本 2图片 */private String msgType;}4.4 ChatService 顶层接口package com.para.system.chat.service;import com.para.system.chat.domain.ChatMessage;import com.para.system.chat.domain.ChatSession;import com.para.system.chat.domain.ChatMsgDTO;import java.util.List;public interface ChatService {/** * 获取或创建双人会话 */ ChatSession getOrCreateSession(Long fromUid, Long toUid); /** * 保存消息并推送 */ void saveAndPushMessage(Long senderId, ChatMsgDTO dto); /** * 查询会话历史消息 */ ListChatMessage history(Long sessionId); /** * 查询用户会话列表支持昵称搜索 */ ListChatSession sessionList(Long uid, String nickName); /** * 批量标记会话全部消息已读 */ void markAllRead(Long sessionId, Long uid); /** * 用户上线推送离线未读消息 */ void pushOfflineMessage(Long uid);}五、代码分层实现5.1 Controller 层ChatController HTTP接口提供前端所有同步查询接口创建会话、会话列表支持昵称搜索、历史消息分页、标记消息已读。统一继承项目基础控制器内置分页工具、登录用户工具类。package com.para.system.chat.controller;import com.para.common.core.controller.BaseController;import com.para.common.core.domain.AjaxResult;import com.para.common.core.page.TableDataInfo;import com.para.common.utils.SecurityUtils;import com.para.system.chat.domain.ChatSession;import com.para.system.chat.service.ChatService;import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;import java.util.List;/**聊天接口控制器提供前端HTTP接口会话管理、消息查询、已读标记*/RestControllerRequestMapping(“/chat”)public class ChatController extends BaseController {Resourceprivate ChatService chatService;/**获取或创建与对方的会话param toUserId 对方用户ID*/GetMapping(“/session”)public AjaxResult session(RequestParam Long toUserId) {try {Long userId SecurityUtils.getLoginUser().getUserId();ChatSession session chatService.getOrCreateSession(userId, toUserId);return AjaxResult.success(session);} catch (RuntimeException e) {return AjaxResult.error(e.getMessage());}}/**获取当前用户的会话列表按最后消息时间倒序支持对方昵称模糊搜索*/GetMapping(“/sessionList”)public TableDataInfo sessionList(RequestParam(required false) String nickName) {startPage();Long userId SecurityUtils.getLoginUser().getUserId();List list chatService.sessionList(userId, nickName);return getDataTable(list);}/**获取会话历史消息分页*/GetMapping(“/history”)public TableDataInfo history(RequestParam Long sessionId) {startPage();return getDataTable(chatService.history(sessionId));}/**标记会话所有消息为已读清空未读计数*/PostMapping(“/read”)public AjaxResult read(RequestParam Long sessionId) {Long userId SecurityUtils.getLoginUser().getUserId();chatService.markAllRead(sessionId, userId);return AjaxResult.success();}}5.2 Service 业务实现层 ChatServiceImpl核心业务处理包含会话防并发创建、消息事务入库、WebSocket推送、离线消息补发、批量已读标记。package com.para.system.chat.service.impl;import com.para.common.utils.SecurityUtils;import com.para.system.chat.domain.ChatMessage;import com.para.system.chat.domain.ChatSession;import com.para.system.chat.domain.ChatMsgDTO;import com.para.system.chat.mapper.ChatMessageMapper;import com.para.system.chat.mapper.ChatSessionMapper;import com.para.system.chat.service.ChatService;import com.para.system.chat.websocket.ChatWebSocketServer;import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;import java.util.Date;import java.util.List;import java.util.Set;import java.util.stream.Collectors;/**聊天业务实现类功能会话管理、消息存储、未读计数、消息推送、离线消息补发*/Slf4jServicepublic class ChatServiceImpl implements ChatService {Resourceprivate ChatSessionMapper sessionMapper;Resourceprivate ChatMessageMapper msgMapper;/**获取或创建两个用户之间的聊天会话防并发重复创建param fromUid 当前登录用户IDparam toUid 对方用户IDreturn 会话对象*/Overridepublic ChatSession getOrCreateSession(Long fromUid, Long toUid) {// 禁止自己和自己聊天if (fromUid.equals(toUid)) {throw new RuntimeException(“不允许与自己创建聊天会话”);}// 统一大小ID存储保证双人会话唯一Long minId Math.min(fromUid, toUid);Long maxId Math.max(fromUid, toUid);Long loginUid SecurityUtils.getLoginUser().getUserId();// 查询已有会话ChatSession session sessionMapper.selectByUserPair(minId, maxId, loginUid);if (session null) {try {session new ChatSession();session.setFromUserId(minId);session.setToUserId(maxId);sessionMapper.insertSession(session);} catch (Exception e) {// 唯一索引冲突其他线程已创建日志告警不抛出异常log.warn(“会话创建冲突重新查询会话记录”);}// 冲突后二次查询返回最新会话return sessionMapper.selectByUserPair(minId, maxId, loginUid);}return session;}/**保存消息并异步推送给接收方数据库操作在事务内WebSocket推送在事务外避免长连接阻塞事务*/Overridepublic void saveAndPushMessage(Long senderId, ChatMsgDTO dto) {Long receiverId dto.getReceiverId();// 事务入库、更新会话、未读数1saveMessageInTransaction(senderId, dto);// 事务外推送消息推送失败不影响消息存储try {ChatWebSocketServer.sendToUser(receiverId, dto);} catch (Exception e) {log.error(“WebSocket推送消息失败发送人:{},接收人:{}”, senderId, receiverId, e);}}/**事务方法保存消息、更新会话最后消息、未读数自增*/Transactional(rollbackFor Exception.class)public void saveMessageInTransaction(Long senderId, ChatMsgDTO dto) {Long sessionId dto.getSessionId();Long receiverId dto.getReceiverId();Date now new Date();// 1. 插入聊天消息默认未读0ChatMessage msg new ChatMessage();msg.setSessionId(sessionId);msg.setSenderId(senderId);msg.setReceiverId(receiverId);msg.setContent(dto.getContent());msg.setMsgType(dto.getMsgType());msg.setReadStatus(“0”);msg.setCreateTime(now);msgMapper.insert(msg);// 2. 更新会话最后一条消息内容和时间sessionMapper.updateLastMsg(sessionId, dto.getContent(), now);// 3. 当前会话未读计数1sessionMapper.incrUnread(sessionId);}/**查询会话分页历史消息*/Overridepublic List history(Long sessionId) {return msgMapper.selectBySession(sessionId);}/**查询当前用户会话列表支持对方昵称模糊检索*/Overridepublic List sessionList(Long uid, String nickName) {return sessionMapper.selectUserSessions(uid, nickName);}/**批量标记会话全部消息已读清空会话未读计数*/OverrideTransactionalpublic void markAllRead(Long sessionId, Long uid) {msgMapper.batchRead(sessionId, uid);sessionMapper.clearUnread(sessionId);}/**用户WebSocket上线补发所有离线未读消息*/Overridepublic void pushOfflineMessage(Long uid) {List unreadList msgMapper.selectUnread(uid);if (unreadList null || unreadList.isEmpty()) {return;}// 逐条推送离线消息for (ChatMessage msg : unreadList) {ChatMsgDTO dto new ChatMsgDTO();dto.setSessionId(msg.getSessionId());dto.setReceiverId(msg.getReceiverId());dto.setContent(msg.getContent());dto.setMsgType(msg.getMsgType());ChatWebSocketServer.sendToUser(uid, dto);}// 推送完成批量标记已读避免重复推送try {Set sessionIds unreadList.stream().map(ChatMessage::getSessionId).collect(Collectors.toSet());for (Long sessionId : sessionIds) {msgMapper.batchRead(sessionId, uid);sessionMapper.clearUnread(sessionId);}} catch (Exception e) {log.error(“离线消息标记已读失败”, e);}}}5.3 Mapper 接口补充ChatSessionMapper.javapackage com.para.system.chat.mapper;import com.para.system.chat.domain.ChatSession;import org.apache.ibatis.annotations.Param;import java.util.List;public interface ChatSessionMapper {ChatSession selectByUserPair(Param(from) Long minId, Param(to) Long maxId, Param(uid) Long loginUid); void insertSession(ChatSession session); void updateLastMsg(Param(sid) Long sessionId, Param(content) String content, Param(time) java.util.Date time); void incrUnread(Param(sid) Long sessionId); void clearUnread(Param(sid) Long sessionId); ListChatSession selectUserSessions(Param(uid) Long uid, Param(nickName) String nickName);}ChatMessageMapper.javapackage com.para.system.chat.mapper;import com.para.system.chat.domain.ChatMessage;import org.apache.ibatis.annotations.Param;import java.util.List;public interface ChatMessageMapper {void insert(ChatMessage message); ListChatMessage selectBySession(Param(sessionId) Long sessionId); void batchRead(Param(sessionId) Long sessionId, Param(uid) Long userId); ListChatMessage selectUnread(Param(uid) Long userId);}5.4 MyBatis Mapper XML ChatSessionMapper.xml核心复杂SQL联表用户获取昵称头像、子查询实时统计未读、昵称模糊搜索、双人会话唯一查询。?xml version1.0 encodingUTF-8?!-- 会话结果映射关联用户昵称、头像、实时未读数量 -- resultMap typecom.para.system.chat.domain.ChatSession idChatSessionResult result propertyid columnid / result propertyfromUserId columnfrom_user_id / result propertytoUserId columnto_user_id / result propertylastContent columnlast_content / result propertylastMsgTime columnlast_msg_time / result propertyunreadCount columnunread_count / result propertyrealSelfUnread columnreal_self_unread/ result propertycreateTime columncreate_time / result propertyupdateTime columnupdate_time / result columnnick_name propertytargetNickName/ result columnavatar propertytargetAvatar/ /resultMap !-- 根据双方用户ID查询唯一会话关联用户信息实时未读统计 -- select idselectByUserPair resultMapChatSessionResult SELECT cs.*, su.nick_name, su.avatar, COALESCE(cm.unread_total,0) AS real_self_unread FROM chat_session cs LEFT JOIN sys_user su ON su.user_id ( CASE WHEN cs.from_user_id #{uid} THEN cs.to_user_id ELSE cs.from_user_id END ) LEFT JOIN ( SELECT session_id, COUNT(1) unread_total FROM chat_message WHERE receiver_id #{uid} AND read_status 0 GROUP BY session_id ) cm ON cs.id cm.session_id WHERE (from_user_id #{from} AND to_user_id #{to}) OR (from_user_id #{to} AND to_user_id #{from}) LIMIT 1 /select !-- 新增双人会话 -- insert idinsertSession parameterTypeChatSession INSERT INTO chat_session ( from_user_id, to_user_id, last_content, last_msg_time, unread_count, create_time, update_time ) VALUES ( #{fromUserId}, #{toUserId}, #{lastContent}, #{lastMsgTime}, #{unreadCount}, sysdate(), sysdate() ) /insert !-- 更新会话最后一条消息内容和时间 -- update idupdateLastMsg UPDATE chat_session SET last_content #{content}, last_msg_time #{time}, update_time sysdate() WHERE id #{sid} /update !-- 会话未读计数 1 -- update idincrUnread UPDATE chat_session SET unread_count unread_count 1, update_time sysdate() WHERE id #{sid} /update !-- 清空会话未读计数 -- update idclearUnread UPDATE chat_session SET unread_count 0, update_time sysdate() WHERE id #{sid} /update !-- 查询当前用户全部会话支持昵称模糊搜索按最后消息倒序 -- select idselectUserSessions resultMapChatSessionResult SELECT cs.*, su.nick_name, su.avatar, COALESCE(cm.unread_total,0) AS real_self_unread FROM chat_session cs LEFT JOIN sys_user su ON su.user_id ( CASE WHEN cs.from_user_id #{uid} THEN cs.to_user_id ELSE cs.from_user_id END ) LEFT JOIN ( SELECT session_id, COUNT(1) unread_total FROM chat_message WHERE receiver_id #{uid} AND read_status 0 GROUP BY session_id ) cm ON cs.id cm.session_id WHERE (cs.from_user_id #{uid} OR cs.to_user_id #{uid}) if testnickName ! null and nickName ! AND su.nick_name LIKE CONCAT(%, #{nickName}, %) /if ORDER BY cs.last_msg_time DESC /select5.5 ChatMessageMapper.xml补充完整 ?xml version1.0 encodingUTF-8?insert idinsert parameterTypecom.para.system.chat.domain.ChatMessage INSERT INTO chat_message (session_id, sender_id, receiver_id, content, msg_type, read_status, create_time) VALUES (#{sessionId},#{senderId},#{receiverId},#{content},#{msgType},#{readStatus},#{createTime}) /insert select idselectBySession resultTypecom.para.system.chat.domain.ChatMessage SELECT * FROM chat_message WHERE session_id #{sessionId} ORDER BY create_time ASC /select !-- 批量标记当前用户该会话所有消息为已读 -- update idbatchRead UPDATE chat_message SET read_status 1 WHERE session_id #{sessionId} AND receiver_id #{uid} AND read_status 0 /update !-- 查询用户所有未读消息离线消息 -- select idselectUnread resultTypecom.para.system.chat.domain.ChatMessage SELECT * FROM chat_message WHERE receiver_id #{uid} AND read_status 0 /select5.6 WebSocket 长连接服务 ChatWebSocketServer 实现握手鉴权、在线用户容器、心跳保活、消息收发、离线消息触发、线程安全发送。package com.para.system.chat.websocket;import cn.hutool.json.JSONUtil;import com.para.common.constant.Constants;import com.para.common.utils.spring.SpringUtils;import com.para.system.chat.domain.ChatMsgDTO;import com.para.system.chat.service.ChatService;import io.jsonwebtoken.Claims;import io.jsonwebtoken.Jwts;import lombok.extern.log4j.Log4j2;import org.springframework.core.env.Environment;import org.springframework.stereotype.Component;import javax.websocket.*;import javax.websocket.CloseReason.CloseCodes;import javax.websocket.server.PathParam;import javax.websocket.server.ServerEndpoint;import java.io.IOException;import java.net.URLDecoder;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;ComponentServerEndpoint(“/ws/chat/{userId}”)Log4j2public class ChatWebSocketServer {// 在线用户容器userId - WebSocket会话 private static final MapLong, Session ONLINE_USER_MAP new ConcurrentHashMap(); // 静态获取Spring容器Bean解决WebSocket无法直接注入Service问题 private static ChatService getChatService() { return SpringUtils.getBean(ChatService.class); } private boolean isBlank(String str) { return str null || str.trim().length() 0; } // 解析URL Query参数兼容带等号的JWT Token private String getQueryParam(Session session, String paramName) { String query session.getQueryString(); log.info(完整URL参数串: {}, query); if (isBlank(query)) { return null; } String[] paramArr query.split(); for (String item : paramArr) { int eqIndex item.indexOf(); if (eqIndex -1) continue; String key item.substring(0, eqIndex); String value item.substring(eqIndex 1); if (paramName.equals(key)) { try { String decodeVal URLDecoder.decode(value, UTF-8); log.info(解析出token值: {}, decodeVal); return decodeVal; } catch (Exception e) { log.error(URL解码token失败, e); return value; } } } return null; } /** * WebSocket握手建立连接 */ OnOpen public void onOpen(Session session, PathParam(userId) Long userId) { log.info(收到WS握手请求路径userId:{}, userId); // 新连接自动剔除旧连接保证单用户仅一条在线通道 ONLINE_USER_MAP.remove(userId); String token getQueryParam(session, token); // 无Token直接断开连接 if (isBlank(token)) { log.warn(拦截断开未携带登录凭证); closeSession(session, new CloseReason(CloseCodes.CANNOT_ACCEPT, 未携带登录凭证)); return; } // 解析JWT Token与后台登录校验逻辑统一 String loginUserKey; try { Environment env SpringUtils.getBean(Environment.class); String secret env.getProperty(token.secret); Claims claims Jwts.parser() .setSigningKey(secret) .parseClaimsJws(token) .getBody(); loginUserKey claims.get(Constants.LOGIN_USER_KEY).toString(); log.info(token解析成功登录唯一标识:{}, loginUserKey); } catch (Exception e) { log.error(拦截断开Token解析异常, e); closeSession(session, new CloseReason(CloseCodes.CANNOT_ACCEPT, Token无效或已过期)); return; } // 校验通过存入在线容器触发离线消息补发 log.info(全部校验通过建立连接路径用户ID:{}, userId); ONLINE_USER_MAP.put(userId, session); ChatService chatService getChatService(); if (chatService ! null) { try { chatService.pushOfflineMessage(userId); } catch (Exception ignored) { log.error(推送离线消息异常, ignored); } } } /** * 同步安全发送消息加锁防止并发发送通道异常 */ private void safeSendText(Session session, String text) throws IOException { synchronized (session) { if (session.isOpen()) { session.getBasicRemote().sendText(text); } } } /** * 统一关闭会话封装方法 */ private void closeSession(Session session, CloseReason reason) { try { session.close(reason); } catch (Exception e) { log.error(关闭会话异常, e); } } /** * 连接关闭回调 */ OnClose public void onClose(Session session, CloseReason reason, PathParam(userId) Long userId) { ONLINE_USER_MAP.remove(userId); log.info(连接关闭用户ID:{} 关闭状态码:{}, userId, reason.getCloseCode()); } /** * 接收前端消息业务消息 PING心跳 */ OnMessage public void onMessage(String message, PathParam(userId) Long senderId) { if (isBlank(message)) { return; } // 心跳PING回复PONG保活 if (PING.equals(message)) { try { Session session ONLINE_USER_MAP.get(senderId); if(session ! null session.isOpen()){ safeSendText(session, PONG); log.info(收到PING心跳回复PONG用户ID:{},senderId); } } catch (Exception e) { log.error(回复PONG心跳失败, e); } return; } // 忽略服务端返回的PONG包 if (PONG.equals(message)) { return; } // 解析业务消息调用Service保存并推送 try { ChatMsgDTO dto JSONUtil.toBean(message, ChatMsgDTO.class); ChatService chatService getChatService(); if (chatService ! null) { chatService.saveAndPushMessage(senderId, dto); } } catch (Exception e) { log.error(处理消息异常, e); } } /** * 通道异常回调过滤正常EOF关闭日志 */ OnError public void onError(Session session, Throwable e, PathParam(userId) Long userId) { if(e instanceof java.io.EOFException){ log.info(WS通道正常关闭用户ID:{}, userId); }else{ log.error(WS异常用户ID:{}, userId, e); } ONLINE_USER_MAP.remove(userId); } /** * 静态工具方法向指定用户推送消息 */ public static void sendToUser(Long userId, Object data) { Session session ONLINE_USER_MAP.get(userId); if (session ! null session.isOpen()) { String json JSONUtil.toJsonStr(data); try { synchronized (session) { session.getBasicRemote().sendText(json); } } catch (Exception e) { log.error(发送消息失败用户ID:{}, userId, e); } } }}六、核心难点与解决方案总结双人会话重复创建并发问题数据库建立 from_user_id, to_user_id 联合唯一索引统一小ID存from、大ID存to保证双人仅一条会话捕获插入唯一索引异常冲突后二次查询避免重复数据不同用户查看同一会话未读数错乱放弃会话表单数字段存储未读使用子查询实时统计当前登录用户未读消息每个用户查询时独立计算多端、多用户隔离准确。WebSocket无法注入Spring Bean采用 SpringUtils.getBean() 静态工具类上下文获取Service解决原生WebSocket非Spring管理无法依赖注入问题。WebSocket并发发送报通道异常消息发送方法加 synchronized(session) 同步锁多线程发送时排队执行规避 IllegalStateException。离线消息重复推送用户上线一次性拉取所有未读消息推送完成后批量分组标记消息已读并清空会话未读数避免每次重连重复补发。WebSocket安全鉴权握手阶段双重校验URL路径userId URL携带JWT Token复用后台登录JWT密钥解析非法/过期Token直接主动关闭连接杜绝非法连接。长连接阻塞数据库事务消息入库、会话更新放入事务WebSocket推送逻辑放到事务外部执行防止推送超时长时间占用数据库连接。七、扩展优化方向消息持久化缓存热门会话消息加入Redis缓存减少MySQL分页查询压力分布式WebSocket当前单机内存存储在线用户集群部署需改造Redis订阅发布实现跨服务推送消息类型扩展图片、文件、语音消息上传OSS后存储访问链接会话置顶、删除、拉黑功能会话表新增状态字段扩展消息撤回、编辑消息表增加撤回标识、原始内容字段心跳超时自动下线增加定时任务清理长期无心跳僵尸连接消息幂等前端携带消息唯一ID避免重复发送重复入库八、项目适配说明该模块基于若依RuoYi-Vue SpringBoot后台开发依赖项目内置工具类SecurityUtils获取当前登录用户IDAjaxResult / TableDataInfo统一前后端返回体SpringUtilsSpring上下文静态获取BeanJWT工具、Token密钥配置 token.secret
一、前言近期基于若依(RuoYi-Vue)后台框架自研一套站内即时聊天模块完整覆盖会话创建、会话列表、历史消息、未读标记、实时消息推送、离线消息补发、心跳保活、Token鉴权等核心能力。整体技术架构分层清晰ControllerHTTP 接口提供前端会话、历史消息、已读标记等查询操作Service核心业务逻辑会话防重复创建、消息入库、未读数统计、离线消息处理Mapper MyBatis复杂联表查询实时计算用户专属未读消息、关联用户昵称头像WebSocket长连接实时推送、JWT鉴权、心跳PING/PONG、在线用户管理、离线消息推送MySQL数据表chat_session会话表 chat_message消息表配套唯一索引、联合查询索引解决并发与查询性能问题本文完整贴出建表SQL、实体DTO、Controller、Service、MapperXML、WebSocket全套源码并附带设计思路、踩坑解决方案可直接集成到现有SpringBoot后台项目。二、整体设计思路会话唯一设计解决重复创建会话并发问题两人聊天仅存在唯一一条会话记录规则不区分发起人与接收人统一 minId小用户ID、maxId大用户ID 存储数据库对 from_user_id, to_user_id 建立联合唯一索引杜绝重复会话并发同时创建会话时捕获唯一索引冲突二次查询返回已有会话避免重复数据未读数实时计算方案放弃会话表冗余未读字段单一计数采用实时子查询统计联表 chat_message 统计当前登录用户所有未读消息支持多端登录、精准区分“发给我的未读”避免不同用户查看同一会话未读数错乱事务与解耦优化消息入库、会话更新、未读数自增放入事务WebSocket消息推送放到事务外部执行防止长连接推送阻塞数据库事务持有推送失败不影响消息持久化用户可通过会话列表轮询拉取离线消息兜底WebSocket 安全设计握手阶段校验URL路径userId URL携带JWT Token双重校验本地解析JWT密钥和后台登录体系统一校验过期、非法Token直接关闭连接单用户仅保留一条在线连接新连接自动剔除旧连接PING/PONG心跳保活处理前端断线重连场景发送消息加同步锁规避多线程并发发送报通道异常离线消息补发用户WebSocket上线时自动查询所有未读消息逐条推送推送完成批量标记已读避免重复推送。三、MySQL 数据库建表语句3.1 chat_session 聊天会话表存储两个用户一对一会话基础信息联合唯一索引保证双人仅一条会话/*Navicat Premium Data TransferSource Server : 46Source Server Type : MySQLSource Server Version : 80045Source Host : 192.168.1.46:3306Source Schema : shan_guo_devTarget Server Type : MySQLTarget Server Version : 80045File Encoding : 65001Date: 10/07/2026 09:45:30*/SET NAMES utf8mb4;SET FOREIGN_KEY_CHECKS 0;– Table structure for chat_sessionDROP TABLE IF EXISTSchat_session;CREATE TABLEchat_session(idbigint NOT NULL AUTO_INCREMENT COMMENT ‘会话主键ID’,from_user_idbigint NOT NULL COMMENT ‘小ID用户’,to_user_idbigint NOT NULL COMMENT ‘大ID用户’,last_contentvarchar(500) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL COMMENT ‘会话最后一条消息内容(列表预览)’,last_msg_timedatetime NULL DEFAULT NULL COMMENT ‘会话最后一条消息发送时间(排序依据)’,unread_countint NULL DEFAULT 0 COMMENT ‘当前会话未读消息数量(冗余字段)’,create_timedatetime NULL DEFAULT CURRENT_TIMESTAMP COMMENT ‘会话创建时间’,update_timedatetime NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT ‘会话更新时间’,PRIMARY KEY (id) USING BTREE,– 双人会话唯一约束防止重复创建会话UNIQUE INDEXuk_user_pair(from_user_id,to_user_id) USING BTREE,INDEXidx_from(from_user_id) USING BTREE,INDEXidx_to(to_user_id) USING BTREE) ENGINE InnoDB AUTO_INCREMENT 3 CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci COMMENT ‘聊天会话表’ ROW_FORMAT Dynamic;SET FOREIGN_KEY_CHECKS 1;优化说明原SQL存在重复唯一索引 uk_from_to、uk_user_pair删除重复索引仅保留 uk_user_pair减少索引占用。3.2 chat_message 聊天消息表存储每条聊天记录联合索引加速未读消息查询/*Navicat Premium Data TransferSource Server : 46Source Server Type : MySQLSource Server Version : 80045Source Host : 192.168.1.46:3306Source Schema : shan_guo_devTarget Server Type : MySQLTarget Server Version : 80045File Encoding : 65001Date: 10/07/2026 09:45:38*/SET NAMES utf8mb4;SET FOREIGN_KEY_CHECKS 0;– Table structure for chat_messageDROP TABLE IF EXISTSchat_message;CREATE TABLEchat_message(idbigint NOT NULL AUTO_INCREMENT COMMENT ‘消息主键ID’,session_idbigint NOT NULL COMMENT ‘所属会话ID’,sender_idbigint NOT NULL COMMENT ‘消息发送人用户ID’,receiver_idbigint NOT NULL COMMENT ‘消息接收人用户ID’,contentvarchar(1000) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT ‘消息内容’,msg_typechar(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT ‘1’ COMMENT ‘消息类型1-文本 2-图片’,read_statuschar(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT ‘0’ COMMENT ‘读取状态0-未读 1-已读’,create_timedatetime NULL DEFAULT CURRENT_TIMESTAMP COMMENT ‘消息发送时间’,PRIMARY KEY (id) USING BTREE,– 根据会话查询历史消息索引INDEXidx_session(session_id) USING BTREE,– 用户未读消息查询联合索引优化离线消息加载INDEXidx_receiver_read(receiver_id,read_status) USING BTREE) ENGINE InnoDB AUTO_INCREMENT 11 CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci COMMENT ‘聊天消息表’ ROW_FORMAT Dynamic;SET FOREIGN_KEY_CHECKS 1;四、实体类 DTO 完整代码4.1 ChatSession 会话实体package com.para.system.chat.domain;import lombok.Data;import java.util.Date;/**聊天会话实体/Datapublic class ChatSession {/* 会话主键/private Long id;/* 较小用户ID/private Long fromUserId;/* 较大用户ID/private Long toUserId;/* 最后一条消息内容/private String lastContent;/* 最后消息时间/private Date lastMsgTime;/* 冗余未读数/private Integer unreadCount;/* 当前用户实时未读数查询计算字段/private Integer realSelfUnread;/* 创建时间/private Date createTime;/* 更新时间 */private Date updateTime;// 联表sys_user查询字段/** 对方昵称/private String targetNickName;/* 对方头像 */private String targetAvatar;}4.2 ChatMessage 消息实体package com.para.system.chat.domain;import lombok.Data;import java.util.Date;/**聊天消息实体/Datapublic class ChatMessage {/* 消息主键/private Long id;/* 会话ID/private Long sessionId;/* 发送人ID/private Long senderId;/* 接收人ID/private Long receiverId;/* 消息内容/private String content;/* 消息类型 1文本 2图片/private String msgType;/* 读取状态 0未读 1已读/private String readStatus;/* 发送时间 */private Date createTime;}4.3 ChatMsgDTO WebSocket传输DTO前后端WebSocket交互统一载体package com.para.system.chat.domain;import lombok.Data;/**WebSocket消息传输DTO/Datapublic class ChatMsgDTO {/* 会话ID/private Long sessionId;/* 接收人ID/private Long receiverId;/* 消息内容/private String content;/* 消息类型 1文本 2图片 */private String msgType;}4.4 ChatService 顶层接口package com.para.system.chat.service;import com.para.system.chat.domain.ChatMessage;import com.para.system.chat.domain.ChatSession;import com.para.system.chat.domain.ChatMsgDTO;import java.util.List;public interface ChatService {/** * 获取或创建双人会话 */ ChatSession getOrCreateSession(Long fromUid, Long toUid); /** * 保存消息并推送 */ void saveAndPushMessage(Long senderId, ChatMsgDTO dto); /** * 查询会话历史消息 */ ListChatMessage history(Long sessionId); /** * 查询用户会话列表支持昵称搜索 */ ListChatSession sessionList(Long uid, String nickName); /** * 批量标记会话全部消息已读 */ void markAllRead(Long sessionId, Long uid); /** * 用户上线推送离线未读消息 */ void pushOfflineMessage(Long uid);}五、代码分层实现5.1 Controller 层ChatController HTTP接口提供前端所有同步查询接口创建会话、会话列表支持昵称搜索、历史消息分页、标记消息已读。统一继承项目基础控制器内置分页工具、登录用户工具类。package com.para.system.chat.controller;import com.para.common.core.controller.BaseController;import com.para.common.core.domain.AjaxResult;import com.para.common.core.page.TableDataInfo;import com.para.common.utils.SecurityUtils;import com.para.system.chat.domain.ChatSession;import com.para.system.chat.service.ChatService;import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;import java.util.List;/**聊天接口控制器提供前端HTTP接口会话管理、消息查询、已读标记*/RestControllerRequestMapping(“/chat”)public class ChatController extends BaseController {Resourceprivate ChatService chatService;/**获取或创建与对方的会话param toUserId 对方用户ID*/GetMapping(“/session”)public AjaxResult session(RequestParam Long toUserId) {try {Long userId SecurityUtils.getLoginUser().getUserId();ChatSession session chatService.getOrCreateSession(userId, toUserId);return AjaxResult.success(session);} catch (RuntimeException e) {return AjaxResult.error(e.getMessage());}}/**获取当前用户的会话列表按最后消息时间倒序支持对方昵称模糊搜索*/GetMapping(“/sessionList”)public TableDataInfo sessionList(RequestParam(required false) String nickName) {startPage();Long userId SecurityUtils.getLoginUser().getUserId();List list chatService.sessionList(userId, nickName);return getDataTable(list);}/**获取会话历史消息分页*/GetMapping(“/history”)public TableDataInfo history(RequestParam Long sessionId) {startPage();return getDataTable(chatService.history(sessionId));}/**标记会话所有消息为已读清空未读计数*/PostMapping(“/read”)public AjaxResult read(RequestParam Long sessionId) {Long userId SecurityUtils.getLoginUser().getUserId();chatService.markAllRead(sessionId, userId);return AjaxResult.success();}}5.2 Service 业务实现层 ChatServiceImpl核心业务处理包含会话防并发创建、消息事务入库、WebSocket推送、离线消息补发、批量已读标记。package com.para.system.chat.service.impl;import com.para.common.utils.SecurityUtils;import com.para.system.chat.domain.ChatMessage;import com.para.system.chat.domain.ChatSession;import com.para.system.chat.domain.ChatMsgDTO;import com.para.system.chat.mapper.ChatMessageMapper;import com.para.system.chat.mapper.ChatSessionMapper;import com.para.system.chat.service.ChatService;import com.para.system.chat.websocket.ChatWebSocketServer;import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;import java.util.Date;import java.util.List;import java.util.Set;import java.util.stream.Collectors;/**聊天业务实现类功能会话管理、消息存储、未读计数、消息推送、离线消息补发*/Slf4jServicepublic class ChatServiceImpl implements ChatService {Resourceprivate ChatSessionMapper sessionMapper;Resourceprivate ChatMessageMapper msgMapper;/**获取或创建两个用户之间的聊天会话防并发重复创建param fromUid 当前登录用户IDparam toUid 对方用户IDreturn 会话对象*/Overridepublic ChatSession getOrCreateSession(Long fromUid, Long toUid) {// 禁止自己和自己聊天if (fromUid.equals(toUid)) {throw new RuntimeException(“不允许与自己创建聊天会话”);}// 统一大小ID存储保证双人会话唯一Long minId Math.min(fromUid, toUid);Long maxId Math.max(fromUid, toUid);Long loginUid SecurityUtils.getLoginUser().getUserId();// 查询已有会话ChatSession session sessionMapper.selectByUserPair(minId, maxId, loginUid);if (session null) {try {session new ChatSession();session.setFromUserId(minId);session.setToUserId(maxId);sessionMapper.insertSession(session);} catch (Exception e) {// 唯一索引冲突其他线程已创建日志告警不抛出异常log.warn(“会话创建冲突重新查询会话记录”);}// 冲突后二次查询返回最新会话return sessionMapper.selectByUserPair(minId, maxId, loginUid);}return session;}/**保存消息并异步推送给接收方数据库操作在事务内WebSocket推送在事务外避免长连接阻塞事务*/Overridepublic void saveAndPushMessage(Long senderId, ChatMsgDTO dto) {Long receiverId dto.getReceiverId();// 事务入库、更新会话、未读数1saveMessageInTransaction(senderId, dto);// 事务外推送消息推送失败不影响消息存储try {ChatWebSocketServer.sendToUser(receiverId, dto);} catch (Exception e) {log.error(“WebSocket推送消息失败发送人:{},接收人:{}”, senderId, receiverId, e);}}/**事务方法保存消息、更新会话最后消息、未读数自增*/Transactional(rollbackFor Exception.class)public void saveMessageInTransaction(Long senderId, ChatMsgDTO dto) {Long sessionId dto.getSessionId();Long receiverId dto.getReceiverId();Date now new Date();// 1. 插入聊天消息默认未读0ChatMessage msg new ChatMessage();msg.setSessionId(sessionId);msg.setSenderId(senderId);msg.setReceiverId(receiverId);msg.setContent(dto.getContent());msg.setMsgType(dto.getMsgType());msg.setReadStatus(“0”);msg.setCreateTime(now);msgMapper.insert(msg);// 2. 更新会话最后一条消息内容和时间sessionMapper.updateLastMsg(sessionId, dto.getContent(), now);// 3. 当前会话未读计数1sessionMapper.incrUnread(sessionId);}/**查询会话分页历史消息*/Overridepublic List history(Long sessionId) {return msgMapper.selectBySession(sessionId);}/**查询当前用户会话列表支持对方昵称模糊检索*/Overridepublic List sessionList(Long uid, String nickName) {return sessionMapper.selectUserSessions(uid, nickName);}/**批量标记会话全部消息已读清空会话未读计数*/OverrideTransactionalpublic void markAllRead(Long sessionId, Long uid) {msgMapper.batchRead(sessionId, uid);sessionMapper.clearUnread(sessionId);}/**用户WebSocket上线补发所有离线未读消息*/Overridepublic void pushOfflineMessage(Long uid) {List unreadList msgMapper.selectUnread(uid);if (unreadList null || unreadList.isEmpty()) {return;}// 逐条推送离线消息for (ChatMessage msg : unreadList) {ChatMsgDTO dto new ChatMsgDTO();dto.setSessionId(msg.getSessionId());dto.setReceiverId(msg.getReceiverId());dto.setContent(msg.getContent());dto.setMsgType(msg.getMsgType());ChatWebSocketServer.sendToUser(uid, dto);}// 推送完成批量标记已读避免重复推送try {Set sessionIds unreadList.stream().map(ChatMessage::getSessionId).collect(Collectors.toSet());for (Long sessionId : sessionIds) {msgMapper.batchRead(sessionId, uid);sessionMapper.clearUnread(sessionId);}} catch (Exception e) {log.error(“离线消息标记已读失败”, e);}}}5.3 Mapper 接口补充ChatSessionMapper.javapackage com.para.system.chat.mapper;import com.para.system.chat.domain.ChatSession;import org.apache.ibatis.annotations.Param;import java.util.List;public interface ChatSessionMapper {ChatSession selectByUserPair(Param(from) Long minId, Param(to) Long maxId, Param(uid) Long loginUid); void insertSession(ChatSession session); void updateLastMsg(Param(sid) Long sessionId, Param(content) String content, Param(time) java.util.Date time); void incrUnread(Param(sid) Long sessionId); void clearUnread(Param(sid) Long sessionId); ListChatSession selectUserSessions(Param(uid) Long uid, Param(nickName) String nickName);}ChatMessageMapper.javapackage com.para.system.chat.mapper;import com.para.system.chat.domain.ChatMessage;import org.apache.ibatis.annotations.Param;import java.util.List;public interface ChatMessageMapper {void insert(ChatMessage message); ListChatMessage selectBySession(Param(sessionId) Long sessionId); void batchRead(Param(sessionId) Long sessionId, Param(uid) Long userId); ListChatMessage selectUnread(Param(uid) Long userId);}5.4 MyBatis Mapper XML ChatSessionMapper.xml核心复杂SQL联表用户获取昵称头像、子查询实时统计未读、昵称模糊搜索、双人会话唯一查询。?xml version1.0 encodingUTF-8?!-- 会话结果映射关联用户昵称、头像、实时未读数量 -- resultMap typecom.para.system.chat.domain.ChatSession idChatSessionResult result propertyid columnid / result propertyfromUserId columnfrom_user_id / result propertytoUserId columnto_user_id / result propertylastContent columnlast_content / result propertylastMsgTime columnlast_msg_time / result propertyunreadCount columnunread_count / result propertyrealSelfUnread columnreal_self_unread/ result propertycreateTime columncreate_time / result propertyupdateTime columnupdate_time / result columnnick_name propertytargetNickName/ result columnavatar propertytargetAvatar/ /resultMap !-- 根据双方用户ID查询唯一会话关联用户信息实时未读统计 -- select idselectByUserPair resultMapChatSessionResult SELECT cs.*, su.nick_name, su.avatar, COALESCE(cm.unread_total,0) AS real_self_unread FROM chat_session cs LEFT JOIN sys_user su ON su.user_id ( CASE WHEN cs.from_user_id #{uid} THEN cs.to_user_id ELSE cs.from_user_id END ) LEFT JOIN ( SELECT session_id, COUNT(1) unread_total FROM chat_message WHERE receiver_id #{uid} AND read_status 0 GROUP BY session_id ) cm ON cs.id cm.session_id WHERE (from_user_id #{from} AND to_user_id #{to}) OR (from_user_id #{to} AND to_user_id #{from}) LIMIT 1 /select !-- 新增双人会话 -- insert idinsertSession parameterTypeChatSession INSERT INTO chat_session ( from_user_id, to_user_id, last_content, last_msg_time, unread_count, create_time, update_time ) VALUES ( #{fromUserId}, #{toUserId}, #{lastContent}, #{lastMsgTime}, #{unreadCount}, sysdate(), sysdate() ) /insert !-- 更新会话最后一条消息内容和时间 -- update idupdateLastMsg UPDATE chat_session SET last_content #{content}, last_msg_time #{time}, update_time sysdate() WHERE id #{sid} /update !-- 会话未读计数 1 -- update idincrUnread UPDATE chat_session SET unread_count unread_count 1, update_time sysdate() WHERE id #{sid} /update !-- 清空会话未读计数 -- update idclearUnread UPDATE chat_session SET unread_count 0, update_time sysdate() WHERE id #{sid} /update !-- 查询当前用户全部会话支持昵称模糊搜索按最后消息倒序 -- select idselectUserSessions resultMapChatSessionResult SELECT cs.*, su.nick_name, su.avatar, COALESCE(cm.unread_total,0) AS real_self_unread FROM chat_session cs LEFT JOIN sys_user su ON su.user_id ( CASE WHEN cs.from_user_id #{uid} THEN cs.to_user_id ELSE cs.from_user_id END ) LEFT JOIN ( SELECT session_id, COUNT(1) unread_total FROM chat_message WHERE receiver_id #{uid} AND read_status 0 GROUP BY session_id ) cm ON cs.id cm.session_id WHERE (cs.from_user_id #{uid} OR cs.to_user_id #{uid}) if testnickName ! null and nickName ! AND su.nick_name LIKE CONCAT(%, #{nickName}, %) /if ORDER BY cs.last_msg_time DESC /select5.5 ChatMessageMapper.xml补充完整 ?xml version1.0 encodingUTF-8?insert idinsert parameterTypecom.para.system.chat.domain.ChatMessage INSERT INTO chat_message (session_id, sender_id, receiver_id, content, msg_type, read_status, create_time) VALUES (#{sessionId},#{senderId},#{receiverId},#{content},#{msgType},#{readStatus},#{createTime}) /insert select idselectBySession resultTypecom.para.system.chat.domain.ChatMessage SELECT * FROM chat_message WHERE session_id #{sessionId} ORDER BY create_time ASC /select !-- 批量标记当前用户该会话所有消息为已读 -- update idbatchRead UPDATE chat_message SET read_status 1 WHERE session_id #{sessionId} AND receiver_id #{uid} AND read_status 0 /update !-- 查询用户所有未读消息离线消息 -- select idselectUnread resultTypecom.para.system.chat.domain.ChatMessage SELECT * FROM chat_message WHERE receiver_id #{uid} AND read_status 0 /select5.6 WebSocket 长连接服务 ChatWebSocketServer 实现握手鉴权、在线用户容器、心跳保活、消息收发、离线消息触发、线程安全发送。package com.para.system.chat.websocket;import cn.hutool.json.JSONUtil;import com.para.common.constant.Constants;import com.para.common.utils.spring.SpringUtils;import com.para.system.chat.domain.ChatMsgDTO;import com.para.system.chat.service.ChatService;import io.jsonwebtoken.Claims;import io.jsonwebtoken.Jwts;import lombok.extern.log4j.Log4j2;import org.springframework.core.env.Environment;import org.springframework.stereotype.Component;import javax.websocket.*;import javax.websocket.CloseReason.CloseCodes;import javax.websocket.server.PathParam;import javax.websocket.server.ServerEndpoint;import java.io.IOException;import java.net.URLDecoder;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;ComponentServerEndpoint(“/ws/chat/{userId}”)Log4j2public class ChatWebSocketServer {// 在线用户容器userId - WebSocket会话 private static final MapLong, Session ONLINE_USER_MAP new ConcurrentHashMap(); // 静态获取Spring容器Bean解决WebSocket无法直接注入Service问题 private static ChatService getChatService() { return SpringUtils.getBean(ChatService.class); } private boolean isBlank(String str) { return str null || str.trim().length() 0; } // 解析URL Query参数兼容带等号的JWT Token private String getQueryParam(Session session, String paramName) { String query session.getQueryString(); log.info(完整URL参数串: {}, query); if (isBlank(query)) { return null; } String[] paramArr query.split(); for (String item : paramArr) { int eqIndex item.indexOf(); if (eqIndex -1) continue; String key item.substring(0, eqIndex); String value item.substring(eqIndex 1); if (paramName.equals(key)) { try { String decodeVal URLDecoder.decode(value, UTF-8); log.info(解析出token值: {}, decodeVal); return decodeVal; } catch (Exception e) { log.error(URL解码token失败, e); return value; } } } return null; } /** * WebSocket握手建立连接 */ OnOpen public void onOpen(Session session, PathParam(userId) Long userId) { log.info(收到WS握手请求路径userId:{}, userId); // 新连接自动剔除旧连接保证单用户仅一条在线通道 ONLINE_USER_MAP.remove(userId); String token getQueryParam(session, token); // 无Token直接断开连接 if (isBlank(token)) { log.warn(拦截断开未携带登录凭证); closeSession(session, new CloseReason(CloseCodes.CANNOT_ACCEPT, 未携带登录凭证)); return; } // 解析JWT Token与后台登录校验逻辑统一 String loginUserKey; try { Environment env SpringUtils.getBean(Environment.class); String secret env.getProperty(token.secret); Claims claims Jwts.parser() .setSigningKey(secret) .parseClaimsJws(token) .getBody(); loginUserKey claims.get(Constants.LOGIN_USER_KEY).toString(); log.info(token解析成功登录唯一标识:{}, loginUserKey); } catch (Exception e) { log.error(拦截断开Token解析异常, e); closeSession(session, new CloseReason(CloseCodes.CANNOT_ACCEPT, Token无效或已过期)); return; } // 校验通过存入在线容器触发离线消息补发 log.info(全部校验通过建立连接路径用户ID:{}, userId); ONLINE_USER_MAP.put(userId, session); ChatService chatService getChatService(); if (chatService ! null) { try { chatService.pushOfflineMessage(userId); } catch (Exception ignored) { log.error(推送离线消息异常, ignored); } } } /** * 同步安全发送消息加锁防止并发发送通道异常 */ private void safeSendText(Session session, String text) throws IOException { synchronized (session) { if (session.isOpen()) { session.getBasicRemote().sendText(text); } } } /** * 统一关闭会话封装方法 */ private void closeSession(Session session, CloseReason reason) { try { session.close(reason); } catch (Exception e) { log.error(关闭会话异常, e); } } /** * 连接关闭回调 */ OnClose public void onClose(Session session, CloseReason reason, PathParam(userId) Long userId) { ONLINE_USER_MAP.remove(userId); log.info(连接关闭用户ID:{} 关闭状态码:{}, userId, reason.getCloseCode()); } /** * 接收前端消息业务消息 PING心跳 */ OnMessage public void onMessage(String message, PathParam(userId) Long senderId) { if (isBlank(message)) { return; } // 心跳PING回复PONG保活 if (PING.equals(message)) { try { Session session ONLINE_USER_MAP.get(senderId); if(session ! null session.isOpen()){ safeSendText(session, PONG); log.info(收到PING心跳回复PONG用户ID:{},senderId); } } catch (Exception e) { log.error(回复PONG心跳失败, e); } return; } // 忽略服务端返回的PONG包 if (PONG.equals(message)) { return; } // 解析业务消息调用Service保存并推送 try { ChatMsgDTO dto JSONUtil.toBean(message, ChatMsgDTO.class); ChatService chatService getChatService(); if (chatService ! null) { chatService.saveAndPushMessage(senderId, dto); } } catch (Exception e) { log.error(处理消息异常, e); } } /** * 通道异常回调过滤正常EOF关闭日志 */ OnError public void onError(Session session, Throwable e, PathParam(userId) Long userId) { if(e instanceof java.io.EOFException){ log.info(WS通道正常关闭用户ID:{}, userId); }else{ log.error(WS异常用户ID:{}, userId, e); } ONLINE_USER_MAP.remove(userId); } /** * 静态工具方法向指定用户推送消息 */ public static void sendToUser(Long userId, Object data) { Session session ONLINE_USER_MAP.get(userId); if (session ! null session.isOpen()) { String json JSONUtil.toJsonStr(data); try { synchronized (session) { session.getBasicRemote().sendText(json); } } catch (Exception e) { log.error(发送消息失败用户ID:{}, userId, e); } } }}六、核心难点与解决方案总结双人会话重复创建并发问题数据库建立 from_user_id, to_user_id 联合唯一索引统一小ID存from、大ID存to保证双人仅一条会话捕获插入唯一索引异常冲突后二次查询避免重复数据不同用户查看同一会话未读数错乱放弃会话表单数字段存储未读使用子查询实时统计当前登录用户未读消息每个用户查询时独立计算多端、多用户隔离准确。WebSocket无法注入Spring Bean采用 SpringUtils.getBean() 静态工具类上下文获取Service解决原生WebSocket非Spring管理无法依赖注入问题。WebSocket并发发送报通道异常消息发送方法加 synchronized(session) 同步锁多线程发送时排队执行规避 IllegalStateException。离线消息重复推送用户上线一次性拉取所有未读消息推送完成后批量分组标记消息已读并清空会话未读数避免每次重连重复补发。WebSocket安全鉴权握手阶段双重校验URL路径userId URL携带JWT Token复用后台登录JWT密钥解析非法/过期Token直接主动关闭连接杜绝非法连接。长连接阻塞数据库事务消息入库、会话更新放入事务WebSocket推送逻辑放到事务外部执行防止推送超时长时间占用数据库连接。七、扩展优化方向消息持久化缓存热门会话消息加入Redis缓存减少MySQL分页查询压力分布式WebSocket当前单机内存存储在线用户集群部署需改造Redis订阅发布实现跨服务推送消息类型扩展图片、文件、语音消息上传OSS后存储访问链接会话置顶、删除、拉黑功能会话表新增状态字段扩展消息撤回、编辑消息表增加撤回标识、原始内容字段心跳超时自动下线增加定时任务清理长期无心跳僵尸连接消息幂等前端携带消息唯一ID避免重复发送重复入库八、项目适配说明该模块基于若依RuoYi-Vue SpringBoot后台开发依赖项目内置工具类SecurityUtils获取当前登录用户IDAjaxResult / TableDataInfo统一前后端返回体SpringUtilsSpring上下文静态获取BeanJWT工具、Token密钥配置 token.secret
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
