异地多活避坑实录：单元化数据延迟与切流演练

为什么异地多活不能只看“机房都活着”

异地多活的目标不是把服务部署到两个城市，而是让任一机房故障时，用户请求仍能在可接受的一致性和延迟范围内继续执行。单元化架构通过用户、商户或订单维度将流量封闭在一个单元内，避免每次写请求都跨城访问主库；但它也把问题从“跨城调用慢”转变成“跨单元数据多久可见、切流后旧数据能不能读到、冲突如何处理”。

很多团队第一次做演练时，只验证了网关能否把流量切到备份机房，却没有验证数据链路。结果切流后出现三类典型故障：用户刚修改的地址在新单元不可见；旧单元仍有异步写入，切回后发生数据覆盖；监控显示同步延迟为 0 毫秒，但业务查询仍读不到数据。根因通常是只监控了数据库复制位点，没有把消息队列、缓存失效、本地索引和业务重试纳入端到端评估。

单元化架构下的数据同步路径

一个典型单元化请求会在网关层根据分片键计算单元归属。正常情况下，用户的读写都在归属单元完成；数据库 Binlog 或 CDC 再把变更同步到其他单元，供查询、容灾和数据校验使用。切流不是简单改 DNS，而是同时调整路由规则、同步状态、写保护和回滚策略。

同步延迟评估：不要只盯数据库毫秒数

建议同时评估四类指标。

• 传输延迟：跨城专线 RTT、CDC 抓取耗时、消息队列堆积和目标库回放耗时。
• 一致性延迟：业务在 A 单元提交写入后，B 单元通过正常读接口能查到该变更的耗时。
• 热点延迟：大促、批量任务或单一大用户造成的分片热点，可能让平均值很好看但尾部延迟失控。
• 恢复延迟：链路抖动后，积压消息追平所需时间；它决定切流窗口，而不是当前一秒延迟。

推荐使用 P50、P95、P99 和最大追平时间。切流门禁至少关注 P99 与积压量，因为切流期间受影响的往往是最近几秒内发生写入的用户。对于余额、库存、权益发放等强一致业务，不能依赖异步复制接管，应使用中心写、分布式事务或“切流后冻结旧归属写入”的业务策略。

三种切流方案对比

整机房切流实现简单，适合机房断电、网络中断等场景，但爆炸半径大，缓存未命中率和连接池抖动会集中爆发。

按分片灰度切流先切 1% 用户或少量分片，可观察同步延迟、错误率和业务一致性，是日常演练首选。

按接口或读流量切流风险最低，只能验证查询链路和备机房容量，不能证明写入接管能力。成熟团队通常采用“只读预热、小流量写入、分片扩大、整机房兜底”的组合。

Go 实现：延迟探测与灰度切流门禁

下面示例不依赖外部框架，使用 Go 标准库即可运行。它模拟写入 A 单元、异步复制到 B 单元，并在切流前检查端到端延迟、复制积压和写保护状态。生产中可将内存存储替换为数据库与 CDC 位点，将路由规则下沉到网关或配置中心。

package main

import (
	"errors"
	"fmt"
	"sync"
	"time"
)

// DataEvent 表示一次业务写事件，Version 使用时间戳模拟版本号。
type DataEvent struct {
	UserID  string
	Content string
	Version int64
}

// Unit 表示一个逻辑单元，生产中可对应城市机房内的服务与数据库集合。
type Unit struct {
	name string
	mu   sync.RWMutex
	data map[string]DataEvent
}

func NewUnit(name string) *Unit {
	return &Unit{name: name, data: make(map[string]DataEvent)}
}

func (u *Unit) Write(e DataEvent) {
	u.mu.Lock()
	defer u.mu.Unlock()
	u.data[e.UserID] = e
}

func (u *Unit) Read(userID string) (DataEvent, bool) {
	u.mu.RLock()
	defer u.mu.RUnlock()
	e, ok := u.data[userID]
	return e, ok
}

// ReplicationLagProbe 统计端到端复制延迟，而不是只看数据库位点。
type ReplicationLagProbe struct {
	mu       sync.Mutex
	lag      []time.Duration
	backlog  int
	maxDelay time.Duration
}

func (p *ReplicationLagProbe) Record(lag time.Duration, backlog int) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.lag = append(p.lag, lag)
	p.backlog = backlog
}

func percentile(values []time.Duration, p float64) time.Duration {
	if len(values) == 0 {
		return 0
	}
	// 示例采用复制后排序；正式代码可使用流式分位数或 Prometheus histogram。
	for i := 0; i < len(values); i++ {
		for j := i + 1; j < len(values); j++ {
			if values[j] < values[i] {
				values[i], values[j] = values[j], values[i]
			}
		}
	}
	idx := int(float64(len(values)-1) * p)
	return values[idx]
}

func (p *ReplicationLagProbe) Allow() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	values := append([]time.Duration(nil), p.lag...)
	p99 := percentile(values, 0.99)
	if p99 > p.maxDelay {
		return fmt.Errorf("P99 延迟 %s 超过阈值 %s", p99, p.maxDelay)
	}
	if p.backlog > 1000 {
		return fmt.Errorf("复制积压 %d 超过 1000", p.backlog)
	}
	return nil
}

// Router 维护用户到单元的归属，Cutover 表示是否已经切到 B。
type Router struct {
	mu      sync.RWMutex
	cutover map[string]bool
}

func (r *Router) Target(userID string) string {
	r.mu.RLock()
	defer r.mu.RUnlock()
	if r.cutover[userID] {
		return "B"
	}
	return "A"
}

func (r *Router) StartCutover(userIDs []string) {
	r.mu.Lock()
	defer r.mu.Unlock()
	for _, id := range userIDs {
		r.cutover[id] = true
	}
}

// 模拟 CDC：A 的写入经过一段延迟后在 B 可见。
func startReplication(a, b *Unit, probe *ReplicationLagProbe, stop <-chan struct{}) {
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()
	seen := make(map[string]int64)
	for {
		select {
		case <-stop:
			return
		case <-ticker.C:
			a.mu.RLock()
			for uid, event := range a.data {
				if seen[uid] >= event.Version {
					continue
				}
				delay := time.Duration(150+len(uid)*20) * time.Millisecond
				time.Sleep(delay / 10)
				b.Write(event)
				seen[uid] = event.Version
				probe.Record(time.Since(time.UnixMilli(event.Version)), 0)
			}
			a.mu.RUnlock()
		}
	}
}

func write(units map[string]*Unit, router *Router, userID, content string) error {
	target := router.Target(userID)
	// 写保护演练：切流后旧单元必须拒绝双写，避免脑裂。
	if target == "B" {
		if _, stillWritable := units["A"].Read(userID); stillWritable {
			return errors.New("旧单元仍存在可写路径，拒绝双写")
		}
	}
	event := DataEvent{UserID: userID, Content: content, Version: time.Now().UnixMilli()}
	units[target].Write(event)
	return nil
}

func main() {
	a, b := NewUnit("A"), NewUnit("B")
	units := map[string]*Unit{"A": a, "B": b}
	router := &Router{cutover: map[string]bool{}}
	probe := &ReplicationLagProbe{maxDelay: 2 * time.Second}
	stop := make(chan struct{})
	defer close(stop)

	go startReplication(a, b, probe, stop)

	users := []string{"u-001", "u-002", "u-003"}
	for _, uid := range users {
		a.Write(DataEvent{UserID: uid, Content: "初始化资料", Version: time.Now().UnixMilli()})
	}
	time.Sleep(800 * time.Millisecond)

	if err := probe.Allow(); err != nil {
		fmt.Println("切流被阻断:", err)
		return
	}
	router.StartCutover([]string{"u-001"})
	if err := write(units, router, "u-001", "新单元修改资料"); err != nil {
		fmt.Println("写入被保护:", err)
	} else {
		fmt.Println("灰度用户已切到 B 单元写入")
	}
	fmt.Printf("u-001 当前归属单元：%s\n", router.Target("u-001"))
	fmt.Printf("u-002 当前归属单元：%s\n", router.Target("u-002"))
}

运行后可以看到，只有通过延迟门禁的用户才会进入切流集合。示例中的写保护故意保留了一个常见问题：B 接管后 A 中仍能读到历史数据，因此直接写入会被拒绝。生产实现应把这个判断替换为网关写禁入、数据库只读、消息消费冻结和幂等修复任务。

演练步骤：从影子流量到故障切换

第一步，准备数据探针。每个分片选择低频、中频和高频用户，持续写入带版本号的业务探针，不使用简单 select 1。

第二步，做只读切流。将 1% 查询流量导向备单元，验证缓存命中率、连接池、索引计划和跨单元查询比例。

第三步，做单分片写切流。先冻结该分片在旧单元的写入，再等待 CDC 积压清零，随后开放新单元写入。观察窗口至少覆盖业务超时时间的 3 到 5 倍。

第四步，扩大灰度并执行回滚。每扩大一批都要记录错误率、P99 延迟、数据校验差异和业务补偿量。回滚也必须演练，因为真实故障中最危险的是“切过去发现不行，却切不回来”。

避坑总结

1. 错误现象：监控延迟为 0，切流后用户仍读不到新数据。 不要只采集数据库 seconds_behind_source，应从业务写接口打点，到目标单元读接口返回新版本为止，统计端到端延迟；消息队列、缓存失效和搜索索引都要单独埋点。

2. 错误现象：切流后两个单元都产生订单或库存扣减。 修正方法是在网关、服务和数据库三层同时加写保护：路由层拒绝旧归属请求，应用层校验单元身份，数据库层将旧单元分片置为只读。只改网关规则无法覆盖内部任务和 MQ 消费者。

3. 错误现象：平均值正常，少量用户持续报资料丢失。 这通常是热点分片或大事务拖慢单通道回放。应按分片维度展示 P99 延迟和积压，切流按分片推进；对超大用户单独限流或拆片，不要让一个用户阻塞整个容灾单元。

4. 错误现象：回滚后新单元写入被旧数据覆盖。 双向同步必须携带写入单元、版本号和逻辑时钟，回滚前设置冲突保护。冲突无法自动判定时，进入修复队列，而不是用最后更新时间简单覆盖。

5. 错误现象：演练都在低峰通过，大促切换失败。 容灾指标要在压测流量下验证，包括备单元容量、缓存预热、数据库连接数和 MQ 消费能力。异地多活的可信度来自常态化、小流量、可回滚的演练，而不是一次性的架构图评审。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。