TPL Dataflow中间块如何分批输出结果以降低内存占用？

方案一：ActionBlock直接发送批次到下游块 #

核心思路是让中间的分批处理块持有下游块的引用，每完成一批处理就立即将结果发送到下游，无需等待所有批次处理完毕。

示例代码：

// 定义下游块，负责处理每一批数据
var downstreamBlock = new ActionBlock<List<YourDataType>>(batch =>
{
    // 这里写入下游处理逻辑
    Console.WriteLine($"处理批次，共{batch.Count}条数据");
});

// 中间分批处理块
var batchProcessingBlock = new ActionBlock<IEnumerable<YourDataType>>(async rawData =>
{
    const int BatchSize = 1000;
    var currentBatch = new List<YourDataType>(BatchSize);

    foreach (var item in rawData)
    {
        currentBatch.Add(item);

        // 达到批次大小，处理并发送
        if (currentBatch.Count == BatchSize)
        {
            await ProcessSingleBatchAsync(currentBatch); // 自定义的批次预处理逻辑
            await downstreamBlock.SendAsync(currentBatch);
            currentBatch.Clear();
        }
    }

    // 处理剩余的不足一批的数据
    if (currentBatch.Count > 0)
    {
        await ProcessSingleBatchAsync(currentBatch);
        await downstreamBlock.SendAsync(currentBatch);
    }

    // 所有批次发送完成后，标记下游块完成
    downstreamBlock.Complete();
});

// 辅助方法：模拟批次处理逻辑
async Task ProcessSingleBatchAsync(List<YourDataType> batch)
{
    // 这里写入你的批次处理代码，比如验证、转换等
    await Task.Delay(100); // 示例异步操作
}

注意事项：

• 如果下游块会修改批次列表内容，建议在发送前创建列表副本，避免影响上游的批次复用
• 若需要处理多个输入源，要注意下游块的并发设置，避免过载

方案二：使用TransformManyBlock结合异步枚举 #

TransformManyBlock支持返回IAsyncEnumerable<T>，通过yield return可以在每批处理完成后立即将结果传递到下游，无需等待所有批次生成，完美解决内存占用过高的问题。

示例代码：

// 定义下游块
var downstreamBlock = new ActionBlock<List<YourDataType>>(batch =>
{
    Console.WriteLine($"处理批次，共{batch.Count}条数据");
});

// 分批转换块：将原始数据拆分为多个批次，实时传递
var batchTransformBlock = new TransformManyBlock<IEnumerable<YourDataType>, List<YourDataType>>(async rawData =>
{
    const int BatchSize = 1000;
    var currentBatch = new List<YourDataType>(BatchSize);

    foreach (var item in rawData)
    {
        currentBatch.Add(item);

        if (currentBatch.Count == BatchSize)
        {
            await ProcessSingleBatchAsync(currentBatch);
            yield return currentBatch;
            currentBatch = new List<YourDataType>(BatchSize); // 新建列表避免共享引用问题
        }
    }

    // 处理剩余数据
    if (currentBatch.Count > 0)
    {
        await ProcessSingleBatchAsync(currentBatch);
        yield return currentBatch;
    }
});

// 链接块并传递完成信号
batchTransformBlock.LinkTo(downstreamBlock, new DataflowLinkOptions { PropagateCompletion = true });

// 辅助方法同上
async Task ProcessSingleBatchAsync(List<YourDataType> batch)
{
    await Task.Delay(100);
}

优势：

• 符合TPL Dataflow的链式编程模式，代码更简洁易维护
• 自动处理完成信号的传递（通过PropagateCompletion），无需手动调用下游块的Complete()

关键注意点 #

• 批次大小的设置要结合业务场景和内存情况，过大可能导致内存占用高，过小则会增加上下文切换开销
• 若处理的是流式数据（比如从数据库或文件逐行读取），可以将输入改为IAsyncEnumerable<YourDataType>，进一步优化内存使用
• 可以通过设置块的MaxDegreeOfParallelism来控制并发处理能力，但要注意线程安全问题

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