使用Dataflow的BatchBlock时生产者过度生产问题排查

你的问题核心是对Dataflow中BoundedCapacity的作用范围理解有误：

• BatchBlock的BoundedCapacity=1000限制的是它内部待批处理的单个元素数量，而非整个数据流的总待处理元素数。当BatchBlock攒够500个元素形成批后，会立即将这个批发送给下游的ActionBlock。只要ActionBlock的输入队列有空间接收批，BatchBlock就会释放这500个元素的容量，生产者就能继续发送新元素，不会被阻塞。
• 你的ActionBlock设置了BoundedCapacity=1000，但它接收的是int[]类型的批数据（每个批含500条原始元素），这意味着它最多能缓存1000个批，对应50万条原始元素，远大于你生成的24万条，所以BatchBlock的批总能顺利发送，不会被阻塞，自然生产者也不会停止。

修复方案 #

要实现“生产者生成1000条消息后等待”的效果，需要在生产者和BatchBlock之间新增一个BufferBlock，用它来限制整个上游的待处理元素总数：

[Test]
public async Task Run()
{
    int producerCount = 0;
    int consumerCount = 0;

    Action logProgress = () =>
    {
        if (consumerCount % 1000 == 0 && consumerCount > 0 || producerCount % 1000 == 0 || consumerCount == producerCount)
        {
            Trace.WriteLine($"Progress - {consumerCount}/{producerCount}");
        }
    };

    // 新增BufferBlock做全局限流，总容量1000
    var bufferBlock = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 1000 });
    var batchBlock = new BatchBlock<int>(500);
    var actionBlock = new ActionBlock<int[]>(async x =>
    {
        await Task.Delay(10);
        foreach (var specificItem in x)
        {
            Interlocked.Increment(ref consumerCount);
        }
        logProgress();
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 1
        // 此处无需设置BoundedCapacity，限流由上游BufferBlock完成
    });

    // 链接各个数据流块
    bufferBlock.LinkTo(batchBlock, new DataflowLinkOptions { PropagateCompletion = true });
    batchBlock.LinkTo(actionBlock, new DataflowLinkOptions { PropagateCompletion = true });

    int[] items = Enumerable.Range(0, 240000).ToArray();
    foreach (var item in items)
    {
        await bufferBlock.SendAsync(item); // 生产者发送到BufferBlock
        Interlocked.Increment(ref producerCount);
        logProgress();
    }

    bufferBlock.Complete(); // 标记BufferBlock完成

    await Task.WhenAll(bufferBlock.Completion, actionBlock.Completion);
}

原理说明 #

• BufferBlock的BoundedCapacity=1000会严格限制内部缓存的单个元素数量，当队列满时，SendAsync会异步等待，直到有元素被BatchBlock取走，从而实现生产者在生成1000条后暂停的效果。
• BatchBlock无需再设置BoundedCapacity，因为上游的BufferBlock已经完成了限流，它只负责按指定大小攒批即可。

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