离线消息与多端同步落地

用户进电梯断网三分钟，出来打开 App，这三分钟里的 12 条消息必须一条不少地出现；同一账号的手机、桌面、Web 三个端，未读数还要各自独立推进——“谁还没收到”服务端必须能精确回答。第 07 篇解决“在线怎么投”，本章解决“不在线怎么办、多端怎么对齐”。

本章目标

用**离线收件箱（offline_msg）+ 会话序列号（conv_seq）+ 设备同步位点（user_sync_cursor）**三张表描述离线与多端状态；
落地“在线推送 + 重连拉取”的推拉结合协议：可增量、可分页、可重入；
给出清理策略（确认回收 + 分区裁剪）与可执行的验证 SQL。

为什么不用“每人一个离线队列”

给每个用户建一个 Redis List、登录时全量倒出再 DEL，会踩三个坑：多端登录时“倒出即删”只能服务一个端；DEL 之后消息永久消失，客户端落库失败无法重拉；List 元素无序号，回答不了“我拉到第几条了”。正确的心智模型是会话级可重放日志 + 每设备进度位点：消息只写一份，进度由客户端位点决定，重拉即增量、天然幂等、天然多端。

发送者 ──▶ msg 表（会话日志，按 conv_id 分片，第 11 篇）──写扩散──▶ offline_msg(uid, conv_id, seq)
客户端重连 ─ Pull(cursors[{conv_id, seq}]) ─▶ 增量补齐位点之后的缺口 ─▶ 收到帧后 ACK，推进 user_sync_cursor

表结构

序号是整个 IM 有序性的根，只能由 conv_seq 原子分配，不能用时间戳或 MySQL 自增 ID：

