核心业务处理，包含会话防并发创建、消息事务入库、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;
/**
聊天业务实现类
功能：会话管理、消息存储、未读计数、消息推送、离线消息补发
*/
@Slf4j
@Service
public class ChatServiceImpl implements ChatService {@Resource
private ChatSessionMapper sessionMapper;@Resource
private ChatMessageMapper msgMapper;/**
获取或创建两个用户之间的聊天会话（防并发重复创建）
@param fromUid 当前登录用户ID
@param toUid 对方用户ID
@return 会话对象
*/
@Override
public 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推送在事务外，避免长连接阻塞事务
*/
@Override
public void saveAndPushMessage(Long senderId, ChatMsgDTO dto) {
Long receiverId = dto.getReceiverId();
// 事务入库、更新会话、未读数+1
saveMessageInTransaction(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. 插入聊天消息，默认未读0
ChatMessage 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. 当前会话未读计数+1
sessionMapper.incrUnread(sessionId);
}
/**
- 查询会话分页历史消息
*/
@Override
public List history(Long sessionId) {
return msgMapper.selectBySession(sessionId);
}
/**
- 查询当前用户会话列表，支持对方昵称模糊检索
*/
@Override
public List sessionList(Long uid, String nickName) {
return sessionMapper.selectUserSessions(uid, nickName);
}
/**
- 批量标记会话全部消息已读，清空会话未读计数
*/
@Override
@Transactional
public void markAllRead(Long sessionId, Long uid) {
msgMapper.batchRead(sessionId, uid);
sessionMapper.clearUnread(sessionId);
}
/**
用户WebSocket上线，补发所有离线未读消息
*/
@Override
public 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.java
package 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); List<ChatSession> selectUserSessions(@Param("uid") Long uid, @Param("nickName") String nickName);}
ChatMessageMapper.java
package 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); List<ChatMessage> selectBySession(@Param("sessionId") Long sessionId); void batchRead(@Param("sessionId") Long sessionId, @Param("uid") Long userId); List<ChatMessage> selectUnread(@Param("uid") Long userId);}
5.4 MyBatis Mapper XML ChatSessionMapper.xml
核心复杂SQL：联表用户获取昵称头像、子查询实时统计未读、昵称模糊搜索、双人会话唯一查询。
<!-- 会话结果映射，关联用户昵称、头像、实时未读数量 --> <resultMap type="com.para.system.chat.domain.ChatSession" id="ChatSessionResult"> <result property="id" column="id" /> <result property="fromUserId" column="from_user_id" /> <result property="toUserId" column="to_user_id" /> <result property="lastContent" column="last_content" /> <result property="lastMsgTime" column="last_msg_time" /> <result property="unreadCount" column="unread_count" /> <result property="realSelfUnread" column="real_self_unread"/> <result property="createTime" column="create_time" /> <result property="updateTime" column="update_time" /> <result column="nick_name" property="targetNickName"/> <result column="avatar" property="targetAvatar"/> </resultMap> <!-- 根据双方用户ID查询唯一会话，关联用户信息+实时未读统计 --> <select id="selectByUserPair" resultMap="ChatSessionResult"> 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 id="insertSession" parameterType="ChatSession"> 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 id="updateLastMsg"> UPDATE chat_session SET last_content = #{content}, last_msg_time = #{time}, update_time = sysdate() WHERE id = #{sid} </update> <!-- 会话未读计数 +1 --> <update id="incrUnread"> UPDATE chat_session SET unread_count = unread_count + 1, update_time = sysdate() WHERE id = #{sid} </update> <!-- 清空会话未读计数 --> <update id="clearUnread">
