我正在使用消息隊列中的消息并使用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 來做到這一點,或者我應該訴諸其他方法?
使用并行性時限制消息隊列消耗
慕田峪9158850
2023-09-09 16:21:58