使用 Parallel.For 时的不可预测的结果

Question

我正在运行一个 Parallel for 循环，该循环最初运行时间 = 处理器数并执行长时间运行的操作。每个任务完成后，检查更多任务，如果找到，再次调用自身。

我的代码如下所示：

static void Main(string[] args)
{
   Int32 numberOfProcessors = Environment.ProcessorCount;

   Parallel.For(0, numberOfProcessors, index => DoSomething(index, sqsQueueURL));

}

private async static Task DoSomething(int index, string queueURL)
{
   var receiveMessageRequest = new ReceiveMessageRequest { QueueUrl = queueURL, WaitTimeSeconds = 20, MaxNumberOfMessages = 1, VisibilityTimeout = 1200 };

   AmazonSQSClient sqsClient = new AmazonSQSClient(new AmazonSQSConfig { MaxErrorRetry = 4 });

   var receiveMessageResponse = sqsClient.ReceiveMessage(receiveMessageRequest);

   foreach (var msg in receiveMessageResponse.Messages)
   {
      PerformALongRunningTask......

      //delete the message

      DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest(queueURL, msg.ReceiptHandle);

      AmazonSQSClient sqsDeleteClient = new AmazonSQSClient();

      sqsDeleteClient.DeleteMessage(deleteMessageRequest);

      //Do it again
      DoSometing(index,queueURL)

   }
}

我得到了非常不可预测的结果。它永远不会完成所有任务。它在完成所有操作之前退出。

我在这里做错了什么？

短代码：

static Int32 TimesToLoop = 143;
static void Main(string[] args)
{

    Int32 numberOfProcessors = Environment.ProcessorCount;

    Parallel.For(0, numberOfProcessors, index => DoSomething(index));

    Console.Read();
}

private async static Task DoSomething(int index)
{
    if(TimesToLoop == 0)
    {
        return;
    }
    Console.WriteLine(index);
    Interlocked.Decrement(ref TimesToLoop);
    DoSomething(index++);
    return;

}

Answer 1

我现在看到各种问题：

• Parallel.For 刚刚开始任务。它不会等待他们完成。它将等待DoSomething 方法调用返回，但它们正在返回代表异步操作的任务，这些操作可能不会同步完成。
• 正如 CarbineCoder 所指出的，您的递归几乎可以肯定是有缺陷的。目前尚不清楚您要实现什么目标，但您需要重新考虑这一方面
• 您的递归不会await 无论如何返回的任务 - 它几乎肯定应该。它可能想要创建在foreach 循环中创建的所有任务的集合并一次性等待它们，或者它可能想要立即await 它们。我们无法判断。

修复第一部分最简单的方法可能是使用Task.WaitAll而不是Parallel.For：

var tasks = Enumerable.Range(0, numberOfProcessors)
                      .Select(index => DoSomething(index, sqsQueueURL))
                      .ToList();
Task.WaitAll(tasks);

与Task.WhenAll 不同，Task.WaitAll 将阻塞直到所有指定的任务完成。请注意，如果任何任务需要在调用WaitAll 的线程上继续执行，这不是安全的，正是因为它阻塞了 - 但如果这是一个控制台应用程序并且你从初始线程，你会没事的，因为无论如何都会在线程池上执行延续。

Answer 2

private async static Task DoSomething(int index, string queueURL)
{
   ...
   foreach (var msg in receiveMessageResponse.Messages)
   {
      ...
      //Do it again
      DoSometing(index,queueURL)

   }
}

您正在递归调用DoSomething，并且没有条件中断/返回它。这可能会导致 stackoverflow 并终止您的程序。