如何为队列添加带超时机制的批量缓冲功能？

我想要开发一个从队列中读取条目并进行处理的应用程序。由于处理这些条目成本较高，我希望以指定数量的批次进行处理。此外，我需要一个「超时」机制：自上次处理后经过指定时长时，即使批次未填满也要处理现有条目。

目前已实现按100条批量处理，但缺少超时机制，希望补充：收到批次中的第一个条目后，若经过X时长则输出该批次；若批次提前填满，则立即输出。

现有代码：

public async Task Run(CancellationToken cancellationToken)
{
    var stream = GetItemsFromQueue(cancellationToken)
        .Buffer(100) // 100 items batches
        .WithCancellation(cancellationToken);
    
    await foreach (var elements in stream)
    {
        // expensive processing...
    }
}

private async IAsyncEnumerable<Item> GetItemsFromQueue([EnumeratorCancellation] CancellationToken cancellationToken)
{
    while (!cancellationToken.IsCancellationRequested)
    {
        yield return await _queue.DequeueAsync(cancellationToken);
    }
}

推荐使用TPL Dataflow的BatchBlock<T>组件，它原生支持批量大小和超时触发逻辑，完美匹配你的需求：达到指定批量数时立即输出批次，或从第一个元素进入块后超时则输出当前已收集的条目。

步骤1：安装依赖包 #

先安装TPL Dataflow的NuGet包：

# NuGet包管理器命令
Install-Package System.Threading.Tasks.Dataflow

# 或.NET CLI命令
dotnet add package System.Threading.Tasks.Dataflow

步骤2：修改实现代码 #

using System.Threading.Tasks.Dataflow;

public async Task Run(CancellationToken cancellationToken)
{
    // 配置BatchBlock：批量100条，超时时间设为5秒（可自行调整）
    var batchBlock = new BatchBlock<Item>(
        batchSize: 100,
        new GroupingDataflowBlockOptions
        {
            BatchingTimeout = TimeSpan.FromSeconds(5),
            CancellationToken = cancellationToken
        });

    // 后台任务：从队列取数据并发送到BatchBlock
    _ = Task.Run(async () =>
    {
        await foreach (var item in GetItemsFromQueue(cancellationToken))
        {
            await batchBlock.SendAsync(item, cancellationToken);
        }
        // 所有数据发送完成后，标记块结束
        batchBlock.Complete();
    }, cancellationToken);

    // 循环接收批次并处理
    while (await batchBlock.OutputAvailableAsync(cancellationToken))
    {
        var batch = await batchBlock.ReceiveAsync(cancellationToken);
        // 执行昂贵的处理逻辑
        // expensive processing...
    }

    // 等待块完成所有剩余操作
    await batchBlock.Completion;
}

private async IAsyncEnumerable<Item> GetItemsFromQueue([EnumeratorCancellation] CancellationToken cancellationToken)
{
    while (!cancellationToken.IsCancellationRequested)
    {
        yield return await _queue.DequeueAsync(cancellationToken);
    }
}

方案说明 #

• BatchBlock<T>自动处理两种触发条件：
  1. 收集满100条时立即输出批次；
  2. 第一个条目进入块后，超过指定超时时间仍未凑够数量，立即输出当前已收集的条目。
• SendAsync和ReceiveAsync会自动响应取消令牌，保证取消操作的正确性。
• 最后调用Complete()和等待Completion确保所有剩余条目都被处理完毕。

替代方案：使用System.Linq.Async的Buffer重载 #

若更倾向于IAsyncEnumerable的写法，可使用System.Linq.Async的Buffer重载实现：

1. 先安装包：

Install-Package System.Linq.Async

2. 修改代码：

public async Task Run(CancellationToken cancellationToken)
{
    var stream = GetItemsFromQueue(cancellationToken)
        .Buffer(
            count: 100,
            timeSpan: TimeSpan.FromSeconds(5),
            cancellationToken: cancellationToken)
        .WithCancellation(cancellationToken);
    
    await foreach (var elements in stream)
    {
        if (elements.Any()) // 过滤取消时可能出现的空批次
        {
            // expensive processing...
        }
    }
}

注意：该方案的超时是固定间隔触发（从订阅开始计时），而非从第一个条目进入批次开始计时，若严格匹配你的补充需求，优先选择TPL Dataflow方案。

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