EF Core同构多数据库通用查询扩展方法实现咨询

针对你遇到的多动态数据库查询需求，这里提供一套基于DbContext扩展方法的通用实现，既能保留并行执行的效率，又能避免重复编写单表查询代码。

核心扩展方法实现 #

public static class DbContextMultiDbExtensions
{
    // 通用多数据库查询：支持传递数据库名到查询逻辑
    public static List<TResult> QueryAcrossDatabases<TDbContext, TResult>(
        this IEnumerable<DbConnectionInfo> connectionInfos,
        Func<TDbContext, string, IQueryable<TResult>> queryFactory,
        bool runInParallel = true,
        int maxDegreeOfParallelism = 16)
        where TDbContext : DbContext
    {
        var results = new ConcurrentBag<TResult>();

        void ProcessDatabase(DbConnectionInfo connInfo)
        {
            using var dbContext = DbHelper.Get_DataDBContext(connInfo.DbName, connInfo.DbInstance);
            var queryResults = queryFactory(dbContext, connInfo.DbName).ToList();
            foreach (var item in queryResults)
            {
                results.Add(item);
            }
        }

        if (runInParallel)
        {
            Parallel.ForEach(connectionInfos, new ParallelOptions { MaxDegreeOfParallelism = maxDegreeOfParallelism }, ProcessDatabase);
        }
        else
        {
            foreach (var connInfo in connectionInfos)
            {
                ProcessDatabase(connInfo);
            }
        }

        return results.ToList();
    }

    // 重载：无需传递数据库名时简化调用
    public static List<TResult> QueryAcrossDatabases<TDbContext, TResult>(
        this IEnumerable<DbConnectionInfo> connectionInfos,
        Func<TDbContext, IQueryable<TResult>> queryFactory,
        bool runInParallel = true,
        int maxDegreeOfParallelism = 16)
        where TDbContext : DbContext
    {
        return connectionInfos.QueryAcrossDatabases((ctx, _) => queryFactory(ctx), runInParallel, maxDegreeOfParallelism);
    }

    // 异步版本：适合IO密集型场景
    public static async Task<List<TResult>> QueryAcrossDatabasesAsync<TDbContext, TResult>(
        this IEnumerable<DbConnectionInfo> connectionInfos,
        Func<TDbContext, string, Task<List<TResult>>> queryFactory,
        bool runInParallel = true,
        int maxDegreeOfParallelism = 16)
        where TDbContext : DbContext
    {
        var tasks = new List<Task<List<TResult>>>();

        foreach (var connInfo in connectionInfos)
        {
            tasks.Add(Task.Run(async () =>
            {
                using var dbContext = DbHelper.Get_DataDBContext(connInfo.DbName, connInfo.DbInstance);
                return await queryFactory(dbContext, connInfo.DbName);
            }));
        }

        if (runInParallel)
        {
            var allResults = await Task.WhenAll(tasks);
            return allResults.SelectMany(x => x).ToList();
        }
        else
        {
            var results = new List<TResult>();
            foreach (var task in tasks)
            {
                results.AddRange(await task);
            }
            return results;
        }
    }
}

调用示例 #

同步并行查询（对应你原来的代码） #

var searchResults = connStrings.QueryAcrossDatabases<DB_DataContext, FindConstructionSearchResult>(
    (dctx, dbName) => dctx.SStavby
        .Where(x => x.Nazev.Contains(request.ConstructionName) || x.Znacka.Contains(request.ConstructionName))
        .Select(x => new FindConstructionSearchResult
        {
            ConstructionMark = x.Znacka,
            ConstructionName = x.Nazev,
            Database = dbName
        }),
    runInParallel: true,
    maxDegreeOfParallelism: 16);

异步串行查询 #

var searchResults = await connStrings.QueryAcrossDatabasesAsync<DB_DataContext, FindConstructionSearchResult>(
    async (dctx, dbName) => await dctx.SStavby
        .Where(x => x.Nazev.Contains(request.ConstructionName) || x.Znacka.Contains(request.ConstructionName))
        .Select(x => new FindConstructionSearchResult
        {
            ConstructionMark = x.Znacka,
            ConstructionName = x.Nazev,
            Database = dbName
        })
        .ToListAsync(),
    runInParallel: false);

关键特性说明 #

• 泛型强类型：通过TDbContext泛型参数绑定你的上下文类型，避免类型转换错误
• 查询逻辑复用：只需要传入单库的查询委托，自动处理多库遍历逻辑
• 灵活执行模式：通过runInParallel一键切换串行/并行，maxDegreeOfParallelism控制并发数
• 资源自动释放：使用using语句管理DbContext，防止数据库连接泄漏
• 异步支持：提供异步版本，适合高并发IO场景，性能更优

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