在线路由与 Redis 落地

用户 A 给 B 发消息时，逻辑层手里只有一个 uid，却必须立刻回答「B 此刻挂在哪台网关的哪条连接上」。这个「用户 → 连接」的映射就是在线路由（online routing），它写少读多：登录、登出、心跳才写，每投递一条消息都要读；允许秒级不一致，但读不到就只能走离线兜底，用户体感是「消息发出去了对方没反应」。

本章目标

用 Redis 三张键把路由表落下来，并在网关内实现「Hub 本机连接 + Redis 路由」两级查找（two-level lookup），让跨网关投递与本机直投走同一条代码路径。

Redis 键设计

route:uid:{uid}        Hash    field=deviceID  value=路由 JSON    TTL = 90s
route:conn:{conn_id}   String  value="uid:deviceID"               TTL = 90s
route:gw:{gateway_id}  Set     member="uid:deviceID"              无 TTL，网关重启时批量清理

用 Hash 而不是若干 String，是为了让「一个用户多端在线」退化成一次 HGETALL：手机、桌面、Web 同时在线是常态，若用 String 就得先知道设备列表才能拼出 MGET 的键。TTL 取 90 秒 = 心跳间隔 30 秒 × 3，这个倍数关系是路由能自愈的基础——客户端异常断网时不发 FIN，靠的正是 TTL 到期摘除幽灵路由。

绑定、续约与查找

const routeTTL = 90 * time.Second // 心跳周期 30s × 3，见下文

type Route struct {
	UID       uint64 `json:"uid"`
	DeviceID  string `json:"device_id"`
	GatewayID string `json:"gateway_id"`
	GRPCAddr  string `json:"grpc_addr"` // 跨网关投递时用它拨号
	ConnID    uint64 `json:"conn_id"`   // 网关内自增的连接号（第 02 章）
}

// Bind 在登录 / 重连成功时写入路由，三张键一次 pipeline 发完，省两次往返。
func (s *RouteStore) Bind(ctx context.Context, rt Route) error {
	raw, err := json.Marshal(rt)
	if err != nil {
		return fmt.Errorf("marshal route: %w", err)
	}
	member := fmt.Sprintf("%d:%s", rt.UID, rt.DeviceID)
	uidKey := fmt.Sprintf("route:uid:%d", rt.UID)
	pipe := s.rdb.Pipeline()
	pipe.HSet(ctx, uidKey, rt.DeviceID, raw)
	pipe.Expire(ctx, uidKey, routeTTL)
	pipe.Set(ctx, fmt.Sprintf("route:conn:%d", rt.ConnID), member, routeTTL)
	pipe.SAdd(ctx, "route:gw:"+rt.GatewayID, member)
	if _, err := pipe.Exec(ctx); err != nil {
		return fmt.Errorf("bind route uid=%d conn=%s: %w", rt.UID, rt.ConnID, err)
	}
	return nil
}

// Renew 由心跳触发：只续 TTL 不重写内容，省掉每次心跳的 JSON 序列化。
func (s *RouteStore) Renew(ctx context.Context, rt Route) error {
	pipe := s.rdb.Pipeline()
	pipe.Expire(ctx, fmt.Sprintf("route:uid:%d", rt.UID), routeTTL)
	pipe.Expire(ctx, fmt.Sprintf("route:conn:%d", rt.ConnID), routeTTL)
	if _, err := pipe.Exec(ctx); err != nil {
		return fmt.Errorf("renew route uid=%d conn=%s: %w", rt.UID, rt.ConnID, err)
	}
	return nil
}

// Lookup 查某设备的路由；redis.Nil 表示离线，返回 (nil, nil)，不算错误。
func (s *RouteStore) Lookup(ctx context.Context, uid uint64, deviceID string) (*Route, error) {
	key := fmt.Sprintf("route:uid:%d", uid)
	raw, err := s.rdb.HGet(ctx, key, deviceID).Bytes()
	switch {
	case errors.Is(err, redis.Nil):
		return nil, nil
	case err != nil:
		return nil, fmt.Errorf("hget %s: %w", key, err)
	}
	var rt Route
	if err := json.Unmarshal(raw, &rt); err != nil {
		return nil, fmt.Errorf("bad route payload %s: %w", key, err)
	}
	return &rt, nil
}

心跳续约（heartbeat renew）由每个连接一条 30 秒的 ticker 驱动；续约失败必须主动断开连接，否则会出现「Redis 里没有路由、连接却还活着」的幽灵连接——这类连接收不到任何消息，却照样占着内存与文件描述符。登出与投递失败走的是摘除路径：HDEL 用户 Hash 里的该设备字段、DEL 连接反查键、SREM 网关闭集合成员；路由是缓存性质的数据，摘除失败最多浪费一次无效投递，绝不能让摘除异常阻断主流程。

两级查找：本机 map 优先，Redis 兜底

网关内的 Hub（第 02 章）维护 byUID[uid] → connID → *Client，一级查找就是在这个集合里按 Client.DeviceID 匹配——同一用户的连接通常只有 1~3 条，遍历成本可忽略（第 04 章的 CloseDevice 用的是同一手法）；只有目标用户不在这台机器上时才查 Redis 这个二级路由：

// Resolve 两级查找：一级是 Hub（本机连接），二级是 Redis 路由。
func (d *Deliverer) Resolve(ctx context.Context, uid uint64, deviceID string) (*Client, *store.Route, error) {
	if c, ok := d.hub.Local(uid, deviceID); ok {
		return c, nil, nil // 本机直投，省掉一次 Redis 往返
	}
	rt, err := d.routes.Lookup(ctx, uid, deviceID)
	if err != nil {
		return nil, nil, fmt.Errorf("lookup route uid=%d device=%s: %w", uid, deviceID, err)
	}
	return nil, rt, nil // rt == nil 即离线，交给离线消息兜底
}

第 04 章的踢下线通过 im:kick 广播，网关的常驻订阅就在这里落地：收到消息后按 (uid, device_id) 调 Hub.CloseDevice(uid, deviceID, 4001, "kicked")，连接关闭流程会顺手摘掉路由，所以被踢的设备不会留下幽灵路由。

本机命中率决定这条链路的成本：单机压测时几乎 100% 走内存，看着很快；一旦扩到多台网关、热聊双方分散在不同机器上，Redis 查询 qps 会随消息量线性上涨，所以 route:uid:{uid} 的读延迟必须单独埋点（第 12 章）。

验证方式

# 客户端连上后路由应立刻出现（Hash：field=deviceID，value=路由 JSON）
redis-cli HGETALL route:uid:1001
# 期望：1) "ios-8a2c"  2) "{\"uid\":1001,\"device_id\":\"ios-8a2c\",\"gateway_id\":\"gw-1\",\"conn_id\":1042}"

redis-cli TTL route:uid:1001     # 期望：始终在 60~90 之间，不会掉到 0
redis-cli SMEMBERS route:gw:gw-1 # 期望：包含 "1001:ios-8a2c"
redis-cli GET route:conn:1042    # 期望："1001:ios-8a2c"

# 直接 kill 掉客户端进程（不发关闭帧），等 90 秒后
redis-cli HGETALL route:uid:1001 # 期望：(empty array)，幽灵路由被 TTL 摘除

常见坑

坑现象正确做法
TTL 与心跳间隔取同一个值一次网络抖动就把在线用户打成离线TTL 取心跳间隔的 3 倍（30s 心跳配 90s）
只续 route:uid 不续 route:conn连接反查拿不到 uid，清理逻辑失效两条键放同一 pipeline 一起续
Hash 字段用 uid 而不是 deviceID多端登录互相覆盖，手机顶掉桌面字段粒度必须是 deviceID
集群下用 Lua 一次写三张键CROSSSLOT 报错，登录整体失败用 pipeline（不保证原子）或加 hash tag 绑同槽

小结：路由表就是三张 Redis 键——route:uid:{uid} 存设备到网关的映射、route:conn:{conn_id} 做连接反查、route:gw:{gateway_id} 做网关重启后的批量清理；TTL 靠心跳续约，投递失败立刻摘除；查询侧用「Hub 本机连接 + Redis 路由」两级查找，把绝大多数投递留在内存里。下一章把这条路串成完整的单聊链路。

笔记加载中…

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

返回知识库