别把容灾做成备份：跨机房 RPO<5min、RTO<30min 的架构决策与踩坑实录

-- MySQL 半同步复制配置（主库端）
-- 安装半同步插件
INSTALL PLUGIN rpl_semi_sync_source SONAME 'semisync_source.so';

-- 开启半同步并设置超时（毫秒）
SET GLOBAL rpl_semi_sync_source_enabled = 1;
SET GLOBAL rpl_semi_sync_source_timeout = 3000;  -- 3秒超时，超时降级为异步

-- 查看半同步状态
SHOW STATUS LIKE 'Rpl_semi_sync_source_status';
SHOW STATUS LIKE 'Rpl_semi_sync_source_yes_tx';  -- 成功的半同步事务数

机房A ←→ 仲裁节点 ←→ 机房B
  ↕                    ↕
  └─── 同步复制 ──────┘

#!/bin/bash
# fencing 脚本：检测到脑裂时，强制隔离备机房
# 通过 API 关闭备机房数据库实例 + 切断流量入口

STANDBY_HOST="10.0.2.10"
STANDBY_API="http://${STANDBY_HOST}:8080/db/failover"

# 检测心跳：连续 3 次超时才执行隔离
HEARTBEAT_FAIL=0
for i in 1 2 3; do
    if ! ping -c 1 -W 2 ${STANDBY_HOST} > /dev/null 2>&1; then
        HEARTBEAT_FAIL=$((HEARTBEAT_FAIL + 1))
    fi
done

if [ ${HEARTBEAT_FAIL} -ge 3 ]; then
    echo "[$(date)] 脑裂检测：备机房连续3次心跳失败，执行隔离"
    # 1. 切断备机房 DNS 入口
    curl -s -X POST http://dns-api.internal/failover?disable=standby
    # 2. 强制关闭备机房数据库
    curl -s -X POST ${STANDBY_API}
    echo "[$(date)] 备机房已隔离，流量全部切到主机房"
fi

时间线：
  主库：  T1(写入100条) → T2(写入200条) → T3(故障)
  备库：  T1(已同步)     → T2(同步中)     → T3(还差150条)

切换流程：
  1. 冻结主库写入（通过中间件拦截或 DNS 下线）
  2. 等待备库追平日志（监控 GTID 差值归零）
  3. 校验关键表行数和校验和
  4. 切换流量到备库

-- 主库：检查 GTID 是否完整
SHOW MASTER STATUS\G
-- 查看已执行和待执行的 GTID 集合

-- 备库：检查 GTID 同步状态
SHOW SLAVE STATUS\G
-- 关注 Retrieved_Gtid_Set 和 Executed_Gtid_Set
-- 如果两者不同，说明有未执行的 GTID

-- 备库：计算 GTID 差值
SELECT @@global.gtid_executed;
-- 与主库的 gtid_executed 对比，差值就是未同步的事务

-- 大事务检测：超过阈值的单个事务
SET GLOBAL binlog_transaction_compression=ON;
SET GLOBAL binlog_transaction_compression_level_zstd=6;

#!/usr/bin/env python3
"""主备数据一致性校验脚本"""
import hashlib
import mysql.connector
import sys

def table_checksum(host, user, password, database, table):
    """计算表的校验和"""
    conn = mysql.connector.connect(host=host, user=user, password=password, database=database)
    cursor = conn.cursor()
    cursor.execute(f"SELECT COUNT(*), COALESCE(SUM(CRC32(CONCAT_WS('#', {get_columns(cursor, table)}))), 0) FROM {table}")
    count, checksum = cursor.fetchone()
    cursor.close()
    conn.close()
    return count, checksum

def get_columns(cursor, table):
    """获取表所有列名"""
    cursor.execute(f"SHOW COLUMNS FROM {table}")
    return ', '.join([f'`{row[0]}`' for row in cursor.fetchall()])

if __name__ == '__main__':
    primary = {'host': '10.0.1.10', 'user': 'check_user', 'password': sys.argv[1], 'database': 'orders'}
    standby = {'host': '10.0.2.10', 'user': 'check_user', 'password': sys.argv[1], 'database': 'orders'}
    
    tables = ['t_order', 't_order_detail', 't_payment', 't_user']
    for table in tables:
        p_count, p_checksum = table_checksum(**primary, table=table)
        s_count, s_checksum = table_checksum(**standby, table=table)
        
        if p_checksum != s_checksum:
            print(f"[ALERT] {table}: checksum mismatch! primary={p_checksum}, standby={s_checksum}")
            print(f"  primary count={p_count}, standby count={s_count}")
        else:
            print(f"[OK] {table}: checksum matched, count={p_count}")

阶段1：故障检测与决策（0-2分钟）
  → 告警触发，SRE 确认故障范围
  → 决策：是否需要切换？切到哪个机房？

阶段2：切换准备（2-8分钟）
  → 冻结主库写入（中间件拦截 + DNS 下线）
  → 等待备库日志追平（GTID 差值归零）
  → 校验关键表数据一致性
  → 通知上下游依赖系统

