快捷導(dǎo)航

C#實(shí)現(xiàn)兩個(gè)DocumentDB實(shí)例之間同步數(shù)據(jù)的解決方案

更新時(shí)間：2026年02月04日 08:54:10 作者：weixin_30777913

這篇文章主要為大家詳細(xì)介紹了C#實(shí)現(xiàn)兩個(gè)DocumentDB實(shí)例之間同步數(shù)據(jù)的解決方案,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

準(zhǔn)備工作

安裝驅(qū)動(dòng)

Install-Package MongoDB.Driver -Version 2.19.0

使用 MongoDB .NET Driver 兼容 DocumentDB

下載證書(shū)

wget https://s3.amazonaws.com/rds-downloads/rds-combined-ca-bundle.pem

DocumentDB 要求 SSL 連接，需 Amazon RDS 根證書(shū)

using MongoDB.Bson;
using MongoDB.Driver;
using MongoDB.Driver.Linq;
using System.Text.Json;
using System.Text.Json.Serialization;

namespace DocumentDBSyncTool
{
    // JSON日志記錄類
    public class SyncLogEntry
    {
        [JsonPropertyName("timestamp")]
        public DateTime Timestamp { get; set; }

        [JsonPropertyName("level")]
        public string Level { get; set; } = "INFO";

        [JsonPropertyName("operation")]
        public string Operation { get; set; } = "";

        [JsonPropertyName("source_id")]
        public string? SourceId { get; set; }

        [JsonPropertyName("batch_start_id")]
        public string? BatchStartId { get; set; }

        [JsonPropertyName("batch_end_id")]
        public string? BatchEndId { get; set; }

        [JsonPropertyName("records_processed")]
        public int RecordsProcessed { get; set; }

        [JsonPropertyName("duration_ms")]
        public long DurationMs { get; set; }

        [JsonPropertyName("error")]
        public string? Error { get; set; }

        [JsonPropertyName("message")]
        public string Message { get; set; } = "";

        [JsonPropertyName("checkpoint")]
        public string? Checkpoint { get; set; }
    }

    // 斷點(diǎn)續(xù)傳狀態(tài)類
    public class CheckpointState
    {
        [JsonPropertyName("last_successful_id")]
        public string LastSuccessfulId { get; set; } = "";

        [JsonPropertyName("last_sync_timestamp")]
        public DateTime LastSyncTimestamp { get; set; }

        [JsonPropertyName("total_processed")]
        public long TotalProcessed { get; set; }

        [JsonPropertyName("last_error")]
        public string? LastError { get; set; }

        [JsonPropertyName("retry_count")]
        public int RetryCount { get; set; }
    }

    // DocumentDB同步器主類
    public class DocumentDBSynchronizer
    {
        private readonly IMongoCollection<BsonDocument> _sourceCollection;
        private readonly IMongoCollection<BsonDocument> _targetCollection;
        private readonly string _checkpointFilePath;
        private readonly string _logFilePath;
        private readonly int _batchSize;
        private readonly int _maxRetries;
        private readonly int _retryDelayMs;
        private CheckpointState _checkpointState;
        private readonly object _logLock = new object();

        public DocumentDBSynchronizer(
            MongoClient sourceClient, 
            MongoClient targetClient,
            string databaseName,
            string collectionName,
            string checkpointFilePath = "checkpoint.json",
            string logFilePath = "sync_log.json",
            int batchSize = 1000,
            int maxRetries = 5,
            int retryDelayMs = 5000)
        {
            var sourceDatabase = sourceClient.GetDatabase(databaseName);
            var targetDatabase = targetClient.GetDatabase(databaseName);
            
            _sourceCollection = sourceDatabase.GetCollection<BsonDocument>(collectionName);
            _targetCollection = targetDatabase.GetCollection<BsonDocument>(collectionName);
            
            _checkpointFilePath = checkpointFilePath;
            _logFilePath = logFilePath;
            _batchSize = batchSize;
            _maxRetries = maxRetries;
            _retryDelayMs = retryDelayMs;
            
            LoadCheckpoint();
        }

