在C# .NET Core控制台应用中按消息动态切换多数据库

控制台应用没有Web API中的HttpContext依赖，所以需要调整动态切换数据库的逻辑，核心思路是根据每条Event Hub消息中的客户ID，动态生成对应数据库的连接字符串并创建/复用DbContext实例。

1. 调整CustomerContext的设计 #

修改CustomerContext，添加支持传入自定义连接字符串的构造函数，并在OnConfiguring中使用该连接串：

public class CustomerContext : DbContext
{
    private readonly string? _connectionString;

    // 保留无参构造函数，用于EF迁移、工具等场景
    public CustomerContext()
    {
    }

    // 接受连接字符串的构造函数
    public CustomerContext(string connectionString)
    {
        _connectionString = connectionString;
    }

    protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
    {
        if (!optionsBuilder.IsConfigured && !string.IsNullOrEmpty(_connectionString))
        {
            optionsBuilder.UseNpgsql(_connectionString);
        }
    }

    // 定义你的实体集合
    // public DbSet<YourBusinessEntity> BusinessEntities { get; set; }
}

2. 封装连接字符串生成逻辑 #

提取连接字符串的生成逻辑，保持和Web API一致的客户ID处理规则：

public static class ConnectionStringGenerator
{
    public static string BuildCustomerDbConnection(string customerId)
    {
        var sanitizedDbName = customerId?.Replace('-', '_');
        if (string.IsNullOrWhiteSpace(sanitizedDbName))
        {
            throw new ArgumentException("客户ID无效，无法生成数据库连接");
        }

        var builder = new NpgsqlConnectionStringBuilder
        {
            Host = "Hostname",
            Port = "port",
            SslMode = SslMode.Require,
            Username = "username",
            Database = sanitizedDbName,
            Password = "password",
            Timeout = 1024,
            CommandTimeout = 60000
        };

        return builder.ConnectionString;
    }
}

3. Event Hub消息处理中动态创建DbContext #

在Event Hub的消息处理逻辑里，解析消息获取客户ID，生成连接串后创建CustomerContext实例进行数据库操作：

// 假设已完成Event Hub订阅的基础配置，以下是单条消息的处理逻辑
async Task ProcessIncomingMessage(EventData eventData)
{
    // 解析消息内容（根据实际消息格式调整序列化方式）
    var messageBody = Encoding.UTF8.GetString(eventData.Body.ToArray());
    var customerMessage = JsonSerializer.Deserialize<CustomerProcessingMessage>(messageBody);

    if (customerMessage == null || string.IsNullOrWhiteSpace(customerMessage.CustomerId))
    {
        Console.WriteLine("收到无效消息，跳过处理");
        return;
    }

    try
    {
        // 生成对应客户数据库的连接字符串
        var connectionString = ConnectionStringGenerator.BuildCustomerDbConnection(customerMessage.CustomerId);
        
        // 创建DbContext实例并执行业务操作
        using var dbContext = new CustomerContext(connectionString);
        
        // 示例：根据消息内容执行数据库操作
        // var targetEntity = await dbContext.BusinessEntities.FirstOrDefaultAsync(e => e.Id == customerMessage.EntityId);
        // if (targetEntity != null)
        // {
        //     targetEntity.Status = customerMessage.TargetStatus;
        //     await dbContext.SaveChangesAsync();
        // }
    }
    catch (Exception ex)
    {
        Console.WriteLine($"处理客户 {customerMessage.CustomerId} 的消息失败: {ex.Message}");
    }
}

// 自定义消息模型，根据实际需求调整字段
public class CustomerProcessingMessage
{
    public string CustomerId { get; set; } = string.Empty;
    // 添加其他处理所需的字段，比如EntityId、TargetStatus等
}

4. 高并发场景优化：使用DbContext池 #

如果消息处理量较大，频繁创建DbContext会带来性能开销，可以通过EF Core的DbContext池来复用实例，配置方式如下：

// 在Program.cs的服务配置中
var services = new ServiceCollection();
services.AddDbContextPool<CustomerContext>(options =>
{
    // 可配置默认连接（可选，实际使用时会动态替换）
    options.UseNpgsql();
});

var serviceProvider = services.BuildServiceProvider();

之后在消息处理中从池中获取实例并替换连接字符串：

async Task ProcessIncomingMessage(EventData eventData, IServiceProvider serviceProvider)
{
    // 解析消息...
    
    var connectionString = ConnectionStringGenerator.BuildCustomerDbConnection(customerMessage.CustomerId);
    
    // 创建作用域并从池中获取DbContext
    using var scope = serviceProvider.CreateScope();
    var dbContext = scope.ServiceProvider.GetRequiredService<CustomerContext>();
    // 动态替换连接字符串
    dbContext.Database.SetConnectionString(connectionString);
    
    // 执行数据库操作...
}

注意：使用DbContext池时，要确保每次操作前都替换正确的连接字符串，避免实例复用导致的连接串混乱。

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