“无限”异步并行 foreach 循环

Question

我有一个包含 50K 到 100K 字的 List<string>

我想以并行和异步方式迭代它

例如，我可以使用

while (true)
{
   Parallel.ForEach(words, new ParallelOptions { MaxDegreeOfParallelism = 100 }, ...)
}

但问题是：

1. Parallel.ForEach 不是异步
2. 当我们到达列表的末尾时，我们必须等待每个线程结束，然后while (true) 语句才能继续
3. 这意味着没有始终运行 100 个线程，这正是我想要的

我怎样才能做到这一点？

如果这令人困惑，或者我不善于解释，请告诉我。

Answer 1

这是一个完全做作的 async 友好的 TPL DataFlow 示例，说明如何实现您的要求。

1. 适用于异步 IO 工作负载
2. 可以取消
3. 它限制了最大并行度
4. 它的容量有限，因此始终有 100 个工作可用
5. 它是无限的

给定

private static CancellationTokenSource _cs;
private static CancellationToken _token;
private static ActionBlock<string> _block;

private static async Task MethodAsync(string something)
{
   // Your async workload
}

public static async Task EndlessRunner(string[] someArray)
{
   try
   {
      var index = 0;
      while (!_token.IsCancellationRequested)
      {
         await _block.SendAsync(someArray[index],_token);
         if (++index >= someArray.Length) index = 0;
      }
   }
   catch (OperationCanceledException)
   {
      Console.WriteLine("Cancelled");
   }
}

示例

private static async Task Main()
{
   _cs = new CancellationTokenSource();
   _token = _cs.Token;

   _block = new ActionBlock<string>(
      MethodAsync, 
      new ExecutionDataflowBlockOptions()
      {
         EnsureOrdered = false,
         MaxDegreeOfParallelism = 100,
         BoundedCapacity = 100,
         CancellationToken = _cs.Token,
         SingleProducerConstrained = true
      });

   var someList = Enumerable
      .Range(0,5000)
      .Select(I => $"something {I}")
      .ToArray();

   Task.Run(() => EndlessRunner(someList));

   Console.ReadKey();

   _cs.Cancel();

   _block.Complete();
   await _block.Completion;

}