阶段3：流量切换（8-15分钟）
  → DNS 权重调整，先切 5% 流量验证
  → 验证通过后切全部流量
  → 更新服务注册中心，刷新连接池

阶段4：验证与观察（15-30分钟）
  → 监控核心指标：错误率、延迟、吞吐量
  → 人工验证核心业务流程
  → 确认稳定后通知业务方

// Java 应用：解决 JVM DNS 缓存问题
// 在应用启动时设置 DNS 缓存 TTL 为 10 秒
import java.security.Security;

public class DnsCacheConfig {
    static {
        // 设置 JVM DNS 缓存 TTL（秒）
        Security.setProperty("networkaddress.cache.ttl", "10");
        Security.setProperty("networkaddress.cache.negative.ttl", "0");
    }
}

# ProxySQL 配置：故障切换时自动更新后端
# proxysql.cnf
mysql_servers:
  - hostgroup_id: 1        # 写组
    hostname: "10.0.1.10"   # 主库
    port: 3306
    max_connections: 200
  - hostgroup_id: 2        # 读组
    hostname: "10.0.2.10"   # 备库
    port: 3306
    max_connections: 200

mysql_galera_hostgroups:
  writer_hostgroup: 1
  backup_writer_hostgroup: 2
  active_writer: "10.0.1.10"  # 当前主库
  # 当主库不可用时，ProxySQL 自动切换到 backup_writer

# Kafka 消费者 offset 持久化到数据库（切换后可恢复）
import json
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'order_events',
    bootstrap_servers='kafka-primary:9092',
    group_id='order-processor',
    enable_auto_commit=False  # 关闭自动提交，手动管理 offset
)

def save_offset_to_db(topic, partition, offset):
    """将 offset 持久化到数据库，切换后可恢复"""
    cursor.execute(
        "INSERT INTO kafka_offset (topic, partition_id, consumer_group, offset_val, updated_at) "
        "VALUES (%s, %s, %s, %s, NOW()) "
        "ON DUPLICATE KEY UPDATE offset_val=VALUES(offset_val), updated_at=NOW()",
        (topic, partition, 'order-processor', offset)
    )

def load_offset_from_db(topic, consumer_group):
    """从数据库恢复 offset"""
    cursor.execute(
        "SELECT partition_id, offset_val FROM kafka_offset "
        "WHERE topic=%s AND consumer_group=%s", (topic, consumer_group)
    )
    return {row[0]: row[1] for row in cursor.fetchall()}

# 消费时手动保存 offset
for message in consumer:
    process_message(message.value)
    save_offset_to_db(message.topic, message.partition, message.offset)
    consumer.commit()  # 同时提交到 Kafka，双保险

# 切换编排配置（Ansible Playbook 片段）
---
- name: 容灾切换编排
  hosts: localhost
  vars:
    primary_dc: "dc-a"
    standby_dc: "dc-b"
    switch_reason: "{{ switch_reason }}"

  tasks:
    # 阶段1：故障确认
    - name: 检查主机房健康状态
      uri:
        url: "http://{{ primary_dc }}-health:8080/health"
        method: GET
        return_content: yes
      register: health_check
      failed_when: health_check.status != 200

    - name: 确认需要切换（需人工确认）
      pause:
        prompt: "主机房健康检查失败，确认切换到 {{ standby_dc }}？(yes/no)"

    # 阶段2：切换准备
    - name: 冻结主机房写入
      uri:
        url: "http://{{ primary_dc }}-proxy:8080/maintenance"
        method: POST
        body: '{"action": "freeze_writes"}'

    - name: 等待备库追平日志
      shell: "mysql -h {{ standby_dc }}-db -u monitor -e 'SHOW SLAVE STATUS\\G'"
      register: slave_status
      retries: 30
      delay: 10
      until: "'Seconds_Behind_Master: 0' in slave_status.stdout"

    - name: 校验关键表数据一致性
      shell: "python3 /opt/dr/checksum_verify.py --primary {{ primary_dc }}-db --standby {{ standby_dc }}-db"
      register: checksum_result
      failed_when: "'MISMATCH' in checksum_result.stdout"

    # 阶段3：流量切换
    - name: 切换 DNS（先切5%验证）
      shell: "python3 /opt/dr/dns_switch.py --to {{ standby_dc }} --weight 5"

    - name: 验证5%流量正常
      uri:
        url: "http://{{ standby_dc }}-monitor:9090/api/v1/query?query=up"
      register: monitor_check
      retries: 6
      delay: 10
      until: monitor_check.status == 200

    - name: 全量切换DNS
      shell: "python3 /opt/dr/dns_switch.py --to {{ standby_dc }} --weight 100"

    # 阶段4：通知
    - name: 发送切换完成通知
      uri:
        url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={{ webhook_key }}"
        method: POST
        body_format: json
        body:
          msgtype: text
          text:
            content: "容灾切换完成: {{ primary_dc }} → {{ standby_dc }}, 原因: {{ switch_reason }}"

