可靠性落地：ACK、重发与去重

长连接会断、进程会重启、弱网下 ACK 会丢。IM 的可靠性要求可以一句话概括：不丢、不重、按会话有序。做到这三点靠的不是“发得可靠”，而是“发送本身不保证，但每条消息都有确认与重试，且任何重复都能被识别”。第 07 篇给了投递链路，本章补上确认、重发与去重三件套。

本章目标

分清三个方向的 ACK（上行确认、下行确认、失败兜底）各自解决什么问题；
落地服务端重发队列（Redis ZSET + 退避重试）与消息去重（布隆 + Redis SETNX + 去重表）；
明确 ACK 与落库的先后顺序，避免“回了 ACK 但消息没了”。

语义选择：至少一次 + 幂等

至多一次（at-most-once）发出去就不管，断线即丢消息，聊天不可接受；恰好一次（exactly-once）需要端到端事务或两阶段提交，宏观代价极高、工程上不划算；IM 只能选至少一次（at-least-once），然后靠幂等把重复消化掉。落地公式就是：至少一次投递 + 全链路幂等 = 用户感知的恰好一次。消息的幂等键是服务端 msg_id，上行请求的幂等键是客户端生成的 client_msg_id（重连重试必须复用同一个）。

三个 ACK 与落库顺序

客户端 A ──UpstreamMsg(client_msg_id)──▶ gateway ──gRPC──▶ logic
                                                            │ 1. 去重（client_msg_id）
                                                            │ 2. 同事务：conv_seq 取号 + 落 msg 表
        ◀──UpstreamAck(msg_id, seq)─────────────────────────┤ 3. 写扩散 offline_msg（第 09 篇）
        ◀──DownstreamMsg(msg_id, seq)───────────────────────┘ 4. 落库成功后才回 ACK 并投递
接收方 ──AckMsg(msg_id)──▶ gateway ──▶ logic 5. 清重发队列 + 推进设备位点

三条纪律：先落库、后 ACK（ACK 一旦回给发送方，消息就必须可被重放；改成异步落库再回 ACK，就会出现“发送方显示成功、接收方永远收不到”）；取号与落库同事务（避免崩溃后序号被消耗而消息不存在）；ACK 只代表“到达对端连接”，不代表已读——已读是另一条业务指令（第 11 篇的 read_seq）。

重发队列

下行消息发出后进入待确认集合 ack:pending:{gateway_id}：ZSET，score 是下次重发时刻的 Unix 毫秒，member 是 {"msg_id","uid","conn_id","conv_id","seq","attempts","next_retry_ms"}。客户端回 ACK 才移除；集合必须放在进程外（Redis），否则网关重启时正在重试的消息全丢。

// backoff 是各次重发前的等待间隔；累计尝试 4 次仍无 ACK 就放弃并转离线拉取兜底，绝不无限重发
var backoff = []time.Duration{time.Second, 3 * time.Second, 7 * time.Second, 15 * time.Second}

func (m *Manager) Register(ctx context.Context, p *Pending) error {
	p.Attempts, p.NextRetry = 1, time.Now().Add(backoff[0]).UnixMilli()
	raw, _ := json.Marshal(p) // 先算好重发时刻再序列化，member 与 score 才是同一份状态
	return m.rdb.ZAdd(ctx, m.pendingKey(), redis.Z{Score: float64(p.NextRetry), Member: string(raw)}).Err()
}

// Scan 由单个 goroutine 每 200ms 调一次：取到期元素，未超次数则按退避改期后放回，超次数则放弃
func (m *Manager) Scan(ctx context.Context) ([]*Pending, error) {
	now := strconv.FormatInt(time.Now().UnixMilli(), 10)
	raws, err := m.rdb.ZRangeByScore(ctx, m.pendingKey(), &redis.ZRangeBy{Min: "-inf", Max: now, Count: 200}).Result()
	if err != nil {
		return nil, err
	}
	due := make([]*Pending, 0, len(raws))
	for _, raw := range raws {
		var p Pending
		m.rdb.ZRem(ctx, m.pendingKey(), raw) // 先摘出来，改期后再按新 score 放回
		if json.Unmarshal([]byte(raw), &p) != nil || p.Attempts >= int32(len(backoff)) {
			metrics.DeliverFailed.WithLabelValues("ack_timeout").Inc() // 脏数据或超次数：放弃，交给重连拉取兜底
			continue
		}
		p.Attempts++
		p.NextRetry = time.Now().Add(backoff[p.Attempts-1]).UnixMilli()
		nraw, _ := json.Marshal(p)
		m.rdb.ZAdd(ctx, m.pendingKey(), redis.Z{Score: float64(p.NextRetry), Member: string(nraw)})
		due = append(due, &p)
	}
	return due, nil
}

重发必须有次数上限：客户端离线后重发毫无意义，越重越多只会把网关写队列塞满；放弃后由离线拉取负责，用户重连照样能拿到——重发队列是“加速器”，收件箱才是“保底”。ACK 处理要保持幂等：在队列里找不到对应成员说明已 ACK 或已放弃，直接返回成功，不报错也不重复推进位点。

去重

