Java双活案例

本文目录导读：

1. 双活架构概述
2. 核心架构代码实现
3. 关键实现要点
4. 最佳实践建议

我为您整理了一个完整的Java双活架构案例，包含架构设计、核心代码实现和关键考虑点。

双活架构概述

架构拓扑图

graph TB
    subgraph "数据中心A"
        LB1[负载均衡器]
        APP1[应用集群A]
        DB1[(数据库A)]
        MQ1[消息队列A]
        CACHE1[(缓存A)]
    end
    subgraph "数据中心B"
        LB2[负载均衡器]
        APP2[应用集群B]
        DB2[(数据库B)]
        MQ2[消息队列B]
        CACHE2[(缓存B)]
    end
    GSLB[全局负载均衡] --> LB1
    GSLB --> LB2
    APP1 <--> |同步| APP2
    DB1 <--> |双向同步| DB2
    MQ1 <--> |消息复制| MQ2
    CACHE1 <--> |缓存同步| CACHE2

核心架构代码实现

多数据中心配置管理

@Component
public class DataCenterConfig {
    @Value("${data.center.id:DC1}")
    private String currentDataCenter;
    @Value("${data.center.peer:DC2}")
    private String peerDataCenter;
    @Value("${data.center.routing.strategy:priority}")
    private String routingStrategy;
    public boolean isPrimary() {
        return "DC1".equals(currentDataCenter);
    }
    public String getCurrentDataCenter() {
        return currentDataCenter;
    }
}

读写分离与路由实现

public class DataSourceRouter extends AbstractRoutingDataSource {
    @Resource
    private DataCenterConfig dataCenterConfig;
    @Override
    protected Object determineCurrentLookupKey() {
        // 根据请求类型和当前数据中心状态决定路由
        String requestType = RequestContextHolder.getRequestType();
        boolean isWrite = "WRITE".equals(requestType);
        if (isWrite) {
            // 写操作路由到主数据中心
            return dataCenterConfig.getCurrentDataCenter() + "_MASTER";
        } else {
            // 读操作可以在本数据中心完成
            return dataCenterConfig.getCurrentDataCenter() + "_SLAVE";
        }
    }
}

分布式ID生成器（跨数据中心）

@Component
public class DistributedIdGenerator {
    @Resource
    private DataCenterConfig dataCenterConfig;
    private static final long SEQUENCE_BITS = 12;
    private static final long MACHINE_BITS = 5;
    private static final long DATA_CENTER_BITS = 5;
    public synchronized long nextId() {
        // 数据中心ID（保证跨数据中心唯一）
        long dataCenterId = dataCenterConfig.getDataCenterId();
        // 机器ID（本数据中心内唯一）
        long machineId = getMachineId();
        // 序列号
        long sequence = getNextSequence();
        // 组合成分布式ID
        return (dataCenterId << (MACHINE_BITS + SEQUENCE_BITS)) 
             | (machineId << SEQUENCE_BITS) 
             | sequence;
    }
}

数据同步实现

