使用并行性时限制消息队列消耗

Question

我正在使用消息队列中的消息并使用Task.Run() 并行处理它们。但是我想将消耗速度限制在某个最大线程数并且在线程数低于该线程数之前不要从消息队列中消耗。

假设我想要最多 100 个线程。在这种情况下，当达到 100 个线程时，它应该停止从消息队列中消费。当一个消息处理任务完成并且线程数下降到 99 时，它应该从队列中再消耗一条消息。

我尝试使用TransformBlock 来实现此目的，下面是用于演示的示例代码：

public partial class MainWindow : Window
    {
        object syncObj = new object();
        int i = 0;
        public MainWindow()
        {
            InitializeComponent();
        }


        private async Task<bool> ProcessMessage(string message)
        {
            await Task.Delay(5000);

            lock (syncObj)
            {
                i++;
                System.Diagnostics.Debug.WriteLine(i);
            }
            return true;
        }

        private async void Button_Click(object sender, RoutedEventArgs e)
        {
            var processor = new TransformBlock<string, bool>(
                    (str) => ProcessMessage(str),
                    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 100 }
                    );

            for(int i = 0; i < 1000; i++)
            {
                await processor.SendAsync("a");
            }


    }
}

限制并行任务的数量按预期工作，但所有消息都会立即发送到 TransformBlock，因此SendAsync 循环在任务处理之前结束。

只要线程数低于最大值，我希望它继续接受消息。允许并行，但在达到 100 时等待。

有没有办法使用 TransformBlock 来做到这一点，还是我应该求助于其他方法？

Answer 1

数据流块具有输入缓冲区。此输入缓冲区充当队列。

如果您想将消息保留在您的自己的队列中，您可以通过限制数据流块愿意接收的项目数来做一些接近您想要的事情：

var processor = new TransformBlock<string, bool>(
    (str) => ProcessMessage(str),
    new ExecutionDataflowBlockOptions
    {
      BoundedCapacity = 100,
      MaxDegreeOfParallelism = 100,
    }
);

请注意，BoundedCapacity 包括块正在处理的项目。由于BoundedCapacity == MaxDegreeOfParallelism，这实际上关闭了数据流块的队列。

所以 SendAsync 循环在任务处理之前结束。

当有（最多）100 个任务要处理时，它仍然会结束。如果您想等到所有项目都完成处理，请致电Complete() 和await Completed。