Galosys.Foundation.EntityFrameworkCore 26.8.21.1

Galosys.Foundation.EntityFrameworkCore

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

简介

Galosys.Foundation.EntityFrameworkCore 基于 Entity Framework Core 提供数据访问层集成，包含自动审计、多租户、软删除、仓储模式等企业级特性。

特性

• 自动审计 - 自动填充创建者、修改者、时间戳
• 多租户支持 - 租户数据隔离和全局过滤器
• 软删除 - 标记删除而非物理删除
• 全局查询过滤器 - 自动应用删除、租户、应用过滤条件
• 雪花 ID 自动生成 - 基于 Snowflake 算法的 ID 生成
• snake_case 列名映射 - 自动将 PascalCase 属性映射到 snake_case 列名
• 分表支持 - 按年/月/日动态分表
• 仓储模式 - 通用仓储接口和实现
• 工作单元模式 - 事务管理
• 动态查询 - 基于表达式的动态查询构建
• 领域事件 - 保存时自动发布领域事件
• 乐观并发控制 - 可选的 RowVersion 并发令牌，支持冲突自动重试
• Outbox 批量处理 - EfCoreOutboxStore 实现 TryAcquireBatchAsync（PostgreSQL SKIP LOCKED）、MarkManySentAsync、MarkManyFailedAsync 批量 API
• 连接弹性 - 四种数据库自动重试瞬态故障
• 读写分离 - DbCommandInterceptor 自动路由查询到副本
• 支持多种数据库 - SQL Server、PostgreSQL、MySQL、SQLite、Oracle

安装

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

配置

1. 连接字符串配置

在 appsettings.json 中配置连接字符串，需包含 provider 信息：

{
  "ConnectionStrings": {
    "Default": "Server=localhost;Database=MyDb;User Id=sa;Password=xxx;Provider=Microsoft.Data.SqlClient",
    "Postgres": "Host=localhost;Database=mydb;Username=postgres;Password=xxx;Provider=Npgsql",
    "MySql": "Server=localhost;Database=mydb;User=root;Password=xxx;Provider=MySqlConnector",
    "Sqlite": "Data Source=mydb.db;Provider=Microsoft.Data.Sqlite"
  }
}

2. 注册 DbContext

使用 [DbContext] 特性标记 DbContext 类，模块会自动发现并注册：

[DbContext("Default")]
public class AppDbContext : DbContext<AppDbContext>
{
    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }

    public DbSet<User> Users => Set<User>();
    public DbSet<Order> Orders => Set<Order>();
}

3. 模块自动注册

模块会自动扫描带有 [DbContext] 特性的类并注册到 DI 容器：

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

前置约定：DbContext<T> 依赖 Galosys.Core 的 AddCore() 注册的上下文服务（ITenantContext/IAppContext/IUserContext/ITimeProvider/ApplicationMessagePublisher 等），宿主须先调用 AddCore()。EF Core 集成包（Actuator/Outbox/Authorization 等）采用"宿主泛型上下文 + 显式装配"约定，详见 docs/adr/0001-efcore-integration-conventions.md。

使用示例

实体定义

// 完整实体（含审计、软删除）
[Table("users")]
public class User : FullEntity<long>
{
    [SnowflakeId]
    public override long Id { get; protected set; }
    public string Name { get; set; }
    [CreatorId]
    public override long CreatorId { get; protected set; }
    [CreatedAt]
    public override DateTime CreatedAt { get; protected set; }
}

// 多租户实体
[Table("orders")]
public class Order : MtMaEntity
{
    [SnowflakeId]
    public override long Id { get; protected set; }
    public string OrderNo { get; set; }
    public decimal Amount { get; set; }
}

DbContext 使用

public class UserService
{
    private readonly AppDbContext _context;

    public UserService(AppDbContext context)
    {
        _context = context;
    }

    public IQueryable<User> GetActiveUsers()
    {
        return _context.Query<User>(); // 自动过滤已删除记录
    }

    public async Task AddUserAsync(User user)
    {
        await _context.Entity<User>().AddAsync(user);
        await _context.SaveChangesAsync();
    }

    public async Task SaveWithEventsAsync()
    {
        await _context.SaveEntitiesAsync(); // 保存并发布领域事件
    }
}

仓储模式

定义业务接口，继承 IRepository<T, TID>，再用 [Repository] 标记实现类：