        // 加載檢查點(diǎn)狀態(tài)
        private void LoadCheckpoint()
        {
            try
            {
                if (File.Exists(_checkpointFilePath))
                {
                    var json = File.ReadAllText(_checkpointFilePath);
                    _checkpointState = JsonSerializer.Deserialize<CheckpointState>(json) 
                        ?? new CheckpointState();
                    
                    LogSyncInfo($"Loaded checkpoint from {_checkpointFilePath}. Last ID: {_checkpointState.LastSuccessfulId}");
                }
                else
                {
                    _checkpointState = new CheckpointState
                    {
                        LastSuccessfulId = "",
                        LastSyncTimestamp = DateTime.MinValue,
                        TotalProcessed = 0,
                        RetryCount = 0
                    };
                    
                    LogSyncInfo("No checkpoint found. Starting from beginning.");
                }
            }
            catch (Exception ex)
            {
                LogSyncError("Failed to load checkpoint", ex);
                _checkpointState = new CheckpointState();
            }
        }

        // 保存檢查點(diǎn)狀態(tài)
        private void SaveCheckpoint(string lastId)
        {
            try
            {
                _checkpointState.LastSuccessfulId = lastId;
                _checkpointState.LastSyncTimestamp = DateTime.UtcNow;
                _checkpointState.LastError = null;
                _checkpointState.RetryCount = 0;

                var json = JsonSerializer.Serialize(_checkpointState, new JsonSerializerOptions
                {
                    WriteIndented = true
                });
                
                File.WriteAllText(_checkpointFilePath, json);
                LogSyncInfo($"Checkpoint saved: {lastId}");
            }
            catch (Exception ex)
            {
                LogSyncError("Failed to save checkpoint", ex);
            }
        }

        // 記錄同步信息日志
        private void LogSyncInfo(string message, string? operation = null, 
            string? sourceId = null, int recordsProcessed = 0, 
            long durationMs = 0, string? batchStartId = null, 
            string? batchEndId = null)
        {
            var logEntry = new SyncLogEntry
            {
                Timestamp = DateTime.UtcNow,
                Level = "INFO",
                Operation = operation ?? "SYNC",
                SourceId = sourceId,
                BatchStartId = batchStartId,
                BatchEndId = batchEndId,
                RecordsProcessed = recordsProcessed,
                DurationMs = durationMs,
                Message = message,
                Checkpoint = _checkpointState.LastSuccessfulId
            };

            AppendLog(logEntry);
        }

        // 記錄錯(cuò)誤日志
        private void LogSyncError(string message, Exception? ex = null, 
            string? operation = null, string? sourceId = null)
        {
            var logEntry = new SyncLogEntry
            {
                Timestamp = DateTime.UtcNow,
                Level = "ERROR",
                Operation = operation ?? "SYNC",
                SourceId = sourceId,
                Message = message,
                Error = ex?.ToString(),
                Checkpoint = _checkpointState.LastSuccessfulId
            };

            AppendLog(logEntry);

            // 更新檢查點(diǎn)中的錯(cuò)誤信息
            _checkpointState.LastError = $"{message}: {ex?.Message}";
            _checkpointState.RetryCount++;
        }

        // 追加日志到文件
        private void AppendLog(SyncLogEntry logEntry)
        {
            lock (_logLock)
            {
                try
                {
                    var logLine = JsonSerializer.Serialize(logEntry);
                    
                    // 使用追加模式寫(xiě)入日志
                    using var writer = new StreamWriter(_logFilePath, true);
                    writer.WriteLine(logLine);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"Failed to write log: {ex.Message}");
                }
            }
        }

