Maui 实践：用 Channel 实现数据库查询时读取速度与内存占用的平衡

技术分享 8个月前 (07-18) 0 999+

作者：夏群林原创 2025.7.17

我们在进行数据库查询时，通常并不是为了取得整个表的数据，而是某些符合过滤条件的记录。比如：

var unassociatedSudokus = await _dbContext.DbSudokus                 .Where(s => !relatedSudokuIds.Contains(s.ID))                 .ToListAsync();

这里 relatedSudokuIds 元素不多，过滤条件 s => !relatedSudokuIds.Contains(s.ID) 简单，运算量不大。所以，一切正常。

但我们经常会加入别的过滤条件，比如：

var unassociatedSudokus = await _dbContext.DbSudokus                 .Where(s => !relatedSudokuIds.Contains(s.ID) && string.Join(string.Empty, s.Ans) == targetAns)                 .ToListAsync();

过滤条件 string.Join(string.Empty, s.Ans) == targetAns 还算简单，但因为在读取数据表的时候现场装配 s.Ans，速度变慢，记录量大时，延迟明显。

推荐方案，是在数据库设计层面优化，添加一个存储拼接后结果的字段并建立索引。一般来说，可以从根本上解决读取速读慢的问题。

偏偏我的应用，DbSudokus 表非常大，而需要这种查询的场景，却不多。我舍不得多加一个 Column，让 DbSudokus 数据表无谓地臃肿。

另一个方案，数据库加载阶段简单过滤，将拼接之类的复杂过滤运算放在数据库加载之后，在内存里过滤：

  var unassociatedSudokus = await _dbContext.DbSudokus             .Where(s => !relatedSudokuIds.Contains(s.ID))             .ToListAsync();    unassociatedSudokus = [.. unassociatedSudokus.Where(s => string.Join(string.Empty, s.Ans) == targetAns)];

代价是一次性加载全部数据，内存占用过多。尤其在我的应用的情形，数据表本身很大，过滤后的结果集很小，总觉得不划算。

于是想到用 Channel。

Channel 主要是通过流处理的方式来平衡性能和内存占用。原理是：

先从数据库分批读取数据（避免一次性加载全部数据）
通过 Channel 将数据逐个或批量传递到消费者
在消费者端进行内存中的字符串拼接和比较等耗时运算
只保留符合条件的结果

这样既避免了在数据库中执行复杂操作，如字符串拼接等，可能无法有效利用索引，又避免了一次性加载所有数据导致的高内存占用。数据一边读取一边处理，通过批次大小和通道容量，可以限制同时加载到内存中的数据量，而且流处理，不需要等待全部数据加载完成。预计到结果集很小时，大部分数据需要被过滤掉，使用 Channel 优势明显，在流处理读取过程中尽早过滤掉不需要的数据，自然降低了内存占用。

我的想法，把过滤条件切分两部分：

第一，简单的部分，放在数据库加载阶段，有 Channel 的生产者处理，并且可接受消费者的通知，提前结束数据库读入：

// 生产者任务：支持提前终止 var producerTask = Task.Run(async () => {     try     {         var page = 0;         while (!stopProcessing) // 当消费者发现足够结果时可以提前停止         {             var batch = await _dbContext.DbSudokus                 .Where(s => !relatedSudokuIds.Contains(s.ID))                 .Skip(page * batchSize)                 .Take(batchSize)                 .ToListAsync(cancellationToken);              if (batch.Count == 0)                 break;              foreach (var sudoku in batch)             {                 // 再次检查是否需要停止，避免写入多余数据                 if (stopProcessing) break;                  await channel.Writer.WriteAsync(sudoku, cancellationToken);             }              page++;         }     }     finally     {         channel.Writer.Complete();     } });

第二，复杂的部分，

 // 消费者任务：找到足够结果后可以提前停止 var consumerTask = Task.Run(async () => {     await foreach (var sudoku in channel.Reader.ReadAllAsync(cancellationToken))     {         // 检查是否已经找到足够的结果         if (maxResults.HasValue && result.Count >= maxResults.Value)         {             stopProcessing = true;             break;         }          // 内存中过滤         var ansString = string.Join(string.Empty, sudoku.Ans);         if (ansString == targetAns)         {             lock (result)             {                 result.Add(sudoku);             }         }     } });

如果知道大概的结果数量，可以设置 maxResults 参数，得到额外的提前终止的好处。例如，如果通常只需要找到 1-2 条匹配结果，就可以将 maxResults 设为 2，系统会在找到 2 条结果后立即停止所有操作。

进一步，我们可以把上面的做法泛型化，核心是分离数据库端和内存端筛选逻辑，以兼顾性能和灵活性。具体做法，是把筛选逻辑包装成委托，作为参数传入。

最后，给出我的实现代码。这是一个 LINQ 风格 IQueryable 扩展方法，具有高度通用性，适用于任何实体类型和筛选场景，调用很方便。替换现有代码，几乎没有侵入性。我就是在自己应用的生产性代码中，原行替换的。

using Microsoft.EntityFrameworkCore; using System.Linq.Expressions;  namespace Zhally.Sudoku.Data;  public static class QueryFilterExtensions {     /// <summary>     /// 流式筛选IQueryable数据，平衡性能和内存占用     /// </summary>     /// <typeparam name="T">实体类型</typeparam>     /// <param name="query">原始查询</param>     /// <param name="productionFilter">数据库端筛选表达式（生产阶段）</param>     /// <param name="consumptionFilter">内存端筛选委托（消费阶段）</param>     /// <param name="batchSize">批次大小</param>     /// <param name="maxResults">最大结果数量（达到后提前终止）</param>     /// <returns>筛选后的结果列表</returns>     public static async Task<List<T>> FilterWithChannelAsync<T>(         this IQueryable<T> query,         Expression<Func<T, bool>> productionFilter,         Func<T, bool> consumptionFilter,         int batchSize = 100,         int? maxResults = null)         where T : class     {         // 创建有界通道控制内存占用         var channel = Channel.CreateBounded<T>(new BoundedChannelOptions(100)         {             FullMode = BoundedChannelFullMode.Wait,             SingleReader = true,             SingleWriter = true         });          var result = new List<T>();         var cancellationToken = CancellationToken.None;         bool stopProcessing = false;          // 消费者任务：处理并筛选数据         var consumerTask = Task.Run(async () =>         {             await foreach (var item in channel.Reader.ReadAllAsync(cancellationToken))             {                 // 检查是否已达到最大结果数                 if (maxResults.HasValue && result.Count >= maxResults.Value)                 {                     stopProcessing = true;                     break;                 }                  // 应用内存筛选条件                 if (consumptionFilter(item))                 {                     lock (result)                     {                         result.Add(item);                     }                 }             }         });          // 生产者任务：从数据库分批读取数据         var producerTask = Task.Run(async () =>         {             try             {                 var page = 0;                 while (!stopProcessing)                 {                     // 应用数据库筛选并分页查询                     var batch = await query                         .Where(productionFilter)                         .Skip(page * batchSize)                         .Take(batchSize)                         .ToListAsync(cancellationToken);                      if (batch.Count == 0)                         break; // 没有更多数据                      // 将批次数据写入通道                     foreach (var item in batch)                     {                         if (stopProcessing) break;                         await channel.Writer.WriteAsync(item, cancellationToken);                     }                      page++;                 }             }             finally             {                 channel.Writer.Complete(); // 通知消费者数据已写完             }         });          // 等待所有任务完成         await Task.WhenAll(producerTask, consumerTask);          return result;     } }

这种设计特别适合以下场景：

需要在数据库端做初步筛选，再在内存中做复杂筛选
预期结果集较小，但源数据集可能很大
希望平衡数据库负载和内存占用

还可以根据实际需求调整批次大小和通道容量，以获得最佳性能。

使用示例：

var result = await _dbContext.DbSudokus     .FilterWithChannelAsync(         // 数据库端筛选：排除关联的Sudoku         s => !relatedSudokuIds.Contains(s.ID),         // 内存端筛选：比较拼接后的答案         s => string.Join(string.Empty, s.Ans) == targetAns,         batchSize: 100,         maxResults: null     );

发表评论