使用 OpenJson (SQL Server) 提升 EF Core 6\7 "Contains" 性能 - NuGet 包

下载 Visual Studio 22 项目文件 4.1MB

下载 SQL Server 脚本 - 创建数据库、表以及100万条记录的脚本

引言

本文档概述了如何配置 Entity Framework Core (6 或 7) 设置，以利用 OpenJson 在使用大量 SQL IN 子句值时提供的性能增益。但无需将项目迁移到 EF Core 8。

目的

您的项目可能处于这样的境地：您现在需要扩展并使用缓存值查询更多数据。但如果无法利用 EF Core 8，因为迁移将为您的团队带来更多的测试和开发工作——您仍然可以使用 EF Core 6 或 7 配合此 NuGet 包或直接将类合并到您的代码库中，从而使用 OpenJson。

范围

本文档的范围是传达开始将 OpenJson 与 EF Core 6\7 一起使用所需的配置步骤。

必备组件

• 了解 (Microsoft 的) TSQL
• 了解 (Microsoft 的) Entity Framework Core
• 了解 (Microsoft 的) Visual Studio 或 Visual Code
• 了解 EF Core 的 Pipeline & Interceptors
• 了解如何 安装 NuGet 包
• 已安装 SQL Server (开发人员版)。
• 已安装 SSMS
• 已安装 Visual Studio 22 Community。

设置测试数据库

运行上述 SQL Server 脚本以

1. 创建一个名为 LinqPerformanceDemo 的新测试数据库
2. 创建一个名为 LargeTable 的新表
3. 向 LargeTable 表中插入 100 万条记录

完成后，您应该会得到以下结果

基准测试（秒）

使用带有大量 IN 子句和少量 NOT IN 子句的简单 LINQ 语句

context.LargeTables.Where(itm => chunkData.Contains(itm.CodeName) && !_listNotIn.Contains(itm.ProjectRef)).ToListAsync();

IN 子句 - 批次大小	标准 EF 设置	OpenJson EF 拦截器
2万	151	204
3万	396	157
4万	Timeout	132
10万	Timeout	123
15万	Timeout	99
20万	Timeout	98
30万	Timeout	91
50万	Timeout	83
70万	Timeout	86
1M	Timeout	105

在 4万时，标准 EF 在第一个 IN 子句中因超时而失败

注意：Entity Framework SQL 查询生成器经常会创建复杂且曲折的 SQL 语句，尤其是在涉及 IN\NOT IN 子句时。它会在需要匹配值的任何连接表中复制该子句。这通常会导致查询失败，因为 SQL Server 由于其复杂性而无法执行它。但使用 OpenJson 可以解决这个问题——通过在分块方法中使用更多值，并使用 concurrentbag 进行多线程处理以维护结果（示例代码在下方）。

使用 NuGet 包添加拦截器或直接将类添加到您的项目

使用 NuGet 包添加拦截器类

NuGet 包位于 Nuget.Org 上，其中包含如何编辑项目中 EF 设置的 Readme.md 文件——选择最新版本进行安装。

您可以在 Visual Studio 中像普通包一样将 NuGet 包添加到您的项目中——搜索 Linq.OpenJson.InClause.Middleware 并点击安装。

 

 

直接添加拦截器类

如果您打开附加的 Visual Studio 解决方案，您会看到三个项目，名为 DemoLinqContainsOpenJson 的项目用于演示和测试新的 EF 拦截器。

第二个项目名为 LinqOpenJsonInClauseMiddleware，是 NuGet 包的内容，包含拦截器重写方法 (ReaderExecutingAsync 和 ReaderExecuting)。

如果您想自定义代码，例如，增加何时使用 OpenJson 而不是普通 IN\NOT IN 子句的阈值，您可以在 SqlServerReaderExecutingOpenJson 类中通过设置常量 MINIMUM_ITEM_SIZE 来完成。在这段代码中，当拦截器发现一个 IN\NOT IN 子句包含超过 1000 个值时，它会将 SQL 转换为使用 OpenJson 语法。

要将 SqlServerReaderExecutingOpenJson 类集成到您的项目中，只需从 LinqOpenJsonInClauseMiddleware 项目中复制该类，然后将类中的命名空间重命名以匹配您的项目命名空间。然后，您可以编辑生成的 SQL 以满足您的需求。

编辑 Entity Framework Core 设置以调用自定义拦截器中间件

您可以通过两种方式配置 SqlServerReaderExecutingOpenJson 类以包含在您的项目中——选择最适合您项目的方式。

Program.cs 类

1. 通过将命名空间 using LinqOpenJsonInClauseMiddleware.Classes; 添加到您的项目中
2. 将代码行 options.AddInterceptors(new SqlServerReaderExecutingOpenJson 添加到每个 dbContext 相关选项（下方标准和工厂上下文）。

您的 dbContext 类

1. 通过将命名空间 using LinqOpenJsonInClauseMiddleware.Classes; 添加到您的项目中
2. 将代码行 options.AddInterceptors(new SqlServerReaderExecutingOpenJson 添加到 OnConfiguring 方法的 options 中

拦截器代码解释

ReaderExecutingAsync | ReaderExecuting

在下面的重写方法中，您可以看到我们拦截了 LINQ（通过 EF 管道）生成的 SQL，并将其替换为我们的 OpenJson 替代 SQL 语句。我们调用了辅助方法 ConvertInClauseToOpenJson 来将原始 EF SQL 转换为我们自定义的面向 OpenJson 的 SQL 语句。

辅助类

ConvertInClauseToOpenJson

ConvertInClauseToOpenJson 方法将解析 IN\NOT IN 子句，然后将值传递给 ConvertToOpenJson 方法，该方法将用 OpenJson SQL 替换现有 SQL。

ConvertToOpenJson

然后，ConvertToOpenJson 方法将确定 IN\NOT IN 子句中的值是整数\小数还是 varchar 相关，并据此构建新的 OpenJson SQL。

您最终可以看到 OpenJson 语句被插入到 SQL 语句中并返回到父 IN\NOT IN 子句。

原始 EF SQL 和生成的 OpenJson SQL 示例

在原始 SQL（为简洁起见，值已简化）中，您可以看到一个标准的 IN 子句及其值。

然后在 OpenJson 生成的 SQL 中，您可以看到这些值与字符串相关，并且 OpenJson 正确使用了相关的 NVARCHAR 数据类型。

注意：NOT IN 子句未转换为 OpenJson，因为其值计数小于 1k。

Varchar 示例

整数示例

下面，您可以看到 IN 子句中与整数相关的值的等效值，并且正确的 OpenJson 数据类型（Int）与查询关联。

如何使用多线程和 ConcurrentBag 整理结果

通过使用 OpenJson，您可以让 SQL Server 执行带有更大 IN\NOT IN 子句的 SQL 语句。默认情况下，EF Core 6\7 将执行一个值计数小于 20K 的 IN 子句，这将是整个语句的总值（因为您的 IN 子句可以在单个 EF 生成的 SQL 语句中多次重复）。

在以前的项目中，我根据每个查询的复杂性采取了执行 OpenJson 查询的方法。我通过限制块的大小来创建值列表。

然后循环此列表，将值插入 OpenJson 查询并执行它，我使用并行循环来利用服务器可用的多个核心，然后我简单地将结果添加到（线程安全）ConcurrentBag 中。

您将在 DemoLinqContainsOpenJson 项目中的 QueryHelper 类中找到执行上述场景的示例代码

下面的代码片段将

1. 定义您的分块大小 (100K)，这是您希望一次处理的值的大小。
2. 创建要使用的并行级别。
3. 检索 100 万个值（整数 ID）
4. 创建一个分块列表（基于分块大小——上面是 1）。
5. 使用后台线程运行查询，将结果添加到您的 ConcurrentBag
6. 每个任务将循环适当的次数（分块列表大小）
7. 等待所有任务完成。

const int CHUNK_SIZE = 100_000;            
var tskDatabaseQueries = new List<Task>(); // maintain the running background tasks
var parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = 2 };
var resultBag = new ConcurrentBag<List<LargeTable>>(); // ConcurrentBag to hold all results safely

// mock a client that uses a cache of data to filter against - retrieve cached data (1M in each collection) - data will be duplicated in cases like char!
List<int> cacheInt = await Retrieve1MCacheIntegerAsync();
List<char> cacheChar = await Retrieve1MCacheCharAsync();
List<double?> cacheDouble = await Retrieve1MCacheDoubleAsync();
List<string> cacheString = await Retrieve1MCacheStringAsync();

// chunk data into 100K batches for looping
var chunkInt = cacheInt.Chunk(CHUNK_SIZE);
var chunkChar = cacheChar.Chunk(CHUNK_SIZE);
var chunkDouble = cacheDouble.Chunk(CHUNK_SIZE);
var chunkString = cacheString.Chunk(CHUNK_SIZE);

// make a linq call using Integer as the IN clause
tskDatabaseQueries.Add(Task.Run(async () =>
{
    await Parallel.ForEachAsync(chunkInt, parallelOptions, async (sublist, cancellationToken) =>
    {
        var queryResult = await RetrieveIntegerChunkAsync(sublist);
        resultBag.Add(queryResult);
    });
}));

// make a linq call using Char as the IN clause
tskDatabaseQueries.Add(Task.Run(async () =>
{
    await Parallel.ForEachAsync(chunkChar, parallelOptions, async (sublist, cancellationToken) =>
    {
        var queryResult = await RetrieveCharChunkAsync(sublist);
        resultBag.Add(queryResult);
    });
}));

// make a linq call using Double as the IN clause
tskDatabaseQueries.Add(Task.Run(async () =>
{
    await Parallel.ForEachAsync(chunkDouble, parallelOptions, async (sublist, cancellationToken) =>
    {
        var queryResult = await RetrieveDoubleChunkAsync(sublist);
        resultBag.Add(queryResult);
    });
}));

// make a linq call using String as the IN clause
tskDatabaseQueries.Add(Task.Run(async () =>
{
    await Parallel.ForEachAsync(chunkString, parallelOptions, async (sublist, cancellationToken) =>
    {
        var queryResult = await RetrieveStringChunkAsync(sublist);
        resultBag.Add(queryResult);
    });
}));

// Wait for all tasks to complete
await Task.WhenAll(tskDatabaseQueries);