Galosys.Foundation.Data 26.10.29.1

Galosys.Foundation.Data

成熟度: 🟢 稳定 — 生产可用，测试充分，活跃维护

简介

Galosys.Foundation.Data 提供 ADO.NET 多数据源支持，包含连接池管理、主从读写分离、健康检查等企业级特性。

特性

• 多数据源管理 - 基于 IDataSource 接口的统一数据源抽象
• 连接池 - DruidDataSource 内置连接池，支持泄漏检测
• 主从读写分离 - DataSourceContext 基于 AsyncLocal 的线程安全上下文切换
• 健康检查池 - HealthCheckedDataSourcePool Round-robin 轮询 + 健康探测 + 全挂降级主库
• 自动 DI 注册 - 通过配置自动注册数据源

安装

<PackageReference Include="Galosys.Foundation.Data" Version="x.x.x" />

配置

1. 连接字符串配置

在 appsettings.json 中配置主库和副本连接字符串：

{
  "ConnectionStrings": {
    "Default": "Server=master-host;Database=MyDb;User Id=sa;Password=xxx;Provider=Microsoft.Data.SqlClient",
    "Default.Replica1": "Server=replica1-host;Database=MyDb;User Id=sa;Password=xxx;Provider=Microsoft.Data.SqlClient",
    "Default.Replica2": "Server=replica2-host;Database=MyDb;User Id=sa;Password=xxx;Provider=Microsoft.Data.SqlClient"
  }
}

最小化配置仅需主库连接字符串（无 Replica 后缀），模块会自动识别副本。

2. 注册数据源

// 在 Program.cs 或 Startup.cs 中
services.AddDataSources(configuration);

有副本时自动创建 HealthCheckedDataSourcePool（Round-robin + 健康探测），无副本时直接使用主库 PooledDataSource。

使用示例

基本查询

public class QueryService
{
    private readonly IDataSource _dataSource;

    public QueryService([FromKeyedServices("Default")] IDataSource dataSource)
    {
        _dataSource = dataSource;
    }

    public async Task<List<User>> GetUsersAsync()
    {
        using var conn = _dataSource.GetConnection();
        // 执行查询...
    }
}

读写分离

通过 DataSourceContext.SwitchTo 切换到副本上下文：

using (DataSourceContext.SwitchTo(DataSourceType.Replica))
{
    // 此范围内的查询走副本
    var users = await GetUsersAsync();
}
// 离开 using 后自动恢复之前的上下文

嵌套切换

using (DataSourceContext.SwitchTo(DataSourceType.Master))
{
    // 写操作走主库
    await SaveOrderAsync(order);

    using (DataSourceContext.SwitchTo(DataSourceType.Replica))
    {
        // 读操作走副本
        var stats = await GetStatsAsync();
    }

    // 自动恢复为 Master
}

DataSourceType 枚举

模块化异常处理

Data 模块提供两个专属异常处理器，均位于 System 命名空间（寄生），绑定具体数据库驱动类型并以 Priority = 200 优先于框架默认 DbExceptionHandler(0) 接管：

• SqlExceptionHandler —— 绑定 Microsoft.Data.SqlClient.SqlException
• NpgsqlExceptionHandler —— 绑定 Npgsql.PostgresException

[ExceptionHandler(200)]
public sealed class SqlExceptionHandler : TypedExceptionHandler<Microsoft.Data.SqlClient.SqlException>
{
    public override ValueTask<ExceptionHandlingOutput> HandleAsync(
        Microsoft.Data.SqlClient.SqlException exception,
        CancellationToken cancellationToken = default)
    {
        var (code, message) = Map(exception.Number);
        return ValueTask.FromResult(new ExceptionHandlingOutput(400, code, message));
    }

    private static (string code, string message) Map(int number) => number switch
    {
        2627 or 2601 => ("A00400", "提示：数据库主键冲突，请联系管理员"),
        547          => ("A00400", "提示：违反外键约束，请联系管理员"),
        _ => (ErrorCode.Of("A", (short)BizErrorCode.ParameterInvalid),
              "提示：数据库操作异常，请联系管理员"),
    };
}

[ExceptionHandler(200)]
public sealed class NpgsqlExceptionHandler : TypedExceptionHandler<Npgsql.PostgresException>
{
    public override ValueTask<ExceptionHandlingOutput> HandleAsync(
        Npgsql.PostgresException exception,
        CancellationToken cancellationToken = default)
    {
        var (code, message) = Map(exception.SqlState);
        return ValueTask.FromResult(new ExceptionHandlingOutput(400, code, message));
    }