@Component
public class DataSyncService {
    @Resource
    private JdbcTemplate jdbcTemplate;
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    /**
     * 跨数据中心数据同步
     */
    public void syncDataToPeerCenter(String tableName, Object data) {
        // 构建同步消息
        SyncMessage syncMessage = SyncMessage.builder()
            .tableName(tableName)
            .data(data)
            .sourceCenter(dataCenterConfig.getCurrentDataCenter())
            .timestamp(System.currentTimeMillis())
            .build();
        // 发送到对等数据中心
        String topic = "data-sync-" + dataCenterConfig.getPeerDataCenter();
        kafkaTemplate.send(topic, JSON.toJSONString(syncMessage));
    }
    @KafkaListener(topics = "data-sync-#{@dataCenterConfig.currentDataCenter}")
    public void handleSyncData(String message) {
        SyncMessage syncMessage = JSON.parseObject(message, SyncMessage.class);
        // 使用分布式锁防止并发问题
        String lockKey = "sync:" + syncMessage.getTableName() + ":" + syncMessage.getId();
        RLock lock = redissonClient.getLock(lockKey);
        try {
            if (lock.tryLock(10, TimeUnit.SECONDS)) {
                // 执行数据同步
                jdbcTemplate.update(
                    "INSERT INTO " + syncMessage.getTableName() + 
                    " (id, data, sync_time) VALUES (?, ?, ?) " +
                    "ON DUPLICATE KEY UPDATE data = VALUES(data), sync_time = VALUES(sync_time)",
                    syncMessage.getId(),
                    JSON.toJSONString(syncMessage.getData()),
                    new Date(syncMessage.getTimestamp())
                );
                // 发布本地消息
                publishLocalEvent(syncMessage);
            }
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

分布式锁服务

@Component
public class DistributedLockService {
    @Resource
    private RedissonClient redissonClient;
    /**
     * 跨数据中心分布式锁
     */
    public boolean tryLock(String lockKey, long timeout, TimeUnit unit) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(timeout, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
    /**
     * 幂等性检查
     */
    public boolean checkIdempotent(String idempotentKey) {
        String key = "idempotent:" + idempotentKey;
        return redissonClient.getBucket(key).trySet("1", 24, TimeUnit.HOURS);
    }
}

健康检查与故障转移

@Component
public class HealthCheckService {
    @Resource
    private DataCenterConfig dataCenterConfig;
    @Resource
    private ApplicationContext applicationContext;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    @PostConstruct
    public void startHealthCheck() {
        // 每5秒检查一次健康状态
        scheduler.scheduleAtFixedRate(this::checkHealth, 0, 5, TimeUnit.SECONDS);
    }
    public void checkHealth() {
        boolean isHealth = true;
        // 检查数据库连接
        if (!checkDatabaseHealth()) {
            isHealth = false;
        }
        // 检查缓存服务
        if (!checkCacheHealth()) {
            isHealth = false;
        }
        // 检查消息队列
        if (!checkMQHealth()) {
            isHealth = false;
        }
        if (!isHealth) {
            handleFailover();
        }
    }
    private void handleFailover() {
        // 更新路由策略
        dataCenterConfig.setRoutingStrategy("HOT_STANDBY");
        // 启动本数据中心的完整服务
        applicationContext.publishEvent(new FailoverEvent(this, dataCenterConfig.getCurrentDataCenter()));
    }
}

双活应用入口配置

@Configuration
public class DualActiveConfig {
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.master")
    public DataSource masterDataSource() {
        return DataSourceBuilder.create().build();
    }
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.slave")
    public DataSource slaveDataSource() {
        return DataSourceBuilder.create().build();
    }
    @Bean
    public DataSource dataSource() {
        Map<Object, Object> targetDataSources = new HashMap<>();
        // 注册数据中心A和数据中心B的数据源
        targetDataSources.put("DC1_MASTER", masterDataSource());
        targetDataSources.put("DC1_SLAVE", slaveDataSource());
        targetDataSources.put("DC2_MASTER", peerMasterDataSource());
        targetDataSources.put("DC2_SLAVE", peerSlaveDataSource());
        DataSourceRouter router = new DataSourceRouter();
        router.setTargetDataSources(targetDataSources);
        router.setDefaultTargetDataSource(masterDataSource());
        return router;
    }
}

事务管理（分布式事务）

@Component
public class DistributedTransactionService {
    @Resource
    private TransactionTemplate transactionTemplate;
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    /**
     * 分布式事务处理
     */
    public void handleTransaction(TransactionRequest request) {
        String transactionId = UUID.randomUUID().toString();
        // 1. 准备阶段 - 记录事务日志
        saveTransactionLog(transactionId, request, "PREPARED");
        // 2. 执行本地事务
        try {
            transactionTemplate.execute(status -> {
                // 业务操作
                processBusinessLogic(request);
                // 记录事务状态
                updateTransactionStatus(transactionId, "COMMIT");
                return true;
            });
            // 3. 发送同步消息到对等数据中心
            sendTransactionSyncMessage(transactionId, request);
        } catch (Exception e) {
            // 4. 失败回滚
            updateTransactionStatus(transactionId, "ROLLBACK");
            handleTransactionFailure(transactionId, e);
        }
    }
    /**
     * 接收对等数据中心的同步请求
     */
    @KafkaListener(topics = "transaction-sync")
    public void handleTransactionSync(TransactionSyncMessage message) {
        String localTransactionId = generateLocalTransactionId(message);
        // 幂等检查
        if (checkIdempotent(localTransactionId)) {
            // 执行本地事务
            transactionTemplate.execute(status -> {
                processSyncedBusinessLogic(message);
                return true;
            });
        }
    }
}

应用配置示例

# application-dc1.yml
spring:
  datasource:
    master:
      url: jdbc:mysql://dc1-core-mysql:3306/business_db
      username: admin
      password: ${DB_PASSWORD}
    slave:
      url: jdbc:mysql://dc1-backup-mysql:3306/business_db
      username: admin
      password: ${DB_PASSWORD}
  kafka:
    bootstrap-servers: dc1-kafka:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: dc1-group
  redis:
    cluster:
      nodes: dc1-redis-1:6379,dc1-redis-2:6379,dc1-redis-3:6379
data:
  center:
    id: DC1
    peer: DC2
    routing:
      strategy: priority  # priority / hot_standby / load_balanced
    sync:
      mode: async  # async / sync
      retry:
        max-attempts: 3
        delay: 1000
# application.yml
spring:
  profiles:
    active: dc1  # 根据部署环境选择
  redis:
    redisson:
      config: classpath:redisson.yaml

Redis配置

# redisson.yaml
singleServerConfig:
  address: "redis://dc1-redis:6379"
  connectionMinimumIdleSize: 10
  connectionPoolSize: 50
  timeout: 3000
  retryAttempts: 3
  retryInterval: 1500
codec: !<org.redisson.codec.JsonJacksonCodec> {}

关键实现要点

数据一致性保证

• 使用分布式锁确保同一数据在同一时间只在一个数据中心被修改
• 采用最后写入胜出(LWW)或版本向量机制处理冲突
• 实现事务日志和补偿机制

故障转移策略

• 健康检查失败自动切换到备机
• 会话保持和用户重定向
• 数据同步断点续传

性能优化

• 本地读写优先，减少网络开销
• 异步批量数据同步
• 缓存优化和预热

监控与告警

@Component
public class DualActiveMonitor {
    @Autowired
    private MeterRegistry meterRegistry;
    @Scheduled(fixedDelay = 60000)
    public void collectMetrics() {
        // 记录数据中心状态指标
        Gauge.builder("data.center.status", 
            () -> dataCenterConfig.isPrimary() ? 1 : 0)
            .register(meterRegistry);
        // 记录同步延迟
        Gauge.builder("data.sync.latency", 
            () -> getSyncLatency())
            .register(meterRegistry);
    }
}

最佳实践建议

1. 分区策略：明确哪些数据需要在数据中心间同步，哪些可以本地分区
2. 冲突解决：定义清晰的冲突解决策略，如时间戳优先、版本号优先等
3. 可观测性：完善日志、指标和链路追踪，便于问题排查
4. 灰度发布：支持流量灰度，逐步切换流量到新数据中心
5. 定期演练：定期进行故障转移演练，确保双活机制的有效性

这个双活架构案例提供了完整的实现框架,您可以根据实际业务需求和基础设施进行调整和优化。