// 1. 定义业务仓储接口
public interface IUserRepository : IRepository<User, long>
{
    Task<User?> GetByNameAsync(string name);
}

// 2. 实现类继承 EfCoreRepository，用 [Repository] 标记自动注册
[Repository]
public class UserRepository : EfCoreRepository<AppDbContext, User, long>, IUserRepository
{
    public UserRepository(AppDbContext ctx) : base(ctx) { }

    public async Task<User?> GetByNameAsync(string name)
    {
        return await Query(u => u.Name == name).FirstOrDefaultAsync();
    }
}

// 3. 使用时注入业务接口
public class UserService
{
    private readonly IUserRepository _repository;

    public UserService(IUserRepository repository)
    {
        _repository = repository;
    }

    public async Task<User?> GetByIdAsync(long id) => await _repository.FindOneAsync(id);

    public async Task AddAsync(params User[] users)
    {
        await _repository.AddAsync(users);
        await _repository.SaveAsync();
    }
}

工作单元

行为变更: CommitAsync 现在会自动调用 SaveChangesAsync，无需手动保存。内部通过 _changesSaved 标志位防止重复保存，兼容已有的手动 SaveChangesAsync + CommitAsync 双调用模式。

简化写法（推荐）：

public class OrderService
{
    private readonly IUnitOfWork _unitOfWork;
    private readonly AppDbContext _context;

    public OrderService(IUnitOfWork unitOfWork, AppDbContext context)
    {
        _unitOfWork = unitOfWork;
        _context = context;
    }

    public async Task CreateOrderAsync(Order order)
    {
        await _unitOfWork.BeginTransactionAsync();
        try
        {
            _context.Entity<Order>().Add(order);
            // CommitAsync 自动调用 SaveChangesAsync，无需手动保存
            await _unitOfWork.CommitAsync();
        }
        catch
        {
            await _unitOfWork.RollbackAsync();
            throw;
        }
    }
}

兼容写法（手动 SaveChanges + CommitAsync）：

public async Task CreateOrderAsync(Order order)
{
    await _unitOfWork.BeginTransactionAsync();
    try
    {
        _context.Entity<Order>().Add(order);
        await _context.SaveChangesAsync();  // 手动保存
        await _unitOfWork.CommitAsync();     // 不会重复保存（_changesSaved 标志位）
    }
    catch
    {
        await _unitOfWork.RollbackAsync();
        throw;
    }
}

ITimeProvider 注入

DbContext<T> 通过构造函数注入 ITimeProvider（Core 模块已注册 Singleton），审计字段自动使用 ITimeProvider 获取时间，确保时区一致且可测试。无需额外配置。

// 单元测试中替换时间源
services.AddSingleton<ITimeProvider>(new TestTimeProvider(fixedTime));

乐观并发控制

继承 ConcurrencyEntity<TID> 即可启用 RowVersion 并发令牌：

[Table("products")]
public class Product : ConcurrencyEntity<long>
{
    public string Name { get; set; }
    public decimal Price { get; set; }
}

冲突时自动重试更新：

public async Task UpdatePriceAsync(Product product)
{
    await _repository.UpdateWithRetryAsync(product, maxRetries: 3);
}

执行策略配置

通过 [DbContext] 特性配置重试参数（默认 3 次 / 5 秒）：

[DbContext("Default", MaxRetryCount = 5, MaxRetryDelaySeconds = 10)]
public class AppDbContext : DbContext<AppDbContext> { }

读写分离

1. 配置主库和副本

在 appsettings.json 中添加副本连接字符串（{名称}.Replica{N} 格式）：

{
  "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"
  }
}

最小化配置仅需主库连接字符串，无副本时不启用读写分离。

2. 注册数据源

services.AddDataSources(configuration); // 自动识别副本，创建 HealthCheckedDataSourcePool

3. 使用 DataSourceContext 切换

// 定义业务仓储接口
public interface IOrderQueryRepository : IRepository<Order, long>
{
    Task<List<Order>> GetRecentOrdersAsync(int count);
}

[Repository]
public class OrderQueryRepository : EfCoreRepository<AppDbContext, Order, long>, IOrderQueryRepository
{
    public OrderQueryRepository(AppDbContext ctx) : base(ctx) { }

    public async Task<List<Order>> GetRecentOrdersAsync(int count)
    {
        using (DataSourceContext.SwitchTo(DataSourceType.Replica)) // SELECT 走副本
        {
            return await Query()
                .OrderByDescending(o => o.CreatedAt)
                .Take(count)
                .ToListAsync();
        }
    }
}