    private static (string code, string message) Map(string sqlState) => sqlState switch
    {
        "23505" => ("A00400", "提示：数据库主键冲突，请联系管理员"),
        "23503" => ("A00400", "提示：违反外键约束，请联系管理员"),
        _ => (ErrorCode.Of("A", (short)BizErrorCode.ParameterInvalid),
              "提示：数据库操作异常，请联系管理员"),
    };
}

两个处理器均继承 System.TypedExceptionHandler<T> 并标注 [ExceptionHandler(200)]，经 AddAnnotationConfiguration 自动注册，优先于框架默认 DbExceptionHandler(0) 接管 SqlServer / PostgreSql 抛出的具体异常。

核心类

HealthCheckedDataSourcePool

行为：

• 轮询选择: Round-robin 依次选择健康副本
• 跳过不健康副本: 健康检查失败自动跳过
• 全挂降级主库: 所有副本不健康时自动回退主库
• 自动恢复: 副本恢复健康后自动重新纳入轮询

DruidDataSource 连接池修复

修复了 GetConnection 中 lock + semaphore 嵌套导致的死锁问题：信号量等待移到 lock 外部，确保等待线程不阻塞连接归还。

配置项（通过 ConnectionStrings 节点下的子键指定）：

CDC 变更数据捕获

Galosys.Foundation.Data 内置 CDC Pipeline，支持 SQL Server 和 PostgreSQL 的变更数据捕获。

前提条件

SQL Server

-- 数据库级别（手动执行一次）
EXEC sys.sp_cdc_enable_db;

-- 确认已开启
SELECT name, is_cdc_enabled FROM sys.databases WHERE name = DB_NAME();

sp_cdc_enable_db 只需要执行一次。之后 MssqlConnector 启动时会自动对指定表执行 sp_cdc_enable_table。

PostgreSQL

-- postgresql.conf 配置（修改后重启）
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

-- 确认当前配置
SHOW wal_level;

如果 wal_level 不是 logical，PostgresConnector 启动不会报错，但逻辑复制不会生效。

架构

ICdcConnector ──▶ CdcEngine (BackgroundService)
     │                     │
     │           OffsetManager ──▶ IOffsetStore (断点续传)
     │                     │
     └──── CdcPipeline (Transforms[]) ──▶ ICdcSink

注册

// Program.cs
services.AddCdc(configuration);
services.AddCdcSource<Order>("orders", sp => new MssqlConnector<Order>(
    "orders",
    sp.GetRequiredKeyedService<IDataSource>("Default"),
    new CdcSourceOptions { Table = "Orders", ConnectionString = "..." }));
services.AddCdcSinkMessageBus(); // CDC → IMessageBus → Outbox

// 或事件回调
var engine = sp.GetRequiredService<CdcEngine>();
engine.OnChange += (_, events) => { /* 处理变更事件 */ };

SQL Server 支持

基于 SQL Server CDC 功能 (cdc.fn_cdc_get_all_changes_* 轮询)，支持多表自动发现：

services.AddCdcSource<Order>("orders", sp => new MssqlConnector<Order>(
    "orders", sp.GetRequiredKeyedService<IDataSource>("Default"),
    new CdcSourceOptions
    {
        ConnectionString = "...",
        Table = "Orders",
        PollInterval = TimeSpan.FromSeconds(5),
        SnapshotMode = CdcSnapshotMode.Initial // Initial / SchemaOnly / Never
    }));

PostgreSQL 支持

基于 PostgreSQL 逻辑复制流式协议 (pgoutput)：

services.AddCdcSource<Order>("orders", sp => new PostgresConnector<Order>(
    "orders", sp.GetRequiredKeyedService<IDataSource>("Default"),
    new CdcSourceOptions
    {
        ConnectionString = "...",
        Table = "Orders",
        SnapshotMode = CdcSnapshotMode.Initial
    }));

自动创建 publication、replication slot，使用 PgOutputReplicationStream 流式解码变更。与旧版 SQL 轮询方案相比，大幅降低 WAL 延迟和 CPU 开销。

多副本 Leader 选举

services.AddCdc(configuration);
services.AddCdcLeaderElection(); // 启用分布式锁选举

只有获得 IDistributedLock 的实例执行 CDC 轮询。

配置

{
  "Cdc": {
    "Coordinator": {
      "PollIntervalMs": 5000,
      "BatchSize": 100,
      "EnableLeaderElection": true,
      "LockKey": "cdc:coordinator:leader",
      "LockExpiry": "00:00:30"
    }
  }
}

核心类

依赖

• Galosys.Foundation.Core