“无限”异步并行foreach循环

如何解决“无限”异步并行foreach循环

我有一个List<string>，包含5万到10万个字

我想以 parallel 和 asynchronous 的方式遍历它

例如，我可以使用

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

但是问题是：

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

我将如何实现这一目标？

请让我知道这是否令人困惑，或者我不善于解释事物。

解决方法

这是一个完全人为的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;

}