作者：夏群林 原创 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 ); 