#!/bin/bash
# 缓存预热脚本：从数据库加载热点数据到新机房 Redis
# 在切换前执行

REDIS_NEW="10.0.2.20:6379"
MYSQL_HOST="10.0.2.10"

echo "[$(date)] 开始缓存预热..."

# 预热用户信息缓存（Top 10000 活跃用户）
mysql -h ${MYSQL_HOST} -u cache_loader -p${DB_PASS} -e "
  SELECT CONCAT('user:', user_id) AS key, 
         JSON_OBJECT('id', user_id, 'name', nickname, 'level', level) AS value
  FROM t_user 
  WHERE last_login_at > DATE_SUB(NOW(), INTERVAL 7 DAY)
  ORDER BY last_login_at DESC 
  LIMIT 10000
" --batch --raw | while IFS=$'\t' read -r key value; do
    redis-cli -h ${REDIS_NEW} SET "${key}" "${value}" EX 3600
done

echo "[$(date)] 用户缓存预热完成"

# 预热商品信息缓存
mysql -h ${MYSQL_HOST} -u cache_loader -p${DB_PASS} -e "
  SELECT CONCAT('product:', product_id), product_name, price, stock
  FROM t_product WHERE status=1 LIMIT 5000
" --batch --raw | while IFS=$'\t' read -r key name price stock; do
    redis-cli -h ${REDIS_NEW} HSET "${key}" name "${name}" price "${price}" stock "${stock}"
    redis-cli -h ${REDIS_NEW} EXPIRE "${key}" 1800
done

echo "[$(date)] 商品缓存预热完成"
echo "[$(date)] 缓存预热全部完成"

业务需求分析
  ├─ 数据丢失容忍度（RPO 要求）
  │   ├─ RPO=0 → 同步复制（同城双活）
  │   ├─ RPO<1min → 半同步复制
  │   └─ RPO<30min → 异步复制
  ├─ 业务中断容忍度（RTO 要求）
  │   ├─ RTO<1min → 自动切换（双活）
  │   ├─ RTO<15min → 半自动切换
  │   └─ RTO<2h → 人工切换
  ├─ 合规要求
  │   ├─ 金融/医疗 → 两地三中心
  │   └─ 一般业务 → 同城双活或主备
  └─ 成本预算
      ├─ 充足 → 两地三中心 + 全同步
      ├─ 适中 → 同城双活 + 半同步
      └─ 紧张 → 主备 + 异步

# MySQL 复制延迟监控
mysql_slave_status_seconds_behind_master{job="mysql-exporter"} > 30

# GTID 差异监控（自定义 exporter）
mysql_gtid_diff{job="mysql-exporter"} > 0

# 同步复制半同步状态
mysql_global_status_rpl_semi_sync_source_yes_tx{job="mysql-exporter"} == 0
# 半同步成功事务数为 0，说明降级为异步复制

# 仲裁节点心跳
probe_success{job="blackbox", instance="quorum-node:8080"} == 0

# Prometheus 告警规则：容灾系统健康监控
groups:
  - name: disaster_recovery
    rules:
      - alert: MySQLReplicationLag
        expr: mysql_slave_status_seconds_behind_master > 30
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "MySQL 复制延迟过高: {{ $value }}秒"
          description: "{{ $labels.instance }} 复制延迟超过30秒，RPO 可能超出目标"

      - alert: MySQLReplicationBroken
        expr: mysql_slave_status_slave_io_running == 0 or mysql_slave_status_slave_sql_running == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "MySQL 复制链路中断"
          description: "{{ $labels.instance }} IO/SQL 线程停止，容灾系统可能失效"

      - alert: SemiSyncDegraded
        expr: mysql_global_status_rpl_semi_sync_source_yes_tx == 0 and mysql_global_status_rpl_semi_sync_source_no_tx > 0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "半同步复制已降级为异步"
          description: "最近5分钟无半同步成功事务，可能已降级，RPO 不再为 0"

      - alert: DataChecksumMismatch
        expr: dr_data_checksum_mismatch > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "主备数据校验和不匹配"
          description: "{{ $labels.table }} 表数据不一致，需立即排查"

阶段1：原主机房恢复（验证修复完成）
  → 确认原主机房基础设施正常
  → 建立反向复制链路（灾备→原主机房）
  → 等待数据同步追平

阶段2：回切准备
  → 在低峰期执行（通常凌晨2-4点）
  → 通知业务方即将回切
  → 冻结灾备机房写入

阶段3：执行回切
  → 确认原主机房数据已追平
  → 校验关键表数据一致性
  → 切换 DNS 和流量到原主机房
  → 验证业务正常

阶段4：恢复容灾配置
  → 重新建立正向复制链路（原主机房→灾备）
  → 更新监控和告警配置
  → 通知业务方回切完成