📖 数据密集型设计

分布式数据库原理与实践

深入探讨分布式数据库架构与分片策略

一、分布式数据库概述

分布式数据库是将数据分散存储在多个节点上的数据库系统，能够提供高可用、高可扩展的数据存储能力。在数据密集型应用中，分布式数据库是处理海量数据的关键基础设施。

二、分布式数据库架构

2.1 分布式数据库架构类型

架构类型 描述 优点 缺点 代表系统
分片式 数据按分片键分布 水平扩展 跨分片查询复杂 MySQL分片、TiDB
分布式SQL 兼容SQL接口 易用性好 复杂度高 TiDB、CockroachDB
NewSQL 分布式+ACID 强一致性 性能开销 TiDB、Spanner
NoSQL 非关系型数据模型 高吞吐、灵活 弱一致性 MongoDB、Cassandra

2.2 分布式数据库架构图

graph TD A[客户端] --> B[分布式数据库] B --> C[协调层] C --> D[分片1] C --> E[分片2] C --> F[分片3] D --> D1[主节点] D --> D2[从节点] E --> E1[主节点] E --> E2[从节点] F --> F1[主节点] F --> F2[从节点] C --> G[元数据管理] G --> H[分片映射]

三、分片策略

3.1 分片策略对比

分片策略 描述 优点 缺点 适用场景
范围分片 按范围划分 范围查询高效 热点问题 时间序列数据
哈希分片 按哈希值划分 分布均匀 范围查询低效 随机访问数据
列表分片 按列表值划分 灵活可控 维护成本高 业务分类数据
复合分片 多种策略组合 兼顾各种场景 复杂度高 复杂业务场景

3.2 分片策略实现

public class ShardingStrategy
{
    public int GetShardIdByRange(long value, int shardCount)
    {
        var rangeSize = long.MaxValue / shardCount;
        
        return (int)(value / rangeSize);
    }
    
    public int GetShardIdByHash(string key, int shardCount)
    {
        var hash = key.GetHashCode();
        
        return Math.Abs(hash % shardCount);
    }
    
    public int GetShardIdByList(string key, Dictionary listMapping)
    {
        return listMapping.TryGetValue(key, out var shardId) ? shardId : 0;
    }
    
    public int GetShardIdByComposite(string key, long rangeValue, int shardCount)
    {
        var hashShard = GetShardIdByHash(key, shardCount);
        var rangeShard = GetShardIdByRange(rangeValue, shardCount);
        
        return (hashShard + rangeShard) % shardCount;
    }
    
    public async Task GetShardInfoAsync(int shardId)
    {
        return await _metadataRepository.GetShardInfoAsync(shardId);
    }
    
    public async Task> GetAllShardsAsync()
    {
        return await _metadataRepository.GetAllShardsAsync();
    }
}

public class ShardInfo
{
    public int ShardId { get; set; }
    public string ConnectionString { get; set; }
    public string Host { get; set; }
    public int Port { get; set; }
    public string DatabaseName { get; set; }
    public ShardStatus Status { get; set; }
}

public enum ShardStatus { Online, Offline, Maintenance }

3.3 分片键选择

public class ShardingKeySelector
{
    public string SelectShardingKey(string tableName)
    {
        return tableName switch
        {
            "orders" => "user_id",
            "products" => "category_id",
            "users" => "id",
            "transactions" => "transaction_date",
            _ => throw new ArgumentException("No sharding key defined for this table")
        };
    }
    
    public ShardingStrategyType SelectStrategy(string tableName)
    {
        return tableName switch
        {
            "orders" => ShardingStrategyType.Hash,
            "products" => ShardingStrategyType.List,
            "users" => ShardingStrategyType.Hash,
            "transactions" => ShardingStrategyType.Range,
            _ => ShardingStrategyType.Hash
        };
    }
    
    public int CalculateShardCount(string tableName, long estimatedRows)
    {
        var rowsPerShard = 10000000;
        
        return (int)Math.Ceiling(estimatedRows / (double)rowsPerShard);
    }
}

public enum ShardingStrategyType { Range, Hash, List, Composite }

四、分布式事务

4.1 分布式事务模式

graph TD A[分布式事务] --> B[2PC] A --> C[3PC] A --> D[Saga] A --> E[TCC] A --> F[本地消息表] B --> B1[协调者] B --> B2[参与者1] B --> B3[参与者2] D --> D1[事务管理器] D --> D2[补偿事务] F --> F1[消息队列]

4.2 分布式事务实现

public class DistributedTransactionService
{
    public async Task ExecuteTwoPhaseCommitAsync(List participants)
    {
        try
        {
            await PreparePhaseAsync(participants);
            await CommitPhaseAsync(participants);
            
            return new TransactionResult { Success = true };
        }
        catch (Exception)
        {
            await RollbackPhaseAsync(participants);
            return new TransactionResult { Success = false };
        }
    }
    
    private async Task PreparePhaseAsync(List participants)
    {
        foreach (var participant in participants)
        {
            await participant.PrepareAsync();
        }
    }
    
    private async Task CommitPhaseAsync(List participants)
    {
        foreach (var participant in participants)
        {
            await participant.CommitAsync();
        }
    }
    
    private async Task RollbackPhaseAsync(List participants)
    {
        foreach (var participant in participants)
        {
            await participant.RollbackAsync();
        }
    }
    
    public async Task ExecuteSagaAsync(List steps)
    {
        var executedSteps = new List();
        
        try
        {
            foreach (var step in steps)
            {
                await step.ExecuteAsync();
                executedSteps.Add(step);
            }
            
            return new TransactionResult { Success = true };
        }
        catch (Exception)
        {
            foreach (var step in executedSteps.Reverse())
            {
                await step.CompensateAsync();
            }
            
            return new TransactionResult { Success = false };
        }
    }
}

public interface ITransactionParticipant
{
    Task PrepareAsync();
    Task CommitAsync();
    Task RollbackAsync();
}

public class SagaStep
{
    public Func Execute { get; set; }
    public Func Compensate { get; set; }
    
    public async Task ExecuteAsync() => await Execute();
    public async Task CompensateAsync() => await Compensate();
}

五、数据一致性

5.1 一致性级别

一致性级别 描述 优点 缺点 适用场景
强一致性 写入后立即一致 数据准确 性能开销大 金融交易
最终一致性 一段时间后一致 性能好 存在不一致窗口 社交、电商
弱一致性 不保证一致 性能最优 数据可能不一致 日志、统计
因果一致性 因果关系一致 兼顾性能和一致性 实现复杂 协作系统

5.2 一致性保障策略

public class ConsistencyService
{
    public async Task EnsureStrongConsistencyAsync(Func operation)
    {
        await _distributedLockService.AcquireLockAsync("consistency_lock");
        
        try
        {
            await operation();
        }
        finally
        {
            await _distributedLockService.ReleaseLockAsync();
        }
    }
    
    public async Task EnsureEventualConsistencyAsync(string key, object value)
    {
        await _databaseService.UpdateAsync(key, value);
        
        await _messageQueueService.PublishAsync("data_changed", new { Key = key, Value = value });
        
        await Task.Delay(1000);
        
        await _cacheService.SetAsync(key, value);
    }
    
    public async Task SyncDataAsync(string sourceShard, string targetShard, string tableName)
    {
        var data = await _databaseService.GetDataAsync(sourceShard, tableName);
        
        await _databaseService.SetDataAsync(targetShard, tableName, data);
    }
    
    public async Task VerifyConsistencyAsync(string key)
    {
        var dbValue = await _databaseService.GetAsync(key);
        var cacheValue = await _cacheService.GetAsync(key);
        
        if (!Equals(dbValue, cacheValue))
        {
            await _cacheService.SetAsync(key, dbValue);
        }
    }
}

六、分布式数据库监控

6.1 分布式数据库监控

public class DistributedDatabaseMonitor
{
    public async Task GetMetricsAsync()
    {
        var shards = await _shardingService.GetAllShardsAsync();
        var metrics = new DatabaseMetrics();
        
        foreach (var shard in shards)
        {
            var shardMetrics = await _shardMonitor.GetShardMetricsAsync(shard.ShardId);
            
            metrics.TotalConnections += shardMetrics.Connections;
            metrics.TotalQueries += shardMetrics.Queries;
            metrics.TotalSlowQueries += shardMetrics.SlowQueries;
            metrics.AverageLatency += shardMetrics.Latency;
            
            if (shard.Status == ShardStatus.Offline)
            {
                metrics.OfflineShards++;
            }
        }
        
        metrics.AverageLatency /= shards.Count;
        
        return metrics;
    }
    
    public async Task MonitorAsync()
    {
        var metrics = await GetMetricsAsync();
        
        if (metrics.OfflineShards > 0)
        {
            await _alertService.SendAlert("分布式数据库分片离线", 
                $"离线分片数: {metrics.OfflineShards}");
        }
        
        if (metrics.AverageLatency > 500)
        {
            await _alertService.SendAlert("分布式数据库延迟过高", 
                $"平均延迟: {metrics.AverageLatency}ms");
        }
    }
}

public class DatabaseMetrics
{
    public int TotalConnections { get; set; }
    public long TotalQueries { get; set; }
    public long TotalSlowQueries { get; set; }
    public double AverageLatency { get; set; }
    public int OfflineShards { get; set; }
}

七、分布式数据库最佳实践

7.1 分片设计最佳实践

  • 选择合适的分片键
  • 避免跨分片查询
  • 预留分片扩展空间
  • 定期检查分片分布
  • 考虑数据迁移方案

7.2 事务处理最佳实践

public class TransactionBestPractices
{
    public async Task ExecuteWithRetryAsync(Func operation, int maxRetries = 3)
    {
        var retryCount = 0;
        
        while (retryCount < maxRetries)
        {
            try
            {
                await operation();
                return;
            }
            catch (Exception)
            {
                retryCount++;
                await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, retryCount)));
            }
        }
        
        throw new Exception("Transaction failed after retries");
    }
    
    public async Task ExecuteWithIdempotencyAsync(string transactionId, Func operation)
    {
        var executed = await _cacheService.GetAsync($"tx:{transactionId}");
        
        if (executed)
        {
            return;
        }
        
        try
        {
            await operation();
            await _cacheService.SetAsync($"tx:{transactionId}", true, TimeSpan.FromDays(7));
        }
        catch (Exception)
        {
            await _cacheService.RemoveAsync($"tx:{transactionId}");
            throw;
        }
    }
}

7.3 一致性最佳实践

  • 根据业务需求选择一致性级别
  • 使用消息队列保证最终一致性
  • 定期校验数据一致性
  • 实现幂等性操作
  • 考虑数据同步延迟

八、总结

分布式数据库是数据密集型应用中处理海量数据的关键基础设施。通过合理选择分片策略、实现分布式事务、保障数据一致性，能够构建高可用、高可扩展的分布式数据库系统。定期监控、维护分片、优化查询，能够保障分布式数据库的高效运行。