IM消息怎么存？从建表到幂等的完整方案

上两篇讲了 WebSocket 实时通道和 outbox 可靠投递，通道有了、投递也靠谱了，接下来要面对一个更基础的问题：消息怎么存？

听起来简单——不就是一张 message 表吗？建个表、写条 INSERT，完事。

说实话，我一开始也是这么想的。等真正开始写 IM 模块的时候才发现，消息存储有一堆细节如果不提前想清楚，后面每一层都会踩坑。

一、建表设计——消息表长什么样

先看 Workforce Hub 里 im_message 的实际表结构：

CREATE TABLE `im_message` (
    `id`                BIGINT       NOT NULL COMMENT '主键',
    `conversation_id`   BIGINT       NOT NULL COMMENT '会话ID',
    `seq_no`            BIGINT       NOT NULL COMMENT '消息序号',
    `sender_type`       VARCHAR(32)  NOT NULL COMMENT '发送人类型: USER/AGENT/SYSTEM',
    `sender_id`         BIGINT       NOT NULL COMMENT '发送人ID',
    `message_type`      VARCHAR(32)  NOT NULL DEFAULT 'TEXT' COMMENT '消息类型',
    `client_message_id` VARCHAR(128) DEFAULT NULL COMMENT '客户端消息幂等键',
    `content_json`      JSON         NOT NULL COMMENT '消息内容JSON',
    `quoted_message_id` BIGINT       DEFAULT NULL COMMENT '引用消息ID',
    `status`            VARCHAR(32)  NOT NULL DEFAULT 'NORMAL' COMMENT '状态',
    `sent_at`           DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    `edited_at`         DATETIME     DEFAULT NULL COMMENT '编辑时间',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_conversation_seq` (`conversation_id`, `seq_no`),
    UNIQUE KEY `uk_im_message_client_message` (`conversation_id`, `sender_type`, `sender_id`, `client_message_id`),
    KEY `idx_sender` (`sender_type`, `sender_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

每一列都有它存在的理由，一个一个说。

1.1 为什么用独立的 `seq_no` 而不是自增 ID？

消息列表是按时间顺序展示的，天然需要一个递增序号。用 id（雪花算法或 UUID）可以唯一标识一条消息，但不能代表顺序。因为：

雪花 ID 趋势递增但不严格有序——多实例并发时可能乱序
消息顺序必须严格单调——第 100 条消息就是第 100 条，不能因为 ID 乱序导致客户端显示错乱

所以 seq_no 是会话级别的严格递增序号，每发一条消息 +1。再加上 (conversation_id, seq_no) 唯一约束，自然防止了同序号重复写入。

flowchart LR subgraph session_a["会话 A"] A1["seq=1 消息1"] --> A2["seq=2 消息2"] --> A3["seq=3 消息3"] end subgraph session_b["会话 B"] B1["seq=1 消息1"] --> B2["seq=2 消息2"] end subgraph id_warning["雪花 ID 的问题"] W1["ID: 998"] --> W2["ID: 1001"] --> W3["ID: 997 ← 乱序!"] end style W3 fill:#ff6b6b,stroke:#c92a2a,color:#fff

seq_no 在每个会话内独立递增，而雪花 ID 是全局生成、趋势递增但不严格有序。消息顺序这种事，不能赌概率。

1.2 发送人为什么拆成 `sender_type` + `sender_id`？

IM 不只是用户发的，还有：

USER — 普通员工
AGENT — AI Agent 回复
SYSTEM — 系统通知（入群、改名等）

如果只有一个 sender_id，AI Agent 的 ID 可能和用户 ID 冲突（不同 ID 生成策略）。拆成 type + id 的组合，既能区分发送人类型，也能支持按类型查询。

1.3 `content_json` 为什么是 JSON 而不是固定字段？

消息类型不同，内容结构完全不同：

// 文本消息
{"text": "今天请假审批过了吗", "mentions": [{"userId": 123, "offset": 0, "length": 2}]}

// 图片消息
{"text": "", "attachments": [{"refId": 456, "type": "image", "url": "/upload/..."}]}

// 卡片消息
{"cardType": "leave_approval", "cardData": {"leaveId": 78, "status": "approved"}}

如果为每种消息类型建不同的列（text_content、image_url、card_data…），表会变得又宽又乱。JSON 列用一个字段容纳所有类型的消息体，新增消息类型不需要改表结构。

血泪教训：我一开始想用固定列来存，写到第三种消息类型的时候就开始后悔了——每加一种就得 ALTER TABLE，还得处理旧数据的 NULL 列。

二、游标分页——为什么不直接用 offset？

消息列表的翻页需求很明确：从最新消息往上拉，加载更早的消息。直觉方案是 LIMIT offset, size：

-- 直觉方案（不要这么写）
SELECT * FROM im_message
WHERE conversation_id = ?
ORDER BY seq_no DESC
LIMIT 50, 20;  -- 翻到第三页：offset 越来越大

这个方案有两个问题：

问题一：性能退化。offset 越大，MySQL 需要扫描并丢弃的行越多。翻到第 50 页时，offset = 1000，实际上扫描了 1020 行只返回 20 行。消息量大的会话（比如公司全员通知群），offset 分页越翻越慢。

问题二：数据不一致。如果在翻页过程中有新消息插入，seq_no 主键的排序会发生偏移——用户可能看到重复消息或漏掉消息。

实际采用的方案是游标分页（cursor-based pagination）：

-- 游标分页：beforeSeq 是上一页最小的 seq_no
SELECT *
FROM (
    SELECT id AS message_id, conversation_id, seq_no,
           sender_type, sender_id, message_type,
           CAST(content_json AS CHAR) AS content_json,
           quoted_message_id, status, sent_at
    FROM im_message
    WHERE conversation_id = ?
      AND seq_no < ?              -- 游标：小于上一页最小编号
      AND status <> 'DELETED'     -- 排除直接删除的消息
      AND NOT EXISTS (            -- 排除用户自己删除的消息
          SELECT 1 FROM im_message_user_action ua
          WHERE ua.message_id = im_message.id
            AND ua.user_id = ?
            AND ua.action_type = 'DELETE'
            AND ua.status = 'ACTIVE'
      )
    ORDER BY seq_no DESC
    LIMIT ?
) previous_page
ORDER BY seq_no ASC  -- 外层再排回来，保证从旧到新

用 seq_no 做游标，每次翻页只扫描目标区间，性能稳定。而且即使有新消息插入（seq_no 更大），也不会影响游标定位的历史消息。

用大白话说就是：不用"第几页第几条"的方式定位，而是说"给我编号小于 100 的消息中最新的 20 条"——位置锚定在数据上，不依赖偏移量。

flowchart TD subgraph offset_bad["offset 分页（❌）"] O1["第 1 页 LIMIT 0,20"] --> O2["第 2 页 LIMIT 20,20 扫描 40 行"] --> O3["第 10 页 LIMIT 200,20 扫描 220 行 🤢"] end subgraph cursor_good["游标分页（✅）"] C1["首页 按 seq DESC"] --> C2["下一页 WHERE seq < 80 只扫描 20 行"] --> C3["下下页 WHERE seq < 60 只扫描 20 行"] end style O3 fill:#ff6b6b,stroke:#c92a2a,color:#fff style C3 fill:#51cf66,stroke:#2f9e44,color:#000

三、客户端幂等——重试不发重复消息

这是我最开始完全没考虑到的点。

场景很简单：用户点"发送"，HTTP 请求发出去了，但因为网络抖动，客户端没收到服务端的响应。用户不知道消息有没有保存成功，只能重试。如果服务端不做处理，同一条消息就会被存两次。

这不只是"消息重复显示"的问题——如果消息是一笔审批提醒，重复推送可能让审批人莫名其妙确认了两次。

Workforce Hub 的方案是客户端消息幂等键（client_message_id）：

ALTER TABLE `im_message`
    ADD COLUMN `client_message_id` VARCHAR(128) DEFAULT NULL
        COMMENT '客户端消息幂等键' AFTER `message_type`;

ALTER TABLE `im_message`
    ADD UNIQUE KEY `uk_im_message_client_message`
        (`conversation_id`, `sender_type`, `sender_id`, `client_message_id`);

客户端在发消息时生成一个 UUID 作为 clientMessageId：

// ImService.sendCurrentEmployeeMessage() 中的幂等逻辑
if (clientMessageId != null) {
    // 先查：这条消息是不是已经存过了？
    ImMessageResponse existing = imMessageRepository
        .findMessageByClientMessageId(
            conversationId,
            USER_SENDER_TYPE,
            userId,
            clientMessageId
        )
        .map(this::toMessageResponse)
        .orElse(null);
    if (existing != null) {
        // 已存在：直接返回，更新已读位置
        imMessageRepository.updateMemberLastReadSeq(conversationId, userId, existing.getSeqNo());
        return existing;
    }
}
// 没存过：走正常 INSERT
try {
    imMessageRepository.insertMessage(...);
} catch (DuplicateKeyException e) {
    // 并发场景：INSERT 时唯一约束冲突
    if (clientMessageId != null) {
        // 说明另一个线程已经写入了，查出来返回
        return imMessageRepository.findMessageByClientMessageId(...)
            .map(this::toMessageResponse)
            .orElseThrow(() -> new BizException(409, "消息幂等写入冲突"));
    }
}

两层保护：

应用层查重：INSERT 前先 SELECT，命中直接返回
数据库唯一约束：防止并发 INSERT 绕过查重，DuplicateKeyException 兜底

为什么用 (conversation_id, sender_type, sender_id, client_message_id) 四元组？ 因为：

client_message_id 是客户端生成的 UUID，不同客户端可能碰撞（虽然概率极低）
加上会话和发送人，把作用范围缩小到"这个人在这个会话里生成的这个 ID"，实际上不可能碰撞

四、消息编辑与撤回——状态流转

消息发出去之后还能改——编辑和撤回。表里用 status 字段跟踪：

状态	含义	触发操作
`NORMAL`	正常状态	新建消息
`EDITED`	已编辑	用户编辑内容
`RECALLED`	已撤回	用户撤回消息

编辑操作的 SQL：

UPDATE im_message
SET content_json = CAST(? AS JSON),
    status = 'EDITED',
    edited_at = ?
WHERE id = ?
  AND status IN ('NORMAL', 'EDITED')

两个细节：

只更新 NORMAL 和 EDITED 状态：已撤回的消息不能再编辑
edited_at 单独记录：前端可以展示"已编辑"标签，区分原文和修改时间

撤回也是更新 content_json——不是把消息删掉，而是把内容替换成"此消息已被撤回"的标识 JSON，通过 WebSocket 推送给会话中所有成员刷新。

五、用户级软删除——每个人都有"删除"的权利

这是 IM 和传统 CRUD 最大的差异之一。

传统做法：删就是 DELETE FROM table WHERE id = ?，然后这条记录就没了。但在 IM 里不行——你删了这条消息，别人还要看。

Workforce Hub 的方案是独立的用户动作表：

CREATE TABLE `im_message_user_action` (
    `id`              BIGINT       NOT NULL COMMENT '主键',
    `conversation_id` BIGINT       NOT NULL COMMENT '会话ID',
    `message_id`      BIGINT       NOT NULL COMMENT '消息ID',
    `user_id`         BIGINT       NOT NULL COMMENT '用户ID',
    `action_type`     VARCHAR(32)  NOT NULL COMMENT '动作类型',
    `payload_json`    JSON         DEFAULT NULL COMMENT '动作附加内容',
    `remind_at`       DATETIME     DEFAULT NULL COMMENT '提醒时间',
    `status`          VARCHAR(32)  NOT NULL DEFAULT 'ACTIVE',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_im_message_user_action` (`message_id`, `user_id`, `action_type`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

消息查询时，用 NOT EXISTS 子查询排除当前用户已删除的消息（第二节的 SQL 里已经写了）。

这张表不只处理删除，还承载了三种用户动作：

DELETE — 删除（仅自己不可见）
FAVORITE — 收藏（标记重要消息）
REMINDER — 设置提醒（定时提醒查看这条消息）

一个动作一条记录，状态可切换（ACTIVE / INACTIVE），支持撤销操作。撤销时把 status 改回 INACTIVE 而不是 DELETE 记录，方便审计追溯。

六、并发安全——一个会话只能有一个人"动笔"

发消息要生成 seq_no = lastSeq + 1，这在并发场景下有经典的竞态问题：A 和 B 同时读取 lastSeq = 100，都算出 seqNo = 101，然后都 INSERT seq_no = 101。结果？唯一约束报错，其中一个失败。

解决方式：悲观锁，写串行化：

// SELECT ... FOR UPDATE 锁定会话行
public Long lockActiveConversationAndGetLastSeq(Long conversationId) {
    List<Long> rows = jdbcTemplate.query("""
        SELECT last_message_seq
        FROM im_conversation
        WHERE id = ?
          AND status = 'ACTIVE'
          AND is_deleted = 0
        FOR UPDATE
        """,
        (rs, rowNum) -> rs.getLong("last_message_seq"),
        conversationId
    );
    return rows.isEmpty() ? null : rows.get(0);
}

这里我选 SELECT ... FOR UPDATE 而不是乐观锁（CAS），原因很简单：

同一个会话的消息写入本身就是低频操作——人打字再快也就每秒几条，锁持有时间极短（毫秒级）
乐观锁失败需要重试——对用户体验来说，"发消息偶尔失败需要重发"比"感觉不到锁的存在"更差
同一个会话的写入天然就是串行化的——一个人同时打两句话，也应该是先发先到

FOR UPDATE 的粒度是会话级，A 给甲发消息和 B 给乙发消息互不阻塞，只有同一会话内的并发写入才会排队。

七、为什么要用 JDBC 而不是 JPA/MyBatis？

可能有人注意到，整个 IM 模块的数据访问层全部是裸 JDBC（JdbcTemplate + 手写 SQL）。这跟项目里其他模块用 MyBatis 的风格不太一样。

不是我在搞技术选型分裂——这是有意为之的：

IM 消息写入是一个极高频、极简单、极其确定的操作。每条消息就是一条 INSERT + 一条 UPDATE（会话的最后消息游标）。MyBatis 的 XML 映射和 JPA 的实体管理在这种场景下都是额外开销，而手写 SQL 可以精确控制：

每行 SQL 都清楚知道自己跑了什么
SELECT ... FOR UPDATE 直接写在 SQL 里，不会被 ORM 包装
批量查询（对话列表含最后一条消息的 JOIN）用 SQL 比 Java 里循环查询快得多

但要承认：手写 JDBC 的维护成本确实更高，尤其是项目人多了之后。如果哪天 IM 模块的查询复杂度暴涨，再考虑切 MyBatis 也不迟。

八、总结

这篇文章覆盖了 IM 消息存储的六个关键技术点：

问题	方案	理由
消息顺序	会话级 `seq_no` + 唯一约束	严格单调，独立于分布式 ID
翻页性能	游标分页（`seq_no < ?`）	避免 offset 扫描退化
重试去重	`client_message_id` 幂等键	应用层 + 数据库双重保障
多类型消息	`content_json` JSON 列	灵活扩展，不频繁改表
用户级删除	`im_message_user_action` 独立表	一人删除不影响他人
并发写入	`SELECT ... FOR UPDATE` 悲观锁	会话级串行化，粒度小性能足够

做 IM，消息存储是真正考验"想清楚再动手"的地方。表结构一旦定了就很难改，因为数据量上来之后 ALTER TABLE 是噩梦。

flowchart TD subgraph write["消息写入路径"] A["客户端发送消息\n携带 clientMessageId"] --> B{"幂等检查\n是否已存在?"} B -->|"是"| C["返回已有消息\n更新已读位"] B -->|"否"| D["SELECT FOR UPDATE\n锁会话获取 lastSeq"] D --> E["seq = lastSeq + 1\nINSERT 消息"] E --> F["UPDATE 会话\nlastMessageSeq"] end subgraph read["消息拉取路径"] G["客户端请求\n携带 beforeSeq"] --> H["WHERE seq < beforeSeq\nORDER BY seq DESC"] H --> I["NOT EXISTS 过滤\n用户已删除消息"] I --> J["返回消息列表"] end C --> K["WebSocket\n推送到会话成员"] F --> K style C fill:#ffd43b,stroke:#fab005,color:#000 style K fill:#339af0,stroke:#1c7ed6,color:#fff

现在回头看，这套方案在 Workforce Hub 的实际运行中表现稳定，没踩到什么需要推翻重来的坑。

一、建表设计——消息表长什么样

1.1 为什么用独立的 `seq_no` 而不是自增 ID？

1.2 发送人为什么拆成 `sender_type` + `sender_id`？

1.3 `content_json` 为什么是 JSON 而不是固定字段？

二、游标分页——为什么不直接用 offset？

三、客户端幂等——重试不发重复消息

四、消息编辑与撤回——状态流转

五、用户级软删除——每个人都有"删除"的权利

六、并发安全——一个会话只能有一个人"动笔"

七、为什么要用 JDBC 而不是 JPA/MyBatis？

八、总结

切换评论系统

热门文章

一、建表设计——消息表长什么样

1.1 为什么用独立的 seq_no 而不是自增 ID？

1.2 发送人为什么拆成 sender_type + sender_id？

1.3 content_json 为什么是 JSON 而不是固定字段？

二、游标分页——为什么不直接用 offset？

三、客户端幂等——重试不发重复消息

四、消息编辑与撤回——状态流转

五、用户级软删除——每个人都有"删除"的权利

六、并发安全——一个会话只能有一个人"动笔"

七、为什么要用 JDBC 而不是 JPA/MyBatis？

八、总结

切换评论系统

热门文章

1.1 为什么用独立的 `seq_no` 而不是自增 ID？

1.2 发送人为什么拆成 `sender_type` + `sender_id`？

1.3 `content_json` 为什么是 JSON 而不是固定字段？