        // 獲取下一批數(shù)據(jù)
        private async Task<List<BsonDocument>> GetNextBatchAsync(string lastId, int retryCount = 0)
        {
            var stopwatch = System.Diagnostics.Stopwatch.StartNew();
            
            try
            {
                var filter = string.IsNullOrEmpty(lastId)
                    ? Builders<BsonDocument>.Filter.Empty
                    : Builders<BsonDocument>.Filter.Gt("_id", lastId);

                var sort = Builders<BsonDocument>.Sort.Ascending("_id");
                
                var batch = await _sourceCollection
                    .Find(filter)
                    .Sort(sort)
                    .Limit(_batchSize)
                    .ToListAsync();

                stopwatch.Stop();
                
                LogSyncInfo($"Fetched {batch.Count} records from source",
                    operation: "FETCH",
                    batchStartId: batch.FirstOrDefault()?["_id"].ToString(),
                    batchEndId: batch.LastOrDefault()?["_id"].ToString(),
                    recordsProcessed: batch.Count,
                    durationMs: stopwatch.ElapsedMilliseconds);

                return batch;
            }
            catch (Exception ex) when (retryCount < _maxRetries)
            {
                stopwatch.Stop();
                LogSyncError($"Failed to fetch batch (attempt {retryCount + 1}/{_maxRetries})", 
                    ex, "FETCH");
                
                await Task.Delay(_retryDelayMs * (retryCount + 1));
                return await GetNextBatchAsync(lastId, retryCount + 1);
            }
            catch (Exception ex)
            {
                stopwatch.Stop();
                throw new InvalidOperationException($"Failed to fetch batch after {_maxRetries} attempts", ex);
            }
        }

        // 批量寫(xiě)入目標(biāo)數(shù)據(jù)庫(kù)
        private async Task WriteBatchAsync(List<BsonDocument> batch, int retryCount = 0)
        {
            var stopwatch = System.Diagnostics.Stopwatch.StartNew();
            
            if (batch.Count == 0)
                return;

            try
            {
                var operations = new List<WriteModel<BsonDocument>>();
                
                foreach (var document in batch)
                {
                    var filter = Builders<BsonDocument>.Filter.Eq("_id", document["_id"]);
                    var update = new ReplaceOneModel<BsonDocument>(filter, document)
                    {
                        IsUpsert = true
                    };
                    operations.Add(update);
                }

                var result = await _targetCollection.BulkWriteAsync(operations, 
                    new BulkWriteOptions { IsOrdered = false });

                stopwatch.Stop();
                
                LogSyncInfo($"Written {result.Upserts.Count + result.ModifiedCount} records to target",
                    operation: "WRITE",
                    batchStartId: batch.First()["_id"].ToString(),
                    batchEndId: batch.Last()["_id"].ToString(),
                    recordsProcessed: batch.Count,
                    durationMs: stopwatch.ElapsedMilliseconds);
            }
            catch (Exception ex) when (retryCount < _maxRetries)
            {
                stopwatch.Stop();
                LogSyncError($"Failed to write batch (attempt {retryCount + 1}/{_maxRetries})", 
                    ex, "WRITE");
                
                await Task.Delay(_retryDelayMs * (retryCount + 1));
                await WriteBatchAsync(batch, retryCount + 1);
            }
            catch (Exception ex)
            {
                stopwatch.Stop();
                throw new InvalidOperationException($"Failed to write batch after {_maxRetries} attempts", ex);
            }
        }

        // 驗(yàn)證連接
        private async Task<bool> ValidateConnectionsAsync()
        {
            try
            {
                LogSyncInfo("Validating source connection...", "VALIDATE");
                var sourceCount = await _sourceCollection.EstimatedDocumentCountAsync();
                
                LogSyncInfo("Validating target connection...", "VALIDATE");
                var targetCount = await _targetCollection.EstimatedDocumentCountAsync();
                
                LogSyncInfo($"Connections validated. Source: ~{sourceCount} records, Target: ~{targetCount} records", 
                    "VALIDATE");
                return true;
            }
            catch (Exception ex)
            {
                LogSyncError("Connection validation failed", ex, "VALIDATE");
                return false;
            }
        }

        // 執(zhí)行同步
        public async Task<bool> StartSyncAsync(CancellationToken cancellationToken = default)
        {
            LogSyncInfo("Starting synchronization process", "START");
            
            if (!await ValidateConnectionsAsync())
            {
                LogSyncError("Connection validation failed. Aborting sync.", null, "START");
                return false;
            }

            try
            {
                bool hasMoreData = true;
                long totalProcessed = 0;
                string lastProcessedId = _checkpointState.LastSuccessfulId;

                while (hasMoreData && !cancellationToken.IsCancellationRequested)
                {
                    var batch = await GetNextBatchAsync(lastProcessedId);
                    
                    if (batch.Count == 0)
                    {
                        hasMoreData = false;
                        LogSyncInfo("No more data to sync", "COMPLETE", 
                            recordsProcessed: (int)totalProcessed);
                        break;
                    }

                    await WriteBatchAsync(batch);
                    
                    // 更新檢查點(diǎn)
                    lastProcessedId = batch.Last()["_id"].ToString();
                    SaveCheckpoint(lastProcessedId);
                    
                    totalProcessed += batch.Count;
                    _checkpointState.TotalProcessed = totalProcessed;

                    // 避免過(guò)度占用資源
                    if (batch.Count == _batchSize)
                    {
                        await Task.Delay(100, cancellationToken);
                    }
                }

                if (cancellationToken.IsCancellationRequested)
                {
                    LogSyncInfo("Sync cancelled by user", "CANCELLED", 
                        recordsProcessed: (int)totalProcessed);
                    return false;
                }

                LogSyncInfo($"Sync completed successfully. Total processed: {totalProcessed}", 
                    "COMPLETE", recordsProcessed: (int)totalProcessed);
                return true;
            }
            catch (Exception ex)
            {
                LogSyncError("Sync failed", ex, "ERROR");
                return false;
            }
        }

        // 獲取同步統(tǒng)計(jì)信息
        public async Task<SyncStatistics> GetStatisticsAsync()
        {
            try
            {
                var sourceCount = await _sourceCollection.EstimatedDocumentCountAsync();
                var targetCount = await _targetCollection.EstimatedDocumentCountAsync();
                
                return new SyncStatistics
                {
                    SourceCount = sourceCount,
                    TargetCount = targetCount,
                    LastCheckpoint = _checkpointState.LastSuccessfulId,
                    TotalProcessed = _checkpointState.TotalProcessed,
                    LastSyncTime = _checkpointState.LastSyncTimestamp
                };
            }
            catch (Exception ex)
            {
                LogSyncError("Failed to get statistics", ex, "STATS");
                return new SyncStatistics();
            }
        }
    }

    // 統(tǒng)計(jì)信息類
    public class SyncStatistics
    {
        [JsonPropertyName("source_count")]
        public long SourceCount { get; set; }
        
        [JsonPropertyName("target_count")]
        public long TargetCount { get; set; }
        
        [JsonPropertyName("last_checkpoint")]
        public string LastCheckpoint { get; set; } = "";
        
        [JsonPropertyName("total_processed")]
        public long TotalProcessed { get; set; }
        
        [JsonPropertyName("last_sync_time")]
        public DateTime LastSyncTime { get; set; }
        
        [JsonPropertyName("sync_status")]
        public string SyncStatus => SourceCount == TargetCount ? "In Sync" : "Out of Sync";
        
        [JsonPropertyName("records_difference")]
        public long RecordsDifference => Math.Abs(SourceCount - TargetCount);
    }

    // DocumentDB連接工廠
    public static class DocumentDBConnectionFactory
    {
        public static MongoClient CreateClient(string username, string password, 
            string host, string databaseName, bool allowInsecureTls = false)
        {
            var settings = new MongoClientSettings
            {
                Servers = new[] { MongoServerAddress.Parse(host) },
                Credential = MongoCredential.CreateCredential(
                    databaseName,
                    username,
                    password
                ),
                ConnectionMode = ConnectionMode.ReplicaSet,
                ReadPreference = ReadPreference.SecondaryPreferred,
                SslSettings = new SslSettings
                {
                    EnabledSslProtocols = System.Security.Authentication.SslProtocols.Tls12
                },
                UseTls = true,
                ServerSelectionTimeout = TimeSpan.FromSeconds(30),
                MaxConnectionPoolSize = 100,
                MinConnectionPoolSize = 5,
                SocketTimeout = TimeSpan.FromSeconds(60),
                ConnectTimeout = TimeSpan.FromSeconds(30)
            };

            if (allowInsecureTls)
            {
                settings.AllowInsecureTls = true;
            }
            else
            {
                // 生產(chǎn)環(huán)境中應(yīng)該使用正確的證書(shū)驗(yàn)證
                // settings.SslSettings.ClientCertificates = LoadCertificates();
            }

            return new MongoClient(settings);
        }
    }

    // 主程序
    class Program
    {
        static async Task Main(string[] args)
        {
            // 配置參數(shù) - 可以從配置文件或環(huán)境變量讀取
            var sourceConfig = new
            {
                Username = "admin",
                Password = "source_password",
                Host = "docdb-source.cluster-xxxxxx.us-east-1.docdb.amazonaws.com",
                Database = "testdb"
            };

            var targetConfig = new
            {
                Username = "admin",
                Password = "target_password",
                Host = "docdb-target.cluster-xxxxxx.us-east-1.docdb.amazonaws.com",
                Database = "testdb"
            };

            var collectionName = "users";
            var checkpointFile = "sync_checkpoint.json";
            var logFile = $"sync_log_{DateTime.Now:yyyyMMdd_HHmmss}.json";
            var batchSize = 500; // 根據(jù)網(wǎng)絡(luò)狀況調(diào)整

            Console.WriteLine("Starting DocumentDB Sync Tool...");
            Console.WriteLine($"Log file: {logFile}");
            Console.WriteLine($"Checkpoint file: {checkpointFile}");

            try
            {
                // 創(chuàng)建連接
                var sourceClient = DocumentDBConnectionFactory.CreateClient(
                    sourceConfig.Username,
                    sourceConfig.Password,
                    sourceConfig.Host,
                    sourceConfig.Database,
                    allowInsecureTls: true // 開(kāi)發(fā)環(huán)境，生產(chǎn)環(huán)境應(yīng)使用證書(shū)
                );

                var targetClient = DocumentDBConnectionFactory.CreateClient(
                    targetConfig.Username,
                    targetConfig.Password,
                    targetConfig.Host,
                    targetConfig.Database,
                    allowInsecureTls: true
                );

                // 創(chuàng)建同步器
                var synchronizer = new DocumentDBSynchronizer(
                    sourceClient,
                    targetClient,
                    sourceConfig.Database,
                    collectionName,
                    checkpointFilePath: checkpointFile,
                    logFilePath: logFile,
                    batchSize: batchSize,
                    maxRetries: 3,
                    retryDelayMs: 3000
                );

                // 設(shè)置取消令牌（支持CTRL+C中斷）
                var cts = new CancellationTokenSource();
                Console.CancelKeyPress += (sender, e) =>
                {
                    Console.WriteLine("\nCancelling sync...");
                    cts.Cancel();
                    e.Cancel = true;
                };

                // 執(zhí)行同步
                var success = await synchronizer.StartSyncAsync(cts.Token);

                // 顯示統(tǒng)計(jì)信息
                var stats = await synchronizer.GetStatisticsAsync();
                Console.WriteLine("\n=== Sync Statistics ===");
                Console.WriteLine($"Source records: {stats.SourceCount}");
                Console.WriteLine($"Target records: {stats.TargetCount}");
                Console.WriteLine($"Last checkpoint: {stats.LastCheckpoint}");
                Console.WriteLine($"Total processed: {stats.TotalProcessed}");
                Console.WriteLine($"Sync status: {stats.SyncStatus}");
                Console.WriteLine($"Difference: {stats.RecordsDifference}");

                if (success)
                {
                    Console.WriteLine("\n? Sync completed successfully!");
                }
                else
                {
                    Console.WriteLine("\n? Sync completed with errors.");
                    Console.WriteLine("Check the log file for details.");
                }

                Console.WriteLine($"\nLog saved to: {logFile}");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"\n? Fatal error: {ex.Message}");
                Console.WriteLine($"Stack trace: {ex.StackTrace}");
            }
        }
    }
}

配置文件示例 (appsettings.json)

{
  "SyncConfig": {
    "SourceDb": {
      "Username": "admin",
      "Password": "source_password",
      "Host": "docdb-source.cluster-xxxxxx.us-east-1.docdb.amazonaws.com",
      "Database": "testdb",
      "Collection": "users"
    },
    "TargetDb": {
      "Username": "admin",
      "Password": "target_password",
      "Host": "docdb-target.cluster-xxxxxx.us-east-1.docdb.amazonaws.com",
      "Database": "testdb",
      "Collection": "users"
    },
    "SyncOptions": {
      "BatchSize": 1000,
      "MaxRetries": 5,
      "RetryDelayMs": 5000,
      "CheckpointFile": "checkpoint.json",
      "LogDirectory": "logs",
      "EnableValidation": true
    }
  }
}

項(xiàng)目文件 (DocumentDBSyncTool.csproj)

<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net8.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="MongoDB.Driver" Version="2.19.0" />
    <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="8.0.0" />
    <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
    <PackageReference Include="System.CommandLine" Version="2.0.0-beta4.22272.1" />
  </ItemGroup>

  <ItemGroup>
    <None Update="appsettings.json">
      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
    </None>
  </ItemGroup>

</Project>

使用說(shuō)明

1.編譯和運(yùn)行

dotnet build
dotnet run

2.帶參數(shù)運(yùn)行

dotnet run -- --source-host docdb-source.amazonaws.com --target-host docdb-target.amazonaws.com

3.特性說(shuō)明

斷點(diǎn)續(xù)傳功能

自動(dòng)保存檢查點(diǎn)（checkpoint.json）
程序重啟后從上次成功的位置繼續(xù)
記錄每次同步的最后一個(gè)ID

容錯(cuò)機(jī)制

自動(dòng)重試失敗的批次（可配置重試次數(shù)）
指數(shù)退避重試延遲
網(wǎng)絡(luò)中斷恢復(fù)

JSON日志記錄

結(jié)構(gòu)化日志，便于分析
包含時(shí)間戳、操作類型、記錄數(shù)量等信息
錯(cuò)誤信息詳細(xì)記錄

性能優(yōu)化

批量讀取和寫(xiě)入
連接池管理
異步操作，支持并發(fā)

監(jiān)控和統(tǒng)計(jì)

實(shí)時(shí)顯示同步進(jìn)度
生成統(tǒng)計(jì)報(bào)告
驗(yàn)證源和目標(biāo)記錄數(shù)量

4.生產(chǎn)環(huán)境建議

安全性：

使用AWS Secrets Manager存儲(chǔ)密碼
啟用TLS證書(shū)驗(yàn)證
限制網(wǎng)絡(luò)訪問(wèn)權(quán)限

監(jiān)控：

集成CloudWatch監(jiān)控
設(shè)置報(bào)警規(guī)則
定期審計(jì)日志

性能：

根據(jù)網(wǎng)絡(luò)延遲調(diào)整批次大小
監(jiān)控DocumentDB實(shí)例性能指標(biāo)
考慮使用分片集合處理大量數(shù)據(jù)

部署：

作為Windows服務(wù)或Linux守護(hù)進(jìn)程運(yùn)行
使用AWS ECS或Lambda進(jìn)行無(wú)服務(wù)器部署
配置自動(dòng)重啟策略

到此這篇關(guān)于C#實(shí)現(xiàn)兩個(gè)DocumentDB實(shí)例之間同步數(shù)據(jù)的解決方案的文章就介紹到這了,更多相關(guān)C# DocumentDB數(shù)據(jù)同步內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: