数据复制策略与实现

一、数据复制概述

数据复制是将数据从一个数据库节点复制到其他节点的技术，能够提高数据可用性、实现读写分离、提供灾难恢复能力。数据复制是构建高可用数据库架构的关键技术。

二、复制策略对比

2.1 复制策略对比表

2.2 复制策略选择

三、同步复制

3.1 同步复制原理

同步复制等待所有副本确认后才返回成功：

3.2 MySQL同步复制配置

-- 主节点配置
server-id = 1
log_bin = /var/log/mysql/mysql-bin.log
binlog_format = ROW
sync_binlog = 1
innodb_flush_log_at_trx_commit = 1

-- 从节点配置
server-id = 2
relay_log = /var/log/mysql/relay-bin.log
read_only = 1

-- 启用半同步复制（MySQL 5.6+）
plugin-load-add = rpl_semi_sync_master.so
plugin-load-add = rpl_semi_sync_slave.so

-- 主节点设置同步模式
SET GLOBAL rpl_semi_sync_master_enabled = 1;
SET GLOBAL rpl_semi_sync_master_timeout = 10000;

3.3 PostgreSQL同步复制配置

-- postgresql.conf 主节点配置
wal_level = replica
max_wal_senders = 10
wal_keep_size = 1GB

-- postgresql.conf 从节点配置
hot_standby = on
max_standby_streaming_delay = 30s

-- pg_hba.conf 允许复制连接
host replication repuser 192.168.1.0/24 md5

-- 设置同步复制
ALTER SYSTEM SET synchronous_standby_names = 'slave1, slave2';
SELECT pg_reload_conf();

-- 查看同步状态
SELECT usename, application_name, state, sync_state FROM pg_stat_replication;

四、异步复制

4.1 异步复制原理

异步复制不等待副本确认，立即返回成功：

4.2 MySQL异步复制配置

-- 主节点配置
server-id = 1
log_bin = /var/log/mysql/mysql-bin.log
binlog_format = ROW

-- 创建复制用户
CREATE USER 'repl'@'%' IDENTIFIED BY 'password';
GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';

-- 查看主节点状态
SHOW MASTER STATUS;

-- 从节点配置
server-id = 2
relay_log = /var/log/mysql/relay-bin.log
read_only = 1

-- 配置复制
CHANGE MASTER TO
    MASTER_HOST='master_host',
    MASTER_USER='repl',
    MASTER_PASSWORD='password',
    MASTER_LOG_FILE='mysql-bin.000001',
    MASTER_LOG_POS=107;

-- 启动复制
START SLAVE;

-- 查看复制状态
SHOW SLAVE STATUS\G

4.3 MongoDB异步复制

# 创建副本集配置
rs.initiate({
    _id: "rs0",
    members: [
        { _id: 0, host: "mongo1:27017" },
        { _id: 1, host: "mongo2:27017" },
        { _id: 2, host: "mongo3:27017" }
    ]
})

# 查看副本集状态
rs.status()

# 设置读写偏好
db.collection.find().readPref("secondaryPreferred")

# 强制读主节点
db.collection.find().readPref("primary")

五、半同步复制

5.1 半同步复制原理

半同步复制等待部分副本确认后返回成功：

5.2 MySQL半同步复制配置

-- 安装半同步插件
INSTALL PLUGIN rpl_semi_sync_master SONAME 'semisync_master.so';
INSTALL PLUGIN rpl_semi_sync_slave SONAME 'semisync_slave.so';

-- 主节点启用半同步
SET GLOBAL rpl_semi_sync_master_enabled = 1;
SET GLOBAL rpl_semi_sync_master_wait_for_slave_count = 1;
SET GLOBAL rpl_semi_sync_master_timeout = 10000;

-- 从节点启用半同步
SET GLOBAL rpl_semi_sync_slave_enabled = 1;

-- 重启从节点复制
STOP SLAVE IO_THREAD;
START SLAVE IO_THREAD;

-- 查看半同步状态
SHOW STATUS LIKE 'Rpl_semi_sync_master_status';
SHOW STATUS LIKE 'Rpl_semi_sync_slave_status';

5.3 半同步复制监控

public class SemiSyncReplicationMonitor
{
    public async Task<ReplicationMetrics> GetMetrics()
    {
        var masterStatus = await _dbContext.QuerySingleAsync<MasterStatus>(
            "SHOW STATUS LIKE 'Rpl_semi_sync_master_status'"
        );
        
        var slaveCount = await _dbContext.QuerySingleAsync<SlaveCount>(
            "SHOW STATUS LIKE 'Rpl_semi_sync_master_wait_for_slave_count'"
        );
        
        var slaveStatus = await _dbContext.QueryAsync<SlaveStatus>(
            "SHOW SLAVE STATUS"
        );
        
        return new ReplicationMetrics
        {
            IsSemiSyncEnabled = masterStatus.Value == "ON",
            RequiredSlaveCount = int.Parse(slaveCount.Value),
            ConnectedSlaves = slaveStatus.Count(),
            HealthySlaves = slaveStatus.Count(s => s.IoRunning == "Yes" && s.SqlRunning == "Yes")
        };
    }
}

六、多主复制

6.1 多主复制原理

多主复制允许多个节点同时接受写入：

6.2 MySQL多主复制配置

-- 节点1配置
server-id = 1
log_bin = /var/log/mysql/mysql-bin.log
binlog_format = ROW
auto_increment_offset = 1
auto_increment_increment = 3

-- 节点2配置
server-id = 2
log_bin = /var/log/mysql/mysql-bin.log
binlog_format = ROW
auto_increment_offset = 2
auto_increment_increment = 3

-- 节点3配置
server-id = 3
log_bin = /var/log/mysql/mysql-bin.log
binlog_format = ROW
auto_increment_offset = 3
auto_increment_increment = 3

-- 配置互相复制
CHANGE MASTER TO
    MASTER_HOST='node2',
    MASTER_USER='repl',
    MASTER_PASSWORD='password',
    MASTER_LOG_FILE='mysql-bin.000001',
    MASTER_LOG_POS=107;

CHANGE MASTER TO
    MASTER_HOST='node3',
    MASTER_USER='repl',
    MASTER_PASSWORD='password',
    MASTER_LOG_FILE='mysql-bin.000001',
    MASTER_LOG_POS=107 FOR CHANNEL 'node3';

6.3 PostgreSQL多主复制（使用BDR）

# 安装BDR扩展
CREATE EXTENSION bdr;

# 初始化BDR节点
SELECT bdr.bdr_group_create(
    local_node_name := 'node1',
    node_external_dsn := 'host=node1 port=5432 dbname=app'
);

# 添加其他节点
SELECT bdr.bdr_group_join(
    local_node_name := 'node2',
    node_external_dsn := 'host=node2 port=5432 dbname=app',
    join_using_dsn := 'host=node1 port=5432 dbname=app'
);

# 查看BDR状态
SELECT * FROM bdr.bdr_nodes;

6.4 多主复制冲突解决

public class MultiMasterConflictResolver
{
    public async Task ResolveConflict(ConflictInfo conflict)
    {
        switch (conflict.Type)
        {
            case ConflictType.Insert:
                await ResolveInsertConflict(conflict);
                break;
            case ConflictType.Update:
                await ResolveUpdateConflict(conflict);
                break;
            case ConflictType.Delete:
                await ResolveDeleteConflict(conflict);
                break;
        }
    }
    
    private async Task ResolveUpdateConflict(ConflictInfo conflict)
    {
        var node1Data = await GetNodeData(conflict.Node1Id, conflict.TableName, conflict.PrimaryKey);
        var node2Data = await GetNodeData(conflict.Node2Id, conflict.TableName, conflict.PrimaryKey);
        
        var latestData = node1Data.UpdatedAt > node2Data.UpdatedAt ? node1Data : node2Data;
        
        await ApplyResolvedData(conflict.TableName, conflict.PrimaryKey, latestData);
        
        await _auditLogger.LogConflictResolution(conflict, ConflictResolutionStrategy.LastWriteWins);
    }
    
    private async Task ResolveInsertConflict(ConflictInfo conflict)
    {
        // 使用UUID避免主键冲突
        await AssignNewId(conflict.Node2Id, conflict.TableName, conflict.PrimaryKey);
    }
}

public enum ConflictResolutionStrategy
{
    LastWriteWins,
    FirstWriteWins,
    Manual,
    Custom
}

七、复制拓扑结构

7.1 复制拓扑类型

7.2 级联复制拓扑

八、复制监控与管理

8.1 复制监控指标

public class ReplicationMetrics
{
    public string MasterHost { get; set; }
    public int ConnectedSlaves { get; set; }
    public int HealthySlaves { get; set; }
    public long ReplicationLagSeconds { get; set; }
    public string ReplicationMode { get; set; }
    public bool IsMasterRunning { get; set; }
}

public class ReplicationMonitor
{
    public async Task<ReplicationMetrics> GetMetrics()
    {
        var slaveStatus = await _dbContext.QueryAsync<SlaveStatus>(
            "SHOW SLAVE STATUS"
        );
        
        return new ReplicationMetrics
        {
            ConnectedSlaves = slaveStatus.Count(),
            HealthySlaves = slaveStatus.Count(s => s.IoRunning == "Yes" && s.SqlRunning == "Yes"),
            ReplicationLagSeconds = slaveStatus.Min(s => s.SecondsBehindMaster),
            IsMasterRunning = true
        };
    }
}

8.2 复制延迟管理

public class ReplicationLagManager
{
    public async Task ManageLag()
    {
        var metrics = await _monitor.GetMetrics();
        
        if (metrics.ReplicationLagSeconds > 60)
        {
            await _alertService.SendAlert("复制延迟过高", 
                $"延迟: {metrics.ReplicationLagSeconds}秒");
            
            await ReduceLag();
        }
    }
    
    private async Task ReduceLag()
    {
        await OptimizeBinlog();
        await IncreaseReplicationThreads();
        await CheckNetwork();
    }
    
    private async Task OptimizeBinlog()
    {
        await _dbContext.Database.ExecuteSqlRawAsync(
            "SET GLOBAL binlog_format = 'ROW'"
        );
    }
    
    private async Task IncreaseReplicationThreads()
    {
        await _dbContext.Database.ExecuteSqlRawAsync(
            "SET GLOBAL slave_parallel_workers = 8"
        );
    }
}

8.3 自动故障转移

public class FailoverManager
{
    public async Task HandleMasterFailure(string masterHost)
    {
        var slaves = await GetHealthySlaves();
        
        if (slaves.Count == 0)
            throw new InvalidOperationException("没有可用的从节点");
        
        var newMaster = SelectNewMaster(slaves);
        
        await PromoteToMaster(newMaster);
        await ReconfigureSlaves(newMaster);
        await UpdateApplicationConfig(newMaster);
        
        await _alertService.SendAlert("主节点故障转移完成", 
            $"新主节点: {newMaster.Host}");
    }
    
    private SlaveInfo SelectNewMaster(List<SlaveInfo> slaves)
    {
        return slaves.OrderByDescending(s => s.ReplicationLag).First();
    }
    
    private async Task PromoteToMaster(SlaveInfo slave)
    {
        await _dbContext.Database.ExecuteSqlRawAsync(
            $"STOP SLAVE; RESET MASTER;",
            new { Connection = slave.Connection }
        );
    }
    
    private async Task ReconfigureSlaves(SlaveInfo newMaster)
    {
        var otherSlaves = await GetOtherSlaves(newMaster);
        
        foreach (var slave in otherSlaves)
        {
            await _dbContext.Database.ExecuteSqlRawAsync(
                $"CHANGE MASTER TO MASTER_HOST='{newMaster.Host}'; START SLAVE;",
                new { Connection = slave.Connection }
            );
        }
    }
}

九、复制最佳实践

9.1 选择合适的复制模式

根据业务需求选择同步、异步或半同步复制。

9.2 监控复制状态

实时监控复制延迟和健康状态。

9.3 实施自动故障转移

使用MHA、Orchestrator等工具实现自动故障转移。

9.4 避免单点故障

部署多个从节点，确保高可用性。

9.5 定期测试复制

定期测试复制恢复能力，确保数据一致性。

十、总结

数据复制是构建高可用数据库架构的关键技术。同步复制提供强一致性但性能较低，异步复制性能高但只有最终一致性，半同步复制在两者之间取得平衡，多主复制适合多地域部署。合理选择复制策略、监控复制状态、实施自动故障转移，能够构建高可用、高性能的数据存储系统。