// 使用时注入业务接口
public class OrderAppService
{
    private readonly IOrderQueryRepository _queryRepo;
    private readonly IOrderRepository _orderRepo;

    public OrderAppService(IOrderQueryRepository queryRepo, IOrderRepository orderRepo)
    {
        _queryRepo = queryRepo;
        _orderRepo = orderRepo;
    }

    public async Task<Order> CreateOrderAsync(Order order)
    {
        await _orderRepo.AddAsync(order); // DML 自动走主库，无需手动切换
        await _orderRepo.SaveAsync();
        return order;
    }

    public Task<List<Order>> GetRecentOrdersAsync(int count)
    {
        return _queryRepo.GetRecentOrdersAsync(count); // 内部切换到副本
    }
}

4. 嵌套切换

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

    using (DataSourceContext.SwitchTo(DataSourceType.Replica))
    {
        var stats = await QueryStatsAsync(); // 读操作走副本
    }
    // 自动恢复为 Master
}

分表

通过 [Table] + [Sharding] 特性标记实体，框架自动拼接表名后缀：

[Table("orders")]
[Sharding(ShardingType.Month)] // 按月分表 → orders_202604
public class Order : FullEntity<long>
{
    public string OrderNo { get; set; }
    public decimal Amount { get; set; }
}

[Table("logs")]
[Sharding(ShardingType.Day)] // 按日分表 → logs_20260415
public class Log : FullEntity<long>
{
    public string Message { get; set; }
}

[Table("reports")]
[Sharding(ShardingType.Year)] // 按年分表 → reports_2026（默认）
public class AnnualReport : FullEntity<long>
{
    public string Title { get; set; }
}

表名后缀基于 DateTimeOffset.Now 自动生成，DbContext OnModelCreating 时自动路由到对应表。需配合 [DbContext(Dynamic = true)] 启用动态模型缓存。

读写分离 + 分表组合

[Table("order_items")]
[Sharding(ShardingType.Month)]
public class OrderItem : FullEntity<long>
{
    public long OrderId { get; set; }
    public string ProductName { get; set; }
    public int Quantity { get; set; }
}

[Repository]
public class OrderItemRepository : EfCoreRepository<AppDbContext, OrderItem, long>, IOrderItemRepository
{
    public OrderItemRepository(AppDbContext ctx) : base(ctx) { }

    public async Task<List<OrderItem>> GetItemsByOrderAsync(long orderId)
    {
        using (DataSourceContext.SwitchTo(DataSourceType.Replica)) // 副本读取 + 按月分表
        {
            return await Query(i => i.OrderId == orderId).ToListAsync();
        }
    }
}

EntityTypeConfiguration 接口驱动配置

通过继承 EntityTypeConfigurationBase，根据实体实现的接口自动获得对应的索引和字段约束配置：

// 用户配置 — 实体实现 IMultiTenancy，自动获得 TenantId 索引
public class SysUserConfiguration : EntityTypeConfigurationBase<SysUser, long>
{
    // 基类自动配置：CreatedAt/CreatorId 索引 + LastModifierId 索引 + TenantId 索引
    // 可覆写 ConfigureCore 添加自定义配置
}

// 菜单配置 — 实体实现 IMultiApplication，自动获得 AppId 索引
public class SysMenuConfiguration : EntityTypeConfigurationBase<SysMenu, long> { }

// 租户应用配置 — 实体同时实现两个接口，自动获得 (TenantId, AppId) 复合索引
public class SysTenantAppConfiguration : EntityTypeConfigurationBase<SysTenantAppConfig, long> { }

// 全局数据 — 仅 FullEntity 审计字段索引
public class SysRegionConfiguration : EntityTypeConfigurationBase<SysRegion, long> { }

说明：全局查询过滤器（软删除、租户、应用）由 DefaultGlobalFilterProvider 基于 IDeletable/IMultiTenancy/IMultiApplication 接口统一应用；蛇形命名映射由 EntityTypeConfigurationBase 处理；IConcurrency 自动检测并配置并发令牌。索引配置完全由接口检测驱动，无需为不同实体类型选择不同基类。

核心类

手动建表

分表场景下，应用启动时需手动创建当前周期的物理表：

