pro/im_server

Fork 0

Files

T

richjohnrambo 139681fadb feat: shard group messages by month

2026-09-15 23:53:51 +08:00

21 KiB

Raw Blame History

IM 群消息按月分表改造方案

1. 背景与目标

当前每个租户库使用单表 im_group_messages 保存全部群消息。随着群数量、消息量和保留时间增长，单表会出现索引膨胀、查询抖动、DDL 风险升高和归档困难等问题。

本方案将群消息正文按自然月拆分为物理表：

im_group_messages_202609
im_group_messages_202610
im_group_messages_202611

目标：

每个租户库独立分表，消息表不增加 tenant_id。
Topic 的 seqid 继续由 Redis 原子递增，跨月不重置。
Kafka 异步落库稳定路由到首次确定的月表。
客户端继续按 topic + seq 拉取，不感知月份。
支持跨月分页、补拉、指定消息、搜索、撤回、编辑和删除。
旧单表平滑迁移，不停服、不丢失、不重复。
Redis 保留 24 小时热消息，MySQL 月表仍是最终数据源。

2. 核心原则

2.1 分表粒度

一个租户库每月一张群消息表，不按单个群建表。
表名固定为 im_group_messages_YYYYMM。
月份统一使用 UTC，但最终写入月份由 Redis 全局 active_month 决定，避免多节点在月初产生交叠。
Redis Lua 一次性返回 seqid + shard_month；生产端把结果固化到 Kafka，重试、补偿不得重算。

2.2 Seq 规则

seqid 在 Topic 内全生命周期单调递增，跨月不归零。
Redis topic:last_seq 是唯一发号源。
月表只保存已分配的 Seq，不通过 MAX(seqid) 发号。
客户端游标继续使用 before/since/seq。

2.3 分层职责

层级	职责
Redis	原子 Seq、全局活动月份、长期 Topic 月段路由、短期发送幂等、24 小时热消息、在线状态、实时推送
Kafka	异步落库、削峰、重试、死信、消费幂等
MySQL 月表	群消息正文最终存储
MySQL 路由镜像	月表目录、Topic 月段路由持久化镜像；Redis 丢失时用于重建，不进入正常查询热路径
`im_dellog`	用户视角删除范围，每张月表查询时继续过滤

3. 表结构

3.1 月表模板

第一阶段保持当前字段兼容，降低 Store 层改造范围。