CREATE TABLE conv_seq (
  conv_id    BIGINT UNSIGNED NOT NULL COMMENT '会话 ID',
  seq        BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '已分配的最大序号',
  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (conv_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='会话序列号';

-- 原子取号：一条语句完成“自增 + 读回”，不需要 SELECT ... FOR UPDATE
UPDATE conv_seq SET seq = LAST_INSERT_ID(seq + 1) WHERE conv_id = ?;
SELECT LAST_INSERT_ID();

收件箱按 uid % 16 分表（offline_msg_00 ~ offline_msg_15），主键把 uid 放最左，保证拉取永远是单分片查询：

CREATE TABLE offline_msg_00 (
  uid        BIGINT UNSIGNED  NOT NULL COMMENT '接收者 uid（分片键）',
  conv_id    BIGINT UNSIGNED  NOT NULL COMMENT '会话 ID',
  seq        BIGINT UNSIGNED  NOT NULL COMMENT '会话内序号，来自 conv_seq',
  msg_id     BIGINT UNSIGNED  NOT NULL COMMENT '全局消息 ID（雪花）',
  sender_uid BIGINT UNSIGNED  NOT NULL,
  msg_type   TINYINT UNSIGNED NOT NULL DEFAULT 1 COMMENT '1=文本 2=图片 3=撤回 4=系统',
  payload    VARBINARY(2048)  NOT NULL COMMENT 'zstd 压缩后的消息体',
  created_at DATETIME(3)      NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  PRIMARY KEY (uid, conv_id, seq),
  UNIQUE KEY uk_msg_id (uid, msg_id),
  KEY idx_created (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='离线收件箱分片 00';

PRIMARY KEY (uid, conv_id, seq) 是为拉取而生：WHERE uid = ? AND conv_id = ? AND seq > ? ORDER BY seq 一次范围扫描，顺序与索引一致无 filesort；uk_msg_id 让写扩散重试天然幂等。位点必须按设备存，这是多端各自推进的关键：

CREATE TABLE user_sync_cursor (
  uid        BIGINT UNSIGNED NOT NULL,
  device_id  VARCHAR(64)     NOT NULL COMMENT '设备标识（第 04 篇登录时下发）',
  conv_id    BIGINT UNSIGNED NOT NULL,
  ack_seq    BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT '该设备已确认的最大连续序号',
  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (uid, device_id, conv_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='设备同步位点';

拉取实现

重连时客户端带上各会话位点，服务端只补缺口；位点只允许前进，用 GREATEST 兜住乱序到达的重复 ACK。PullReq.Cursors 是 [{conv_id, seq}]，seq 表示客户端本地最大的连续序号：

// Pull 按会话补齐缺口；Next 是本次拉取后客户端可前进到的位点。
func (s *OfflineStore) Pull(ctx context.Context, req PullReq) (*PullResp, error) {
	limit := req.Limit
	if limit <= 0 || limit > 200 {
		limit = 200 // 分页上限，防止老会话积压把网关内存打爆
	}
	resp := &PullResp{}
	for _, c := range req.Cursors {
		tbl := fmt.Sprintf("offline_msg_%02d", uint64(req.UID)%16) // 分片号来自取模，不接受外部输入
		q := fmt.Sprintf(`SELECT seq, msg_id, sender_uid, msg_type, payload, created_at
		                  FROM %s WHERE uid = ? AND conv_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?`, tbl)
		rows, err := s.db.QueryContext(ctx, q, req.UID, c.ConvID, c.Seq, limit)
		if err != nil {
			return nil, fmt.Errorf("pull uid=%d conv=%d: %w", req.UID, c.ConvID, err)
		}
		// 这里逐行 Scan + zstd 解压后 append 到 resp.Msgs；
		// 拉满一页就把位点推进到本页最后一条，剩下的下次继续
		rows.Close()
		resp.Next = append(resp.Next, c)
	}
	return resp, nil
}

推与拉必须衔接好：网关只能在增量订阅成功之后才允许推进位点，否则会出现“推送还没订阅、位点已经前进”的窗口，那条消息永远丢。固定顺序是：先拉取补齐 → 再订阅增量 → 收到帧后 ACK 推进位点。

清理策略

收件箱只服务“还没确认”的消息，确认后必须回收，否则它会长成全库最大的表，三层控制：确认回收（删除 seq <= 该会话所有设备最小 ack_seq 的行，保留 7 天安全期，避免位点上报告失败后无法重拉）、分区裁剪（按 created_at 建 RANGE 分区后 DROP PARTITION，秒级删除且不产生碎片）、兜底 TTL（超过 30 天仍未确认的行直接清理，长期不登录的设备不该拖住整个表）。

-- 每天 03:00 执行，循环直到 affected rows = 0，避免一次删几百万行造成主从延迟
DELETE FROM offline_msg_07 WHERE created_at < DATE_SUB(NOW(), INTERVAL 30 DAY) LIMIT 5000;

一个 MySQL 硬约束：分区表的所有唯一索引必须包含分区列。uk_msg_id 不含 created_at，所以 offline_msg 不能直接分区——要么放弃唯一键（重复写扩散改由应用层保证），要么按时段建表（offline_msg_00_202501）后 DROP TABLE。

验证方式

-- 1) 某设备在某会话还差多少条：期望只列出仍有缺口的会话，且 pending 与客户端红点数一致
SELECT o.conv_id, MAX(o.seq) AS server_max, COALESCE(c.ack_seq, 0) AS device_ack, COUNT(*) AS pending
FROM offline_msg_03 o
LEFT JOIN user_sync_cursor c ON c.uid = o.uid AND c.conv_id = o.conv_id AND c.device_id = 'ios-9f2c'
WHERE o.uid = 1001 AND o.seq > COALESCE(c.ack_seq, 0) GROUP BY o.conv_id, device_ack;

-- 2) 拉取走索引无 filesort：期望 type=range、key=PRIMARY、rows 约等于返回条数
EXPLAIN SELECT seq, msg_id, payload FROM offline_msg_03
WHERE uid = 1001 AND conv_id = 88001 AND seq > 1200 ORDER BY seq ASC LIMIT 50;

-- 3) 行为验证：断网客户端连发 20 条，恢复后应恰好收到 20 条且 seq 连续；同一设备重复 Pull 两次集合完全一致；
--    换一个 device_id 应能拉到该会话至今未被该端确认的全部历史。

常见坑

位点回退：乱序到达的旧 ACK 会把 ack_seq 改小导致重复推送，更新一律用 SET ack_seq = GREATEST(ack_seq, ?)。
位点跳跃：客户端拉到 seq=5 和 seq=7 就推到 7，seq=6 永远不会再拉；位点必须是最大连续前缀，缺号时留在 5 并触发补齐。
只存一个位点：多端共用 conv_id → seq，手机会把桌面的进度顶上去，桌面永远丢消息；必须按 (uid, device_id, conv_id) 分开。
一次拉全量：老会话积压几十万条，一次 Pull 打爆网关内存；必须分页 + 客户端循环拉 + 服务端限流。
收件箱当消息库 / 未读数用 COUNT(*)：用 offline_msg 渲染历史列表，清理后就查不到历史了——历史永远查 msg 表，收件箱只管“没确认的增量”；未读数用 max_seq - read_seq 算（第 11 篇），O(1) 且不必 count。

小结：离线与多端同步的本质是“会话内可重放日志 + 每设备进度位点”：conv_seq 原子取号保证有序不跳号，offline_msg 按 uid 分片做写扩散收件箱并用 (uid, conv_id, seq) 主键支撑单分片增量拉取，user_sync_cursor 按设备记录最大连续位点并用 GREATEST 防回退；顺序固定为“先拉后订阅再推进位点”，清理靠确认回收、分区裁剪与兜底 TTL 三层，并注意分区表唯一索引必须包含分区列。

笔记加载中…

标签：#IM #Go #项目实战

返回知识库