上行重复有三个来源：客户端超时重试、网关重试转发、用户连点。按“从快到慢”分三层挡：客户端把未确认的 client_msg_id 原样复用（零成本，从源头减少重复）；逻辑层用布隆过滤器（Bloom filter）判“一定没发过”，命中“可能存在”时再用 Redis SETNX dedup:{uid}:{client_msg_id}（TTL 10 分钟）做跨实例精确判重；最后落到一张带唯一键的 msg_dedup 表，Redis 掉线时由它兜底，并能查回原 msg_id。

布隆过滤器只能用来否定：判“不存在”就一定没发过，可直接放行；判“可能存在”必须回查 Redis 或库表，绝不允许据此返回“重复”——误判会把用户的新消息吞掉，是数据丢失级事故。误判率按 1% 配（100 万元素约 1.2 MB 位图），本地布隆 + 定时从 msg_dedup 重建。

CREATE TABLE msg_dedup (
  uid           BIGINT UNSIGNED NOT NULL COMMENT '发送者 uid',
  client_msg_id VARCHAR(64)     NOT NULL COMMENT '客户端生成的幂等键（UUID/雪花）',
  msg_id        BIGINT UNSIGNED NOT NULL COMMENT '服务端消息 ID',
  conv_id       BIGINT UNSIGNED NOT NULL,
  seq           BIGINT UNSIGNED NOT NULL COMMENT '会话内序号',
  created_at    DATETIME(3)     NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  PRIMARY KEY (uid, client_msg_id),
  KEY idx_created (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='上行去重表';

// 去重命中要返回原 msg_id 与 seq，客户端据此认为“发送成功”，才不会在新旧两条之间反复
func (s *SendService) dedup(ctx context.Context, uid int64, clientMsgID string) (*DedupHit, error) {
	ok, err := s.rdb.SetNX(ctx, fmt.Sprintf("dedup:%d:%s", uid, clientMsgID), "1", 10*time.Minute).Result()
	if err != nil {
		return nil, err // Redis 故障时不判重，转由唯一键兜底：宁可重复一次，也不丢消息
	}
	if ok {
		return nil, nil // 首次发送
	}
	var d DedupHit
	err = s.db.QueryRowContext(ctx, `SELECT msg_id, conv_id, seq FROM msg_dedup WHERE uid = ? AND client_msg_id = ?`,
		uid, clientMsgID).Scan(&d.MsgID, &d.ConvID, &d.Seq)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil // 布隆/SETNX 的假阳性，按首次发送处理
	}
	metrics.DedupHit.WithLabelValues("client_msg_id").Inc()
	return &d, err
}

msg_dedup 只保留 3 天（重复都在分钟级内发生），每天 DELETE ... WHERE created_at < DATE_SUB(NOW(), INTERVAL 3 DAY) LIMIT 5000 循环清理；Redis 键靠 TTL 过期，无需任务。

验证方式

# 1) 待确认队列：稳态只留几十条（正在等 ACK），score 是毫秒时间戳并逐步前进；堆积上千条说明 ACK 路径有问题
redis-cli ZCARD ack:pending:gw-1 && redis-cli ZRANGE ack:pending:gw-1 0 2 WITHSCORES

# 2) 重复消息检查（期望 0 行）
mysql -uroot -p im -e "SELECT conv_id, seq, COUNT(*) c FROM msg_07 GROUP BY conv_id, seq HAVING c > 1;"

# 3) 去重：同 client_msg_id 连发两次，响应里的 msg_id / seq 必须一致，im_dedup_hit_total 期望 +1
curl -sS -X POST http://127.0.0.1:8080/api/msg/send -H 'Content-Type: application/json' \
  -d '{"conv_id":88001,"client_msg_id":"c-1a2b3c","text":"hi"}'

# 4) 故障演练：kill -9 网关重启后，ack:pending 里的消息应在 15 秒内由离线拉取补齐

常见坑

ACK 超时设得太短：弱网 RTT 抖动到 2 秒很常见，超时设 1 秒会造成整网重发；下行建议 3 秒起、退避到 15 秒。
重发无上限：客户端离线两天，队列每 15 秒重发一次，网关被自己的重试流量打满；必须有次数上限并转离线。
把布隆当精确判重：误判成“重复”就吞掉用户的新消息，属于丢数据事故。
ACK 后不推进位点：消息投到了但位点没动，下次重连再拉一遍，用户看到重复消息。
幂等键换新：客户端重连后用新的 client_msg_id 重试，服务端只能当新消息，重复去不掉；未确认的上行必须原样重发同一个键。
重发与离线双投 / 去重表无限增长：重发成功、客户端又刚好重连拉到同一条，靠 msg_id 在客户端去重即可，不要加分布式锁；msg_dedup 只增不删会让主键索引比数据还大，必须按 3 天窗口清理。

小结：IM 的可靠性等于“至少一次投递 + 全链路幂等”：上行以 client_msg_id 为幂等键（布隆否定 + Redis SETNX + msg_dedup 唯一键三层），下行以 msg_id 为幂等键并放进 Redis 待确认 ZSET 按 1/3/7/15 秒退避重发，超过 4 次就放弃交给离线拉取兜底；纪律是“先落库后 ACK”“取号与落库同事务”“ACK 同时推进设备位点”，守住这三点，用户看到的就是恰好一次。

笔记加载中…

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

返回知识库