CREATE TABLE IF NOT EXISTS im_group_messages_202609 (
  message_id BIGINT NOT NULL COMMENT 'Snowflake 全局消息 ID',
  topic VARCHAR(64) NOT NULL COMMENT '群 Topic',
  group_id BIGINT NOT NULL COMMENT '业务群 ID',
  seqid BIGINT NOT NULL COMMENT 'Topic 内递增，跨月不重置',
  from_user BIGINT NOT NULL,
  client_msg_id VARCHAR(64) NULL,
  msg_type VARCHAR(32) NOT NULL DEFAULT 'text',
  status VARCHAR(32) NOT NULL DEFAULT 'normal',
  content JSON NULL,
  head JSON NULL,
  mentioned_user_ids JSON NULL,
  mention_all TINYINT(1) NOT NULL DEFAULT 0,
  reply_seqid BIGINT NULL,
  root_seqid BIGINT NULL,
  file_count INT NOT NULL DEFAULT 0,
  content_size INT NOT NULL DEFAULT 0,
  has_risk TINYINT(1) NOT NULL DEFAULT 0,
  audit_status VARCHAR(32) NOT NULL DEFAULT 'pass',
  version INT NOT NULL DEFAULT 1,
  recalled_at DATETIME(3) NULL,
  recalled_by BIGINT NULL,
  deleted_at DATETIME(3) NULL,
  expires_at DATETIME(3) NULL,
  created_at DATETIME(3) NOT NULL,
  updated_at DATETIME(3) NOT NULL,
  PRIMARY KEY (message_id),
  UNIQUE KEY uk_topic_seq (topic, seqid),
  UNIQUE KEY uk_topic_sender_client_mid (topic, from_user, client_msg_id),
  KEY idx_group_seq (group_id, seqid),
  KEY idx_topic_created (topic, created_at),
  KEY idx_topic_from_seq (topic, from_user, seqid),
  KEY idx_audit_created (audit_status, created_at),
  KEY idx_expires (expires_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

说明：

message_id 使用 Snowflake 全局唯一 ID，作为月表物理主键和跨表引用 ID。
聊天协议定位、补拉和已读水位仍统一使用 topic + seqid，不向客户端暴露物理主键。
单表唯一索引不能覆盖跨月重试，跨月幂等由 Redis、消费日志和幂等表共同保证。
文件二进制仍在 OSS，月表仅保存消息和文件元数据。

3.2 月表目录

CREATE TABLE IF NOT EXISTS im_group_message_shards (
  shard_month CHAR(6) NOT NULL COMMENT 'UTC 月份 YYYYMM',
  table_name VARCHAR(64) NOT NULL,
  shard_start DATETIME(3) NOT NULL,
  shard_end DATETIME(3) NOT NULL,
  status VARCHAR(16) NOT NULL DEFAULT 'prepared'
    COMMENT 'prepared/writable/sealing/sealed/archived',
  row_count BIGINT NOT NULL DEFAULT 0,
  min_created_at DATETIME(3) NULL,
  max_created_at DATETIME(3) NULL,
  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3)
    ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (shard_month),
  UNIQUE KEY uk_table_name (table_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

表名只允许由服务端根据月份生成，并且必须存在于目录表，禁止使用请求参数直接拼接 SQL。

3.3 Redis Topic 月段路由

在线查询使用 Redis 路由，路由 Key 长期不过期：

# 当前可写月份
im:{tenant_id}:groupmsg:active_month
  STRING = 202610

# Topic 月份段，score 为该月首条 seq
im:{tenant_id}:groupmsg:route:{topic}
  ZSET: 202609 => 1
        202610 => 10087

# 月段完整范围，field 为 YYYYMM
im:{tenant_id}:groupmsg:route_meta:{topic}
  HASH: 202609 => min_seq=1,max_seq=10086,count=10086
        202610 => min_seq=10087,max_seq=12000,count=1914

路由规则：

route 和 route_meta 不设置 TTL，只在群物理删除或对应消息保留期结束后清理。
精确 Seq 定位使用 ZREVRANGEBYSCORE route seq -inf LIMIT 0 1，找到最大 min_seq <= seq 的月份，再校验 Hash 中的 max_seq。
倒序分页通过 ZREVRANGE 按月段倒序读取，正向补拉通过 ZRANGE 读取。
路由 Key 使用 Redis Cluster hash tag，保证 Lua 和 Pipeline 访问同一槽位。
Redis 开启 AOF everysec、主从复制和持久化备份；同时保留 MySQL 镜像作为灾备重建来源。

活动月份和 Seq 必须在同一个 Lua 脚本中读取/分配，两个 Key 使用相同的租户 hash tag：

local shard_month = redis.call('GET', KEYS[1])
if not shard_month or shard_month < ARGV[1] then
  redis.call('SET', KEYS[1], ARGV[1])
  shard_month = ARGV[1]
end

local seq_id = redis.call('INCR', KEYS[2])
return {seq_id, shard_month}

其中 KEYS[1] 是租户活动月份，KEYS[2] 是 Topic Seq，ARGV[1] 来自 Redis TIME 转换后的 UTC 月份。业务节点不得根据本机时间计算月份；首次写入可初始化活动月份，之后 Lua 只允许月份单向推进，旧节点不能回拨。

3.4 MySQL 路由镜像

CREATE TABLE IF NOT EXISTS im_group_message_topic_shards (
  topic VARCHAR(64) NOT NULL,
  shard_month CHAR(6) NOT NULL,
  min_seq BIGINT NOT NULL,
  max_seq BIGINT NOT NULL,
  message_count BIGINT NOT NULL DEFAULT 0,
  min_created_at DATETIME(3) NOT NULL,
  max_created_at DATETIME(3) NOT NULL,
  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3)
    ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (topic, shard_month),
  KEY idx_topic_max_seq (topic, max_seq),
  KEY idx_month_topic (shard_month, topic)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

该表不参与正常消息查询，只用于 Kafka 事务持久化、Redis 路由重建、历史迁移、数据核对和归档。Redis 与镜像不一致时，以月表实际数据和 MySQL 镜像修复 Redis。

路由一致性规则：

Kafka 消费事务同时提交消息和 MySQL 路由镜像，保证最终数据与镜像一致。
MySQL 提交成功后再更新 Redis route/route_meta；Redis 更新失败不回滚消息，进入路由修复重试队列。
查询发现 Redis 路由 Key 不存在时，从 MySQL 镜像加载该 Topic 全部有效月段并回填 Redis，然后重试查询。
Shard Manager 启动时可按租户增量重建路由，并定时抽样核对 Redis、MySQL 镜像和月表实际 Seq 范围。
Redis 路由不设置 TTL；只有群被物理删除，或消息保留策略确认清除对应月份后，才同步删除相应月段。

3.5 跨月幂等表

CREATE TABLE IF NOT EXISTS im_group_message_dedup (
  topic VARCHAR(64) NOT NULL,
  from_user BIGINT NOT NULL,
  client_msg_id VARCHAR(64) NOT NULL,
  seqid BIGINT NOT NULL,
  shard_month CHAR(6) NOT NULL,
  message_id BIGINT NULL,
  status VARCHAR(16) NOT NULL DEFAULT 'pending',
  expires_at DATETIME(3) NOT NULL,
  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3)
    ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (topic, from_user, client_msg_id),
  KEY idx_expires (expires_at),
  KEY idx_topic_seq (topic, seqid)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

Redis SET NX 是入口快速幂等。
Kafka 消费日志按 event_id 防止同一事件重复消费。
该表防止月末发送、月初重试时同一 client_msg_id 落入两张表。
建议保留 7 至 30 天后清理，不作为永久消息索引。

4. 建表和月切换

每个租户库运行统一 Shard Manager：

每天巡检上月、当月和未来两个月。
缺表时根据固定模板创建，成功后登记目录。
使用 Redis 分布式锁和 Redis TIME 判断月界，确认新月表可写后再原子更新 active_month。
当月设置为 writable，所有节点通过 Redis 读取同一个活动月份。
上月进入 sealing，建议继续接受 7 天延迟 Kafka 消息。
确认没有积压和死信后改为 sealed。
建表失败立即告警，不切换 active_month，禁止在业务写入事务中临时执行 DDL。

月切换示例：

2026-09-30 23:59:59 UTC seq=10086 -> im_group_messages_202609
2026-10-01 00:00:00 UTC seq=10087 -> im_group_messages_202610

Kafka 事件必须固定路由信息：

{
  "topic": "grp_xxx",
  "seq_id": 10087,
  "created_at": "2026-10-01T00:00:00.123Z",
  "shard_month": "202610",
  "client_msg_id": "client_mid_xxx"
}

5. 群消息落库

5.1 生产端

校验群成员、禁言、群状态和内容策略。
使用 Redis 幂等 Key 抢占 client_msg_id。
使用 Redis Lua 原子获取 Topic 下一个 seqid 和全局 active_month。
生成服务端 UTC created_at，并把 Lua 返回的月份作为 shard_month。
把 shard_month 写入 Kafka；所有重试沿用原值。
Kafka 生产成功后按现有策略向客户端返回 Seq。
写入 Redis 热消息并执行实时推送。

5.2 Kafka 消费端

消费端按 tenant + shard_month 分组批量写入：

校验月份格式和允许范围。
从目录确认目标表已存在。
开启事务。
插入或确认 im_group_message_dedup。
插入目标月表。
在同一事务中 UPSERT MySQL 路由镜像的最小/最大 Seq、时间和消息数。
更新消费幂等状态并提交事务。
提交成功后更新 Redis route/route_meta，并把 Redis 幂等状态改为 done。

路由 UPSERT：

INSERT INTO im_group_message_topic_shards
  (topic, shard_month, min_seq, max_seq, message_count,
   min_created_at, max_created_at)
VALUES (?, ?, ?, ?, 1, ?, ?)
ON DUPLICATE KEY UPDATE
  min_seq = LEAST(min_seq, VALUES(min_seq)),
  max_seq = GREATEST(max_seq, VALUES(max_seq)),
  message_count = message_count + 1,
  min_created_at = LEAST(min_created_at, VALUES(min_created_at)),
  max_created_at = GREATEST(max_created_at, VALUES(max_created_at));

5.3 异常处理

目标月表不存在：消费失败、进入重试并高优告警。
sealing 月表：允许携带原 shard_month 的延迟事件写入。
sealed 月表：进入补偿队列，人工确认后临时开放或补写。
重复 event_id：消费日志直接判定成功。
重复 client_msg_id：返回原 seqid/shard_month，不新增消息。
超过最大重试次数：进入死信队列，不能静默丢弃。
Redis 热缓存失败不回滚 MySQL，客户端按 Seq 回源补拉。
Redis 路由更新失败：进入独立修复重试；查询侧可从 MySQL 镜像重建，不影响已提交消息。

6. 消息查询

6.1 总流程

客户端 Topic Seq 请求
  -> ACL/成员身份校验
  -> Redis 24 小时热消息
  -> Redis 是否完整覆盖请求范围
  -> Redis Topic 月段路由
  -> 月表逐张查询
  -> 按 seqid 合并、去重、排序、截断
  -> 应用 im_dellog 用户视角过滤
  -> 返回客户端

Redis 只有完整覆盖整页时才能直接返回；有缺口必须回源 MySQL。

6.2 最新消息倒序分页

请求 topic=grp_xxx&before=10088&limit=50：

从 Redis Topic 月段路由找出 min_seq < before 的月份，按月倒序。
先查最新候选月：

SELECT ...
FROM im_group_messages_202610 AS m
LEFT JOIN im_dellog AS d
  ON d.topic = m.topic
 AND m.seqid BETWEEN d.low AND d.hi - 1
 AND d.deletedfor = ?
WHERE m.topic = ?
  AND m.seqid < ?
  AND m.status != 'deleted'
  AND d.deletedfor IS NULL
ORDER BY m.seqid DESC
LIMIT ?;

数量不足时继续查前一个月。
服务层按 seqid DESC 合并并截取 limit。
下一页游标仍是本页最小 seqid，客户端不感知月份。

禁止对所有历史表执行无上限 UNION ALL。

6.3 Since 正向补拉

从 Redis Topic 月段路由查找 max_seq > since_seq 的月份。
按月升序执行 seqid > since_seq ORDER BY seqid ASC LIMIT N。
合并后按 Seq 升序返回。
达到上限后返回最后一条 Seq，客户端继续请求。

6.4 指定消息

撤回、回复预览、举报取证按 topic + seqid 定位：

ZREVRANGEBYSCORE im:{tenant_id}:groupmsg:route:{topic} {seqid} -inf LIMIT 0 1
HGET im:{tenant_id}:groupmsg:route_meta:{topic} {shard_month}

先找到最大 min_seq <= seqid 的月份，再校验该月段 max_seq >= seqid，随后执行 WHERE topic=? AND seqid=? LIMIT 1。Redis 路由 Key 缺失时读取 MySQL 镜像并回填 Redis；镜像也缺失时，仅降级扫描当前月和最近若干月并记录修复任务，不允许扫描全部历史表。

6.5 上下文和范围

查询目标消息前后各 20 条时：

定位目标所在月。
在本月分别查询目标前后消息。
任一方向不足时继续查询相邻月份。
合并、去重后按 Seq 正序返回。

6.6 搜索和统计

短期搜索必须带起止时间，最多覆盖 3 个月，逐月查询后归并。
中长期通过 Kafka 建独立搜索索引；结果只返回 topic + seqid，服务端回源月表并重新校验 ACL。
Admin 统计使用按日/月汇总表，不实时扫描所有月表执行 COUNT(*)。

7. 撤回、编辑和删除

撤回、编辑流程：

根据 topic + seqid 定位目标月表。
在目标月表事务中 SELECT ... FOR UPDATE。
校验版本、权限和状态。
更新目标月表并写修订记录、业务事件。
提交后更新或删除 Redis 热消息。

修订记录建议增加月份：

ALTER TABLE im_message_revisions
  ADD COLUMN shard_month CHAR(6) NULL AFTER topic;

用户侧删除继续使用 im_dellog 的 Topic Seq 范围，每张月表查询都应用相同过滤。全局删除或合规处置先定位月表，再更新 status/deleted_at/content/head；OSS 附件仍走文件生命周期任务。

8. Store 层设计

所有动态表访问统一经过路由组件：

type GroupMessageShardRouter interface {
    WriteTable(createdAt time.Time, shardMonth string) (string, error)
    CandidateTablesBySeq(topic string, seq int64) ([]string, error)
    TablesBefore(topic string, before int64) ([]string, error)
    TablesSince(topic string, since int64) ([]string, error)
}

改造范围：

TenantGroupMessageSave：按事件 shard_month 写入。
TenantGroupMessageGetOne：路由后查询目标月。
TenantGroupMessageGetAll：逐月查询并归并。
TenantGroupMessageGetByClientID：先查 Redis/幂等表，再查询目标月。
TenantGroupMessageRecall/Edit：定位目标月后执行事务。
TenantMessageSearch：按时间选月，或转独立搜索。
Admin 内部消息查询、举报、收藏、回复和群文件引用同步改造。

动态表名必须满足 ^im_group_messages_[0-9]{6}$ 且月份已登记。SQL 参数不能绑定表名，必须严格白名单。

9. 旧表迁移

原 im_group_messages 先保留，不删除。

推荐步骤：

上线路由表、幂等表和 Shard Manager。
预建历史月份、当月和未来两个月表。
先发布支持“旧表 + 月表”双读的代码，写入仍走旧表。
记录切换水位 cutover_seq/cutover_time。
切换新消息只写月表，Kafka 开始携带 shard_month。
按 created_at 分批把旧表数据迁入对应月表。
每批使用 INSERT IGNORE，校验 topic + seqid、数量和内容哈希。
回填 MySQL Topic 月段路由镜像，并由重建任务写入 Redis 路由。
双读期间按 topic + seqid 去重。
全量校验通过后停止读取旧表。
旧表改名 im_group_messages_legacy，只读保留。
观察至少一个完整保留周期后再单独审批清理。

不建议长期双写旧表和月表，否则两个写目标部分成功会扩大一致性问题。

迁移批次要求：

使用主键或 created_at + id 游标，禁止 OFFSET 深分页。
单批 1,000 至 5,000 条，根据线上延迟调整。
每批独立事务，记录开始/结束游标和成功/失败数。
支持暂停、续跑和重复执行。

10. 归档与保留

状态流转：

prepared -> writable -> sealing -> sealed -> archived

writable：当月正常写。
sealing：月末后接受延迟消息，默认 7 天。
sealed：在线 MySQL 只读。
archived：转对象存储或低成本库，在线查询走归档服务。

归档前确认 Kafka 无积压/死信、路由完整、消息数量和 Seq 范围一致，并确认举报、审计、收藏和群文件引用仍可定位。

11. 监控与告警

指标	说明
`group_message_persist_total{month,status}`	各月落库成功/失败
`group_message_persist_latency`	Kafka 到 MySQL 提交耗时
`group_message_shard_missing_total`	目标月表不存在
`group_message_query_tables`	单次请求扫描表数
`group_message_query_latency{source}`	Redis、单月、跨月耗时
`group_message_dedup_hit_total`	幂等命中
`group_message_route_miss_total`	Redis 路由缺失
`group_message_route_rebuild_total{status}`	从 MySQL 镜像重建 Redis 路由
`group_message_route_repair_backlog`	Redis 路由修复积压
`group_message_dead_letter_total`	落库死信

未来月表未创建、当月不可写、active_month 缺失、Kafka 延迟超限、单次扫描超过 3 张表、Redis 路由缺失或修复积压增长时必须告警。

12. 验收测试

月末和月初消息落入正确表，Seq 连续。
Kafka 次月才消费时仍写事件指定月份。
同一 client_msg_id 跨月重试只保留一条。
最新 50 条跨两个月时数量、顺序、游标正确。
since_seq 跨月补拉无遗漏、无重复。
topic + seq 可定位历史消息。
回复上下文可跨月读取。
撤回、编辑、举报、收藏、群文件引用可定位旧月。
用户删除范围跨月时过滤正确。
Redis 热消息不完整时正确回源。
月表缺失时进入重试/死信并告警。
迁移双读无重复，迁移任务可暂停、续跑。
万人群高并发时 Seq 唯一且落库稳定。
非法月份和非法动态表名被拒绝。
多节点跨月并发时 Lua 返回同一活动月份，Seq 不重号且月段不交叠。
删除 Redis 路由后可从 MySQL 镜像自动重建，查询结果不缺失。

13. 实施阶段

第一阶段：基础设施

建月表模板、目录、MySQL 路由镜像和跨月幂等表。
实现 Shard Manager，预建未来两个月。
建立长期不过期的 Redis Topic 月段路由和原子 seqid + active_month Lua。
Kafka 群消息事件增加 shard_month。

第二阶段：写入切换

消费者按月落库。
完成幂等、重试、死信、MySQL 镜像和 Redis 路由更新。
Redis 热消息与实时推送保持不变。

第三阶段：查询切换

完成精确查询、倒序分页、Since 补拉和上下文跨月查询。
改造撤回、编辑、删除、举报及 Admin 内部查询。
增加跨月查询监控。

第四阶段：历史迁移

上线旧表/月表双读。
回填历史消息与 MySQL 路由镜像，重建 Redis Topic 路由。
校验后把旧表改为只读保留。

第五阶段：搜索和归档

消息搜索切换独立索引或限制时间范围。
建立 sealed/archived 生命周期和历史恢复机制。

14. 结论

群消息按月分表不能只改写入表名，必须一起实现固定 shard_month、Redis Topic 月段路由、MySQL 路由镜像、跨月幂等、逐月游标查询、历史消息定位和旧表迁移。

最终链路：

发送：Redis Lua 原子分配 Seq 和 active_month -> Kafka 固定 shard_month -> MySQL 月表
实时：Redis Pub/Sub/Stream -> WebSocket
热读：Redis 24 小时消息
冷读：Redis Topic 月段路由 -> MySQL 月表逐月查询
容灾：MySQL 路由镜像 -> 重建长期 Redis 路由
补偿：客户端按 Topic Seq 拉取

客户端协议无需增加月份字段，月表路由完全由服务端处理。

21 KiB Raw Blame History Unescape Escape