#多活架构设计

// MySQL双主同步配置
// 机房A MySQL配置
[mysqld]
server-id = 1
log-bin = mysql-bin
binlog-format = ROW
gtid-mode = ON
enforce-gtid-consistency = ON

# 双主配置
log-slave-updates = ON
auto-increment-offset = 1
auto-increment-increment = 2

# 机房间同步
replicate-do-db = business_db
replicate-ignore-db = mysql

// 机房B MySQL配置
[mysqld]
server-id = 2
auto-increment-offset = 2
auto-increment-increment = 2
# 其他配置同机房A

// Redis双活同步
@Service
public class RedisMultiActiveService {

    @Autowired
    private RedisTemplate redisTemplateA;  // 机房A

    @Autowired
    private RedisTemplate redisTemplateB;  // 机房B

    @Autowired
    private RedisClusterClient clusterClientA;

    /**
     * 写入：双写
     */
    public void set(String key, String value) {
        // 写入本机房
        redisTemplateA.opsForValue().set(key, value);

        // 异步同步到另一个机房
        CompletableFuture.runAsync(() -> {
            try {
                redisTemplateB.opsForValue().set(key, value);
            } catch (Exception e) {
                log.error("Redis同步失败，key={}", key, e);
                // 记录同步失败，等待重试
                syncFailedKey(key, value);
            }
        });
    }

    /**
     * 读取：本机房优先
     */
    public String get(String key) {
        // 优先读本机房
        String value = (String) redisTemplateA.opsForValue().get(key);
        if (value != null) {
            return value;
        }

        // 本机房没有，读另一个机房
        value = (String) redisTemplateB.opsForValue().get(key);
        if (value != null) {
            // 回填本机房
            redisTemplateA.opsForValue().set(key, value);
        }

        return value;
    }
}

// 流量路由服务
@Service
public class TrafficRouter {

    @Autowired
    private ConfigService configService;

    /**
     * 根据用户ID路由到机房
     * 路由规则：hash(userId) % 机房数
     */
    public String route机房(Long userId) {
        // 从配置中心获取机房列表
        List<String> regions = configService.getRegions();

        // 按用户ID哈希
        int index = Math.abs(userId.hashCode()) % regions.size();

        return regions.get(index);
    }

    /**
     * 根据地域路由
     */
    public String routeByRegion(String region) {
        Map<String, String> regionMapping = Map.of(
            "华南", "广州机房",
            "华东", "上海机房",
            "华北", "北京机房",
            "西北", "西安机房"
        );
        return regionMapping.getOrDefault(region, "北京机房");
    }

    /**
     * 故障切换
     */
    public void switchTo(String targetRegion) {
        // 更新DNS/负载均衡配置
        configService.setActiveRegion(targetRegion);

        // 发送告警
        alertService.sendAlert("机房切换", "切换到" + targetRegion);

        log.info("流量切换到{}", targetRegion);
    }
}

// 异地数据同步
@Service
public class CrossRegionSyncService {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 异步同步：消息队列
     */
    public void asyncSync(String table, Object data) {
        // 发送消息到跨机房Topic
        kafkaTemplate.send("cross-region-sync",
            table,
            JSON.toJSONString(new SyncEvent(table, data, System.currentTimeMillis()))
        );
    }

    /**
     * 同步：强一致数据用DTS
     */
    @Transactional
    public void syncBalance(Long userId, BigDecimal change) {
        // 1. 本机房更新
        accountService.updateBalance(userId, change);

        // 2. 发送同步消息
        kafkaTemplate.send("balance-sync",
            String.valueOf(userId),
            JSON.toJSONString(new BalanceSyncEvent(userId, change))
        );
    }

    /**
     * 冲突处理：后写者胜出
     */
    @KafkaListener(topics = "balance-sync")
    public void handleBalanceSync(String message) {
        BalanceSyncEvent event = JSON.parseObject(message, BalanceSyncEvent.class);

        // 检查版本
        AccountDO local = accountService.getByUserId(event.getUserId());
        if (local.getVersion() >= event.getVersion()) {
            // 本地版本更新，跳过
            return;
        }

        // 更新本地
        accountService.updateBalanceWithVersion(event);
    }
}

// 自动故障切换
@Service
public class FailoverService {

    @Autowired
    private HealthCheckService healthCheck;

    @Autowired
    private TrafficRouter trafficRouter;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    private final AtomicInteger failureCount = new AtomicInteger(0);
    private static final int FAILURE_THRESHOLD = 3;

    /**
     * 健康检查
     */
    @Scheduled(fixedRate = 5000)
    public void healthCheck() {
        String currentRegion = configService.getCurrentRegion();

        boolean healthy = healthCheck.checkRegion(currentRegion);
        if (!healthy) {
            int count = failureCount.incrementAndGet();
            if (count >= FAILURE_THRESHOLD) {
                log.error("机房{}健康检查连续失败{}次，触发切换", currentRegion, count);
                triggerFailover(currentRegion);
            }
        } else {
            failureCount.set(0);
        }
    }

    /**
     * 触发故障切换
     */
    private void triggerFailover(String failedRegion) {
        // 1. 找到备用机房
        String standbyRegion = findStandbyRegion(failedRegion);
        if (standbyRegion == null) {
            log.error("没有可用备用机房！");
            alertService.sendCriticalAlert("无备用机房可用");
            return;
        }

        // 2. 切换流量
        trafficRouter.switchTo(standbyRegion);

        // 3. 发送切换事件
        kafkaTemplate.send("failover-event",
            failedRegion,
            JSON.toJSONString(new FailoverEvent(failedRegion, standbyRegion))
        );

        // 4. 记录切换日志
        log.warn("故障切换完成：{} -> {}", failedRegion, standbyRegion);

        // 5. 发送告警
        alertService.sendAlert("机房切换", failedRegion + "切换到" + standbyRegion);
    }

    private String findStandbyRegion(String failedRegion) {
        List<String> allRegions = configService.getAllRegions();
        return allRegions.stream()
            .filter(r -> !r.equals(failedRegion))
            .filter(this::checkRegionHealth)
            .findFirst()
            .orElse(null);
    }

    private boolean checkRegionHealth(String region) {
        try {
            return healthCheck.checkRegion(region);
        } catch (Exception e) {
            return false;
        }
    }
}