// 在 Program.cs 或 Startup.cs 中调用一次
using (var scope = app.Services.CreateScope())
{
    var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
    await db.EnsureShardingTablesAsync(); // CREATE TABLE IF NOT EXISTS 语义
}

EnsureShardingTablesAsync 扫描所有 [Sharding] 实体，为当前周期创建物理表。表已存在时不报错。

模型缓存策略

DynamicModelCacheKeyFactory 根据 [Sharding] 粒度优化模型缓存刷新频率：

混合粒度取最细（Day > Month > Year）。未标记 [Sharding] 的 DbContext 不启用动态缓存。

审计属性

行为变更 (v3): 审计字段填充已从 SavingChanges 事件移至 SaveChanges/SaveChangesAsync 重写中，确保 ShardingCore 路由评估完成后再填充。对调用方无感知。

审计字段同时支持 DateTime 和 DateTimeOffset 类型，通过 ITimeProvider 获取时间。

全局过滤器

模块通过 IGlobalFilterProvider 接口自动为以下接口实现全局查询过滤器：

• IDeletable - 过滤已删除记录 (e.Deleted == false)
• IMultiTenancy - 租户数据隔离（ITenantContext.TenantId == 0 || e.TenantId == ITenantContext.TenantId）
• IMultiApplication - 应用数据隔离（IAppContext.AppId == 0 || e.AppId == IAppContext.AppId）

过滤器动态求值机制

过滤器表达式捕获 ITenantContext/IAppContext 对象引用（而非标量值），EF Core 每次 query 时动态求值：

• 无 header（TenantId == 0）：表达式短路为 true，查全部数据
• 有 header（TenantId > 0）：按 TenantId 过滤，仅返回对应租户数据
• AppId 同理

无需按租户缓存不同模型（IModelCacheKeyFactory），单个模型 + 动态表达式即可服务所有请求。

Outbox 存储

services.AddEfCoreOutboxStore<MyDbContext>();

自动发现 OutboxMessageEntityTypeConfiguration（表 base.base_outbox_msg，列名 snake_case），一行注册即可，无需修改 DbContext 的 OnModelCreating。

OutboxSaveChangesInterceptor

AddEfCoreOutboxStore 自动注册 OutboxSaveChangesInterceptor 到 DI（Singleton）。该拦截器在 SaveChanges 时从 PendingMsgCol（AsyncLocal）取出待发送消息，同事务写入 base_outbox_msg 表。

业务代码无需额外配置：

await _db.Orders.AddAsync(order);
await _bus.PublishAsync("order.created", order);  // → PendingMsgCol，不入库
await _db.SaveChangesAsync();  // 拦截器同事务写入 orders + outbox_msg

EfCoreOutboxStore

PollAsync — LINQ AsNoTracking，跨数据库兼容（PostgreSQL 使用 FOR UPDATE SKIP LOCKED 原生 SQL）。标记操作 — 原生 SQL UPDATE，绕过 ChangeTracker。

测试覆盖

dotnet test framework/test/Galosys.Foundation.EntityFrameworkCore.Tests/ --filter "FullyQualifiedName~EfCoreOutboxStoreTests"
dotnet test framework/test/Galosys.Foundation.EntityFrameworkCore.Tests/ --filter "FullyQualifiedName~OutboxSaveChangesInterceptor"

依赖

• Microsoft.EntityFrameworkCore
• Microsoft.EntityFrameworkCore.Relational
• Microsoft.EntityFrameworkCore.SqlServer
• Microsoft.EntityFrameworkCore.Sqlite
• Npgsql.EntityFrameworkCore.PostgreSQL
• Pomelo.EntityFrameworkCore.MySql
• Oracle.EntityFrameworkCore
• Galosys.Foundation.Core
• Galosys.Foundation.Data

CDC EF Core 位置存储

本模块提供 EfOffsetStore<TDbContext>，基于 EF Core 持久化 CDC 位置到数据库（线程安全，使用 IDbContextFactory<TDbContext> 每操作新建上下文）。

// 注册 EF Core 位置存储替代默认的 InMemoryOffsetStore
services.RemoveAll<IOffsetStore>();
services.AddSingleton<IOffsetStore, EfOffsetStore<YourDbContext>>();

自动保存 CDC 最后读取的 LSN 位置，支持断点续传和多连接器 namespace 隔离。

旧版 EfPositionStore<TDbContext> 已标记 [Obsolete]，请迁移到 EfOffsetStore<TDbContext>。