数据库不可用时的查询缓冲方案及Entity Framework 8集成咨询

通用设计模式 #

• 断路器模式：核心用于检测数据库健康状态，当连续失败达到阈值时触发"断路"，自动将查询路由至缓冲逻辑；数据库恢复后自动"闭合"，回归正常操作流程。
• 本地消息队列模式：将待执行查询作为消息持久化到文件（如.PMQ），本质是轻量级本地队列，确保查询不会丢失，后续按顺序执行。
• 重试模式：数据库恢复后，对缓冲查询进行重试，结合幂等性设计避免重复执行导致的数据异常。

现成实现方案 #

• 自定义文件队列：直接基于文件系统实现，将查询（含参数、SQL语句或实体变更）序列化为JSON/二进制格式，通过"临时文件+重命名"的原子操作写入，保证数据完整性。
• 本地队列库：使用成熟的轻量级存储库替代纯文件，比如LiteDB（文档型本地数据库）、SQLite，这类库自带事务和数据校验，比手动文件操作更可靠。
• EF第三方扩展：部分EF Core扩展支持离线缓冲，比如EntityFrameworkCore.Offline，但需验证是否兼容EF8，复杂场景建议自定义实现。

核心处理要点 #

• 查询序列化：完整保存查询的必要信息——EF实体变更需记录类型、变更状态（新增/修改/删除）、字段值；原始SQL需保存语句和参数集合，敏感数据要加密后写入。
• 缓冲文件管理：
  • 写入时用原子操作（先写.tmp临时文件，成功后重命名为.pmq），避免中途失败导致文件损坏。
  • 维护已执行记录的唯一ID列表，防止重复执行。
  • 定期清理已成功执行的文件，将失败文件移至错误目录等待人工排查。
• 幂等性保障：每个缓冲查询生成唯一ID，执行前先校验是否已处理；写入操作尽量设计为幂等（比如用业务唯一主键，重复执行不会产生重复数据）。
• 后台消费逻辑：用.NET HostedService定时检测数据库状态，恢复后按顺序读取缓冲文件执行查询；失败时设置重试次数，超过次数标记为异常。

EF8集成实践 #

1. 基于Polly实现断路器+缓冲逻辑 #

借助Polly库实现断路器，在自定义DbContext中重写SaveChangesAsync方法，捕获数据库不可用异常时序列化变更到缓冲文件：

using Microsoft.EntityFrameworkCore;
using Polly;
using System.Text.Json;

public class AppDbContext : DbContext
{
    private readonly AsyncCircuitBreakerPolicy _circuitBreaker;

    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options)
    {
        // 配置断路器：连续2次失败后断路5分钟
        _circuitBreaker = Policy
            .Handle<SqlException>()
            .Or<DbUpdateException>()
            .Or<InvalidOperationException>(ex => ex.Message.Contains("database"))
            .CircuitBreakerAsync(2, TimeSpan.FromMinutes(5));
    }

    public override async Task<int> SaveChangesAsync(CancellationToken cancellationToken = default)
    {
        try
        {
            return await _circuitBreaker.ExecuteAsync(() => base.SaveChangesAsync(cancellationToken));
        }
        catch (BrokenCircuitException)
        {
            // 序列化当前变更
            var bufferedChanges = ChangeTracker.Entries()
                .Where(e => e.State != EntityState.Unchanged)
                .Select(e => new BufferedEntityChange
                {
                    ChangeId = Guid.NewGuid(),
                    EntityTypeName = e.Entity.GetType().FullName,
                    EntityState = e.State,
                    PropertyValues = e.Properties.ToDictionary(p => p.Name, p => p.CurrentValue)
                })
                .ToList();

            var json = JsonSerializer.Serialize(bufferedChanges, new JsonSerializerOptions
            {
                WriteIndented = true,
                ReferenceHandler = ReferenceHandler.IgnoreCycles
            });

            // 原子写入文件
            var tempFileName = $"buffer_{Guid.NewGuid()}.pmq.tmp";
            var finalFileName = tempFileName.Replace(".tmp", "");
            await File.WriteAllTextAsync(tempFileName, json, cancellationToken);
            File.Move(tempFileName, finalFileName, overwrite: true);

            ChangeTracker.Clear();
            return 0;
        }
    }

    private class BufferedEntityChange
    {
        public Guid ChangeId { get; set; }
        public string EntityTypeName { get; set; } = string.Empty;
        public EntityState EntityState { get; set; }
        public Dictionary<string, object?> PropertyValues { get; set; } = new();
    }
}

2. 后台缓冲执行服务 #

创建HostedService定时扫描缓冲文件，尝试执行变更：

using Microsoft.EntityFrameworkCore;
using System.Text.Json;

public class BufferExecutionService : BackgroundService
{
    private readonly IServiceProvider _serviceProvider;
    private readonly string _bufferDir = Path.Combine(AppContext.BaseDirectory, "QueryBuffers");

    public BufferExecutionService(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;
        Directory.CreateDirectory(_bufferDir);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            if (await IsDatabaseAvailable())
            {
                var bufferFiles = Directory.GetFiles(_bufferDir, "*.pmq")
                    .OrderBy(f => File.GetCreationTime(f))
                    .ToList();

                foreach (var file in bufferFiles)
                {
                    try
                    {
                        var json = await File.ReadAllTextAsync(file, stoppingToken);
                        var changes = JsonSerializer.Deserialize<List<AppDbContext.BufferedEntityChange>>(json);
                        if (changes == null) continue;

                        using var scope = _serviceProvider.CreateScope();
                        var dbContext = scope.ServiceProvider.GetRequiredService<AppDbContext>();

                        foreach (var change in changes)
                        {
                            var entityType = Type.GetType(change.EntityTypeName);
                            if (entityType == null) continue;

                            var entity = Activator.CreateInstance(entityType);
                            if (entity == null) continue;

                            // 赋值缓冲属性
                            foreach (var prop in change.PropertyValues)
                            {
                                var propInfo = entityType.GetProperty(prop.Key);
                                if (propInfo != null)
                                {
                                    var value = prop.Value != null 
                                        ? Convert.ChangeType(prop.Value, propInfo.PropertyType) 
                                        : null;
                                    propInfo.SetValue(entity, value);
                                }
                            }

                            dbContext.Entry(entity).State = change.EntityState;
                        }

                        await dbContext.SaveChangesAsync(stoppingToken);
                        File.Delete(file);
                    }
                    catch
                    {
                        // 执行失败，移至错误目录
                        var errorDir = Path.Combine(_bufferDir, "Errors");
                        Directory.CreateDirectory(errorDir);
                        File.Move(file, Path.Combine(errorDir, Path.GetFileName(file)), overwrite: true);
                    }
                }
            }

            await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
        }
    }

    private async Task<bool> IsDatabaseAvailable()
    {
        try
        {
            using var scope = _serviceProvider.CreateScope();
            var dbContext = scope.ServiceProvider.GetRequiredService<AppDbContext>();
            return await dbContext.Database.CanConnectAsync();
        }
        catch
        {
            return false;
        }
    }
}

3. 服务注册 #

在Program.cs中注册DbContext和后台服务：

builder.Services.AddDbContext<AppDbContext>(options =>
    options.UseSqlServer(builder.Configuration.GetConnectionString("DefaultConnection")));

builder.Services.AddHostedService<BufferExecutionService>();

注意事项 #

• 序列化兼容性：在缓冲实体中加入版本号，应对实体类结构变更导致的反序列化失败问题。
• 性能优化：高并发场景下，改用SQLite或LiteDB替代纯文件存储，提升读写效率。
• 监控告警：添加缓冲文件数量监控，超过阈值时触发告警，及时排查数据库问题。

内容的提问来源于stack exchange，提问作者Pan Michal