并行消费消息队列时的流量控制：TransformBlock使用疑问

你遇到的问题其实是因为TransformBlock默认的内部队列是无界的，所以SendAsync会一股脑把1000条消息都塞进去，哪怕只有100个线程在处理——剩下的消息都在队列里排队，完全没达到“线程数满了就暂停发送，有空位再继续”的效果。

其实用TransformBlock完全可以实现你想要的逻辑，只需要调整一个配置项就行，下面给你两种可行的方案：

方案1：调整TransformBlock的容量限制 #

只需要在ExecutionDataflowBlockOptions里加上BoundedCapacity = 100，这个参数会限制TransformBlock内部能容纳的消息总数（包括正在处理的和等待处理的）。当块的容量满了之后，SendAsync会自动异步等待，直到有消息处理完成、腾出空间后才会完成发送。

修改后的代码如下：

public partial class MainWindow : Window { 
    private readonly object _syncObj = new object(); 
    private int _processedCount = 0; 

    public MainWindow() { 
        InitializeComponent(); 
    } 

    private async Task<bool> ProcessMessage(string message) { 
        await Task.Delay(5000); 
        lock (_syncObj) { 
            _processedCount++; 
            System.Diagnostics.Debug.WriteLine(_processedCount); 
        } 
        return true; 
    } 

    private async void Button_Click(object sender, RoutedEventArgs e) { 
        var processor = new TransformBlock<string, bool>(
            str => ProcessMessage(str),
            new ExecutionDataflowBlockOptions { 
                MaxDegreeOfParallelism = 100,
                BoundedCapacity = 100 // 关键：限制块的总容量
            } 
        ); 

        for(int i = 0; i < 1000; i++) {
            // 当块满了时，SendAsync会等待，直到有空闲处理槽
            await processor.SendAsync("a"); 
        } 

        // 标记块不再接受新消息，否则它会一直等待
        processor.Complete();
        // 等待所有消息处理完成
        await processor.Completion;
    } 
}

这样修改后，你的SendAsync循环就不会一下子跑完了——当100个线程都在处理消息时，TransformBlock的容量就满了，后续的SendAsync会暂停，直到有一个任务处理完成、腾出位置，才会继续发送下一条消息，完美匹配你的需求。

方案2：用SemaphoreSlim手动控制并发 #

如果你觉得Dataflow的配置不够直观，也可以用SemaphoreSlim来直接控制并发数，逻辑更直白：

private async void Button_Click(object sender, RoutedEventArgs e) {
    // 初始化信号量，最多允许100个并发任务
    var semaphore = new SemaphoreSlim(100);
    var processingTasks = new List<Task>();

    for(int i = 0; i < 1000; i++) {
        // 等待有空闲的并发槽位
        await semaphore.WaitAsync();
        processingTasks.Add(Task.Run(async () => {
            try {
                await ProcessMessage("a");
            } finally {
                // 任务完成后释放槽位，允许下一个任务启动
                semaphore.Release();
            }
        }));
    }

    // 等待所有消息处理完成
    await Task.WhenAll(processingTasks);
}

这个方案通过信号量直接控制同时运行的任务数，达到100上限时就等待，任务完成后释放信号量，让下一个任务启动，效果和方案1完全一致。

小提示 #

不管用哪种方案，记得最后要等待所有任务处理完成，否则程序可能在消息处理完毕前就退出了，或者导致资源泄漏。

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