聊天功能开发通用模板

通用即时通讯能力集成指南，适用于订单/会话维度的实时文字+图片聊天场景。

一、功能概述

为业务系统提供基于 WebSocket 的实时通信能力，支持：

  • 实时消息：在线用户之间即时收发文字/图片消息
  • 消息持久化：所有消息异步写入数据库，支持历史记录查询
  • 离线消息：离线用户可通过 REST 接口拉取历史消息
  • 会话管理：按业务维度（如订单、工单、群组）组织聊天会话

核心流程：

用户 WebSocket 连接 → 发送消息 → 服务器持久化 → 广播给同房间在线用户
                                                → 离线用户通过 REST 拉取历史

二、技术选型参考

层次 推荐技术 说明
后端实时通信 Spring Boot WebSocket (JSR-356) 原生 WebSocket，与 Spring Boot 无缝集成
消息异步持久化 RocketMQ / RabbitMQ / Kafka 发送消息后异步写入数据库，避免 WebSocket 线程阻塞
消息历史存储 MySQL + MyBatis-Plus 结构化存储，按业务ID和时间索引
图片/文件存储 MinIO / OSS / S3 对象存储，持久化媒体文件
前端实时通信 原生 WebSocket API 轻量，无需额外依赖
前端 UI Element Plus / Ant Design 项目中已有的 UI 框架即可

变体方案

  • 无 MQ 场景：可在 Handler 中直接调用 Mapper 持久化（同步写，适合低并发）
  • 单聊场景：房间 = 订单ID，通过 sender/receiver 确定通信双方
  • 群聊场景：房间 = 群组ID，所有群成员共享房间

三、数据库设计

3.1 核心消息表

CREATE TABLE `chat_msg` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '自增主键',
  `msg_no` varchar(64) NOT NULL COMMENT '消息唯一ID(防重)',
  `biz_id` bigint NOT NULL COMMENT '关联业务ID（订单/工单/群组ID）',
  `sender_type` tinyint NOT NULL COMMENT '发送方角色类型（业务自定义）',
  `sender_id` bigint NOT NULL COMMENT '发送人用户ID',
  `receiver_id` bigint DEFAULT NULL COMMENT '接收人用户ID（群聊可为空）',
  `msg_type` int NOT NULL COMMENT '消息类型：1文字 2图片 3文件 4系统消息',
  `text_content` varchar(2048) DEFAULT '' COMMENT '文字内容',
  `file_url` varchar(255) DEFAULT '' COMMENT '文件/图片地址',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '发送时间',
  `del_flag` int DEFAULT '0' COMMENT '逻辑删除 0正常 1删除',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_msg_no` (`msg_no`),
  KEY `idx_biz_id` (`biz_id`,`create_time`)
) ENGINE=InnoDB COMMENT='聊天消息表';

3.2 索引说明

索引 作用
uk_msg_no 消息去重（MQ 可能重复投递，利用唯一索引保证幂等）
idx_biz_id 按业务维度查询历史消息

3.3 扩展字段建议

可按业务需求在 chat_msg 表增加字段：

  • parent_id：回复某条消息的消息ID
  • extra：扩展 JSON 字段（如 @某人、消息引用等）
  • is_read：是否已读（单聊场景）

四、后端实现

4.1 依赖配置

<!-- WebSocket -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
​
<!-- MQ（按需选择） -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>

4.2 WebSocket 配置注册

作用：注册 WebSocket 处理器端点，配置握手拦截器。

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
​
    @Autowired
    private ChatWebSocketHandler chatWebSocketHandler;
​
    @Autowired
    private ChatWebSocketInterceptor chatWebSocketInterceptor;
​
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(chatWebSocketHandler, "/ws/chat/{bizId}")
                .addInterceptors(chatWebSocketInterceptor)
                .setAllowedOrigins("*");
    }
}

关键点：

  • 端点路径 {bizId} 为路径变量，标识业务维度（订单/工单ID）
  • 需要在 HTTP 拦截器配置中排除 /ws/** 路径

4.3 握手拦截器

作用：在 WebSocket 握手阶段校验用户身份，将用户信息存入 session attributes。

@Component
public class ChatWebSocketInterceptor implements HandshakeInterceptor {
​
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
                                   WebSocketHandler wsHandler, Map<String, Object> attributes) {
        // 1. 从 query 参数中获取 token
        String query = request.getURI().getQuery();
        if (query == null || !query.contains("token=")) {
            return false;
        }
​
        // 2. 解析 token 获取用户信息
        String token = extractToken(query);
        if (token == null || token.isEmpty()) {
            return false;
        }
​
        try {
            // 3. 校验 token，提取 userId / role 等信息
            // （替换为项目实际的 JWT 工具类）
            DecodedJWT decoded = JWTUtil.verify(token);
            Long userId = decoded.getClaim("userId").asLong();
            String role = decoded.getClaim("role").asString();
​
            attributes.put("userId", userId);
            attributes.put("role", role);
            return true;
        } catch (Exception e) {
            return false;
        }
    }
​
    private String extractToken(String query) {
        for (String param : query.split("&")) {
            if (param.startsWith("token=")) {
                return param.substring(6);
            }
        }
        return null;
    }
​
    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
                               WebSocketHandler wsHandler, Exception exception) {
    }
}

注意：需要在项目公共 Web 拦截器配置中将 /ws/** 排除，避免 HTTP 拦截器干扰 WebSocket 握手：

// WebMvcConfig 或 WebConfig 中
registry.addInterceptor(authInterceptor)
        .addPathPatterns("/**")
        .excludePathPatterns("/ws/**");

4.4 核心消息处理器

作用：WebSocket 连接生命周期管理 + 消息收发核心逻辑。

@Component
public class ChatWebSocketHandler extends TextWebSocketHandler {
​
    private static final Logger log = LoggerFactory.getLogger(ChatWebSocketHandler.class);
​
    /** 房间映射：bizId -> 在线 WebSocket 会话列表 */
    private static final ConcurrentHashMap<Long, CopyOnWriteArrayList<WebSocketSession>> ROOM_MAP = new ConcurrentHashMap<>();
​
    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        Long bizId = getBizIdFromSession(session);
        if (bizId == null) return;
​
        CopyOnWriteArrayList<WebSocketSession> sessions =
                ROOM_MAP.computeIfAbsent(bizId, k -> new CopyOnWriteArrayList<>());
        sessions.add(session);
        log.info("WS连接建立: bizId={}, sessionId={}, 当前在线={}", bizId, session.getId(), sessions.size());
    }
​
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
        try {
            ChatMessageDTO dto = JSONUtil.toBean(message.getPayload(), ChatMessageDTO.class);
​
            // 心跳处理
            if ("ping".equals(dto.getType())) {
                session.sendMessage(new TextMessage("{\"type\":\"pong\"}"));
                return;
            }
​
            if (!"chat".equals(dto.getType())) {
                return;
            }
​
            Long userId = (Long) session.getAttributes().get("userId");
            String role = (String) session.getAttributes().get("role");
            Long bizId = getBizIdFromSession(session);
​
            // 填充服务端字段
            dto.setMsgNo(UUID.randomUUID().toString().replace("-", ""));
            dto.setSenderId(userId);
            dto.setSenderType(resolveSenderType(role));
            dto.setCreateTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
​
            // ① 异步持久化（MQ 或直接线程池）
            asyncPersist(dto);
​
            // ② 广播给同房间其他用户
            broadcastToRoom(bizId, dto, userId);
​
            // ③ 返回 ACK 给发送方
            JSONObject ack = JSONUtil.createObj()
                    .set("type", "ack")
                    .set("msgNo", dto.getMsgNo())
                    .set("createTime", dto.getCreateTime())
                    .set("status", "ok");
            session.sendMessage(new TextMessage(ack.toString()));
​
        } catch (Exception e) {
            log.error("处理 WS 消息失败", e);
        }
    }
​
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        Long bizId = getBizIdFromSession(session);
        if (bizId == null) return;
​
        CopyOnWriteArrayList<WebSocketSession> sessions = ROOM_MAP.get(bizId);
        if (sessions != null) {
            sessions.remove(session);
            if (sessions.isEmpty()) {
                ROOM_MAP.remove(bizId);
            }
        }
        log.info("WS连接关闭: bizId={}, sessionId={}", bizId, session.getId());
    }
​
    /** 广播给房间内除发送者外的用户 */
    private void broadcastToRoom(Long bizId, ChatMessageDTO dto, Long excludeUserId) {
        CopyOnWriteArrayList<WebSocketSession> sessions = ROOM_MAP.get(bizId);
        if (sessions == null) return;
​
        JSONObject broadcast = buildBroadcastPayload(dto);
        String payload = broadcast.toString();
​
        for (WebSocketSession ws : sessions) {
            if (ws.isOpen()) {
                Long uid = (Long) ws.getAttributes().get("userId");
                if (uid != null && !uid.equals(excludeUserId)) {
                    try {
                        ws.sendMessage(new TextMessage(payload));
                    } catch (IOException e) {
                        log.error("广播失败, sessionId={}", ws.getId(), e);
                    }
                }
            }
        }
    }
​
    /** 从路径中提取 bizId */
    private Long getBizIdFromSession(WebSocketSession session) {
        String path = session.getUri().getPath();
        String[] segments = path.split("/");
        try {
            return Long.parseLong(segments[segments.length - 1]);
        } catch (NumberFormatException e) {
            return null;
        }
    }
​
    private JSONObject buildBroadcastPayload(ChatMessageDTO dto) {
        return JSONUtil.createObj()
                .set("type", "chat")
                .set("msgNo", dto.getMsgNo())
                .set("bizId", dto.getBizId())
                .set("senderId", dto.getSenderId())
                .set("senderType", dto.getSenderType())
                .set("msgType", dto.getMsgType())
                .set("textContent", dto.getTextContent())
                .set("fileUrl", dto.getFileUrl() != null ? dto.getFileUrl() : "")
                .set("createTime", dto.getCreateTime());
    }
​
    /** 子类重写此方法实现持久化逻辑 */
    protected void asyncPersist(ChatMessageDTO dto) {
        // 默认空实现，由子类或消费者实现
    }
​
    /** 子类重写此方法实现角色映射 */
    protected int resolveSenderType(String role) {
        // 自定义映射逻辑
        return 1;
    }
}

4.5 消息传输对象

@Data
public class ChatMessageDTO {
    private String type;        // 消息类型: chat / ping / ack
    private Long bizId;         // 业务ID（订单/工单/群组ID）
    private Integer msgType;    // 1文字 2图片 3文件
    private String textContent; // 文字内容
    private String fileUrl;     // 文件/图片URL
​
    // 以下由服务端填充
    private String msgNo;       // 消息唯一ID(UUID)
    private Long senderId;      // 发送人用户ID
    private Integer senderType; // 发送方角色类型
    private String createTime;  // 服务端时间
    private String status;      // ACK状态
}

4.6 消息处理流程

客户端发送 → Handler接收
                ├─ ① 异步持久化（MQ / 线程池）
                ├─ ② 广播给同房间其他用户
                └─ ③ 返回 ACK 给发送方

4.7 消息持久化

方式一：MQ 消费者（推荐，异步解耦）

@Component
@RocketMQMessageListener(
        topic = "chat-persist-topic",
        consumerGroup = "chat-persist-group",
        consumeMode = ConsumeMode.CONCURRENTLY,
        messageModel = MessageModel.CLUSTERING
)
public class ChatMsgConsumer implements RocketMQListener<String> {
​
    @Autowired
    private ChatMsgMapper chatMsgMapper;
​
    @Override
    public void onMessage(String msg) {
        try {
            ChatMessageDTO dto = JSONUtil.toBean(msg, ChatMessageDTO.class);
            ChatMsg entity = ChatMsg.builder()
                    .msgNo(dto.getMsgNo())
                    .bizId(dto.getBizId())
                    .senderType(dto.getSenderType())
                    .senderId(dto.getSenderId())
                    .msgType(dto.getMsgType())
                    .textContent(dto.getTextContent() != null ? dto.getTextContent() : "")
                    .fileUrl(dto.getFileUrl() != null ? dto.getFileUrl() : "")
                    .build();
            chatMsgMapper.insert(entity);
        } catch (DuplicateKeyException e) {
            // msg_no 唯一索引保证幂等，重复投递直接忽略
            log.warn("消息重复: msgNo={}", dto.getMsgNo());
        } catch (Exception e) {
            log.error("消息持久化失败: {}", msg, e);
        }
    }
}

幂等机制：msg_no 唯一索引 + DuplicateKeyException 捕获，天然防重复入库。

方式二：线程池直接写入（无需 MQ）

// 在 ChatWebSocketHandler 中重写 asyncPersist
@Autowired
private ChatMsgMapper chatMsgMapper;
​
private final ExecutorService persistExecutor = Executors.newFixedThreadPool(4);
​
@Override
protected void asyncPersist(ChatMessageDTO dto) {
    persistExecutor.submit(() -> {
        try {
            ChatMsg entity = ChatMsg.builder()
                    .msgNo(dto.getMsgNo())
                    // ... 映射字段
                    .build();
            chatMsgMapper.insert(entity);
        } catch (DuplicateKeyException e) {
            // 忽略重复
        }
    });
}

4.8 实体与 Mapper

实体

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@TableName("chat_msg")
public class ChatMsg {
    @TableId(type = IdType.AUTO)
    private Long id;
    private String msgNo;
    private Long bizId;
    private Integer senderType;
    private Long senderId;
    private Long receiverId;
    private Integer msgType;
    private String textContent;
    private String fileUrl;
    private Date createTime;
    private Integer delFlag;
}

Mapper

@Mapper  // 或继承 BaseMapper 并在启动类上使用 @MapperScan
public interface ChatMsgMapper extends BaseMapper<ChatMsg> {
}

4.9 Service 接口定义

public interface ChatService {
​
    /** 分页查询聊天历史 */
    IPage<ChatMsg> getHistory(Long bizId, Long userId, int page, int size);
​
    /** 获取当前用户的会话列表（每个业务的最新消息摘要） */
    List<ConversationVO> getConversations(Long userId);
​
    /** 上传聊天文件/图片 */
    String uploadFile(MultipartFile file, Long userId);
}

4.10 Controller（REST 接口）

@RestController
@RequestMapping("/chat")
@Tag(name = "聊天", description = "聊天接口")
public class ChatController {
​
    @Autowired
    private ChatService chatService;
​
    @GetMapping("/history/{bizId}")
    @Operation(summary = "获取聊天记录")
    public Result<IPage<ChatMsg>> history(
            @PathVariable Long bizId,
            @RequestParam(defaultValue = "1") int page,
            @RequestParam(defaultValue = "20") int size) {
        Long userId = getCurrentUserId();
        return Result.success(chatService.getHistory(bizId, userId, page, size));
    }
​
    @PostMapping("/upload")
    @Operation(summary = "上传聊天文件")
    public Result<String> uploadFile(@RequestParam("file") MultipartFile file) {
        Long userId = getCurrentUserId();
        return Result.success(chatService.uploadFile(file, userId));
    }
​
    @GetMapping("/conversations")
    @Operation(summary = "获取会话列表")
    public Result<List<ConversationVO>> conversations() {
        Long userId = getCurrentUserId();
        return Result.success(chatService.getConversations(userId));
    }
}