198｜异地多活架构：单元化、数据分区、流量路由、冲突解决和容灾切换

┌─────────────────────────────────────────────────────┐
│                      全局路由层                       │
│          (按用户ID哈希 → 路由到对应单元)                │
└──────┬──────────────┬──────────────┬────────────────┘
       │              │              │
  ┌────▼────┐    ┌────▼────┐    ┌────▼────┐
  │ 单元 A   │    │ 单元 B   │    │ 单元 C   │
  │ 华东机房  │    │ 华北机房  │    │ 华南机房  │
  │          │    │          │    │          │
  │ 接入层    │    │ 接入层    │    │ 接入层    │
  │ 业务层    │    │ 业务层    │    │ 业务层    │
  │ 数据库    │    │ 数据库    │    │ 数据库    │
  │ 缓存      │    │ 缓存      │    │ 缓存      │
  │ 消息队列  │    │ 消息队列  │    │ 消息队列  │
  └──────────┘    └──────────┘    └──────────┘

单元编号 = hash(userId) % 单元总数

单元 = 根据用户所属地域映射

单元A = 订单服务集群
单元B = 商品服务集群
单元C = 支付服务集群

┌──────────┐
│  DNS 层   │  ← 智能 DNS 解析，根据用户 IP 返回最近机房入口 IP
└────┬─────┘
┌────▼─────┐
│ 接入层    │  ← Nginx / API Gateway，解析 Cookie/Header 中的路由信息
└────┬─────┘
┌────▼─────┐
│ 路由层    │  ← 根据 userId 查路由表，判断目标单元，必要时转发
└────┬─────┘
     │
  ┌──▼──┐
  │ 单元 │
  └─────┘

-- 简化的路由表结构
CREATE TABLE route_table (
    user_id      BIGINT PRIMARY KEY,
    cell_id      VARCHAR(16) NOT NULL,   -- 目标单元
    version      BIGINT NOT NULL,         -- 乐观锁版本号
    updated_at   TIMESTAMP NOT NULL
);

-- 分片规则：hash(user_id) % 单元数
-- 路由表的查询：O(1) 哈希计算，不需要真正查数据库

  单元A (写)                    单元B
  ┌──────┐                    ┌──────┐
  │ 业务  │──写入本地DB         │      │
  │ 服务  │──发MQ消息──────────▶│ 消费  │
  │      │                    │ 写DB  │
  └──────┘                    └──────┘

  单元A                          单元B
  ┌──────┐                    ┌──────┐
  │ 业务  │──写本地DB            │      │
  │ 服务  │──RPC调用───────────▶│ 写DB  │
  │      │◀──返回结果───────────│      │
  └──────┘                    └──────┘

单元A 写入 → 同步到 单元B → 单元B 的同步机制又发回 单元A → ...

public Data merge(Data local, Data remote) {
    if (remote.getVersion() > local.getVersion()) {
        return remote;
    }
    return local;
}

// G-Counter 示例
public class GCounter {
    private Map<String, Long> counts = new HashMap<>(); // nodeId → count

    public void increment(String nodeId) {
        counts.merge(nodeId, 1L, Long::sum);
    }

    public long value() {
        return counts.values().stream().mapToLong(Long::longValue).sum();
    }

    public void merge(GCounter other) {
        other.counts.forEach((node, count) ->
            counts.merge(node, count, Math::max));
    }
}

// 示例：购物车合并
public Cart mergeCart(Cart cart1, Cart cart2) {
    Cart merged = new Cart();
    // 合并两个购物车的商品，相同商品数量取最大值
    for (Item item : union(cart1.getItems(), cart2.getItems())) {
        int qty1 = cart1.getQuantity(item.getSku());
        int qty2 = cart2.getQuantity(item.getSku());
        merged.addItem(item, Math.max(qty1, qty2));
    }
    return merged;
}

// 版本向量：[单元A:3, 单元B:5, 单元C:2]
// 含义：该数据在单元A被修改了3次，单元B 5次，单元C 2次

public class VersionVector {
    private Map<String, Long> versions = new HashMap<>();

    public void increment(String cellId) {
        versions.merge(cellId, 1L, Long::sum);
    }

    public ConflictType compare(VersionVector other) {
        boolean thisGreater = false, otherGreater = false;
        Set<String> allCells = new HashSet<>();
        allCells.addAll(this.versions.keySet());
        allCells.addAll(other.versions.keySet());

        for (String cell : allCells) {
            long v1 = this.versions.getOrDefault(cell, 0L);
            long v2 = other.versions.getOrDefault(cell, 0L);
            if (v1 > v2) thisGreater = true;
            if (v2 > v1) otherGreater = true;
        }

        if (thisGreater && !otherGreater) return ConflictType.THIS_WINS;
        if (otherGreater && !thisGreater) return ConflictType.OTHER_WINS;
        if (!thisGreater && !otherGreater) return ConflictType.EQUAL;
        return ConflictType.CONCURRENT; // 并发冲突，需要合并
    }
}

┌──────────────┐
│ 1. 故障检测   │  ← 监控系统发现单元A不可用
└──────┬───────┘
       ▼
┌──────────────┐
│ 2. 切流决策   │  ← 自动/人工确认，生成切换指令
└──────┬───────┘
       ▼
┌──────────────┐
│ 3. 路由更新   │  ← 更新路由表，将单元A的用户指向单元B
└──────┬───────┘
       ▼
┌──────────────┐
│ 4. 流量迁移   │  ← DNS/接入层生效新路由，流量逐步切到单元B
└──────┬───────┘
       ▼
┌──────────────┐
│ 5. 数据补偿   │  ← 单元A恢复后，补偿切换期间丢失的数据
└──────────────┘

目标：将单元A的 10% 用户迁移到单元B

步骤：
1. 更新路由表，将 10% 用户的 route 指向单元B
2. 等待数据同步完成（单元B拉取这 10% 用户的全量数据）
3. 验证单元B中数据完整性
4. DNS/接入层生效新路由
5. 观察 30 分钟，确认无异常
6. 继续迁移下一批

目标：单元A完全不可用，所有流量切到单元B

步骤：
1. 确认单元A确实故障（多探测点 + 人工确认）
2. 立即更新路由表，所有单元A的用户指向单元B
3. 推送路由变更到所有接入层
4. 单元B接收全部流量，可能触发自动扩容
5. 故障恢复后，逐步回切流量

// 数据补偿的关键逻辑
public void compensateData(String sourceCell, String targetCell, long failoverTime) {
    // 1. 查询故障期间在 sourceCell 上的所有变更
    List<ChangeLog> changes = changeLogService.query(
        sourceCell,
        failoverTime,  // 故障开始时间
        System.currentTimeMillis()  // 当前时间
    );

    // 2. 逐条应用到 targetCell
    for (ChangeLog change : changes) {
        try {
            dataSyncService.apply(targetCell, change);
        } catch (ConflictException e) {
            // 3. 冲突时使用版本向量判断
            conflictResolver.resolve(change, targetCell);
        }
    }

    // 4. 验证数据一致性
    boolean consistent = dataValidator.verify(sourceCell, targetCell);
    if (!consistent) {
        alertService.send("数据补偿后仍不一致，需人工处理");
    }
}

┌─────────────────┐
│   华东机房        │
│  全部服务 + 全部数据 │
└─────────────────┘

┌──────────┐    ┌──────────┐
│ 华东机房A  │    │ 华东机房B  │
│ 主库(写)   │◀──▶│ 从库(读)   │
│           │ 同步 │           │
└──────────┘    └──────────┘

┌──────────┐  异步复制  ┌──────────────┐
│ 同城双活   │──────────▶│ 异地灾备中心    │
│ 上海A+B   │           │ 深圳（冷备）    │
└──────────┘           └──────────────┘

┌──────────┐   ┌──────────┐   ┌──────────┐
│ 华东单元   │   │ 华北单元   │   │ 华南单元   │
│ (上海)    │   │ (北京)    │   │ (深圳)    │
│ 用户: 33% │   │ 用户: 33% │   │ 用户: 34% │
└──────────┘   └──────────┘   └──────────┘
      │              │              │
      └──────────────┼──────────────┘
                     │
              ┌──────▼──────┐
              │  全局数据层   │
              │ (商品、配置)  │
              └─────────────┘