c# - 我是否应该在服务器上使用Parallel.ForEach进行多个同时的Web请求

Question

我已经阅读了许多有关Parallel.ForEach的内容，但是并没有真正找到我的问题的可靠答案。

我们有一个Windows服务，每隔几分钟就会从多个数据库中提取行，并使用foreach循环，通过网络请求将这些行发送出去以完成操作。因此，所有这些Web请求当前都是顺序执行的，并且花费的时间太长，因此我们希望并行运行它们。

我的最初调查使我相信Producer-Consumer approach using threads最好，在这种情况下，生产者每两分钟将行放入线程安全的队列中，并且在服务初始化期间，我只是启动了许多使用者线程(例如10个) ，但可能是100或更多)，它会不断检查队列以查看是否存在需要通过网络请求发送的行。

一位同事建议，只需将我们的foreach循环更改为Parallel.ForEach即可。我对此的第一个担心是，ForEach将阻塞所有操作，直到枚举中的所有项目都完成为止，因此，如果它在5秒内完成10项并且完成9项，在5分钟内完成一项，那么实际上除了一个请求持续4分55秒。只需在新线程中执行Parallel.ForEach即可解决，如下所示:

Task.Factory.StartNew( () => Parallel.ForEach<Item>(items, item => DoSomething(item)));

因此，每隔几分钟就会发生一次新的Parallel.ForEach循环，该循环将使用自上次检查以来已添加到数据库中的所有新行，即使先前的Parallel.ForEach循环尚未完成(即5分钟的请求不会阻止新请求的产生)。

这很容易做到，并且最大程度地减少了需要进行的代码更改，但是我仍然担心在托管其他服务和网站的服务器上运行此代码。我读过Parallel.ForEach可以将服务器上的所有CPU固定，即使简单的Web请求不会占用大量CPU资源。我知道我可以通过使用MaxDegreeOfParallelism property来限制循环使用的线程数，因此我可以将其设置为10或100或任何其他值。这很好，因为Parallel.ForEach不会连续运行10或100个任务，而是什么也不做，而是将其旋转很多，然后在循环完成时将其关闭。但是我仍然很犹豫，它可能会消耗服务器上太多的资源。

那么，这些选项(或其他选项)中哪一个最适合我的情况？我对在服务器计算机上使用Parallel.ForEach感到担心吗？它肯定看起来像“简单”和“懒惰”的解决方案，所以我只想确保如果我们使用它，它不会再次咬我。另外，我也不在乎将此解决方案扩展到多个服务器。仅在也可以运行其他服务和网站的单个服务器上运行。

更新

这些注释要求提供一些源代码以提供更多上下文。

这是我们目前正在做的简化版本:

void FunctionGetsCalledEvery2Minutes()
{
    // Synchronously loop over each database that we need to check.
    foreach (var database in databasesToCheck)
    {
        // Get the rows from this database.
        var rows = database.GetRowsFromTable();

        // Synchronously send each row to a web service to be processed.
        foreach (var request in rows)
        {
            SendRequestToWebServiceToBeProcessed(request);
        }
    }
}

SendRequestToWebServiceToBeProcessed(DatabaseRow request)
{
    // Request may take anywhere from 1 second to 10 minutes.
    Thread.Sleep(_randomNumberGenerator.Next(1000, 600000));
}

这是使用Parallel.ForEach的代码外观的简化版本:

void FunctionGetsCalledEvery2Minutes()
{
    // Synchronously loop over each database that we need to check.
    foreach (var database in databasesToCheck)
    {
        // Get the rows from this database.
        var rows = database.GetRowsFromTable();

        // Asynchronously send each row to a web service to be processed, processing no more than 30 at a time.
        // Call the Parallel.ForEach from a new Task so that it does not block until all rows have been sent.
        Task.Factory.StartNew(() => Parallel.ForEach<DatabaseRow>(rows, new ParallelOptions() { MaxDegreeOfParallelism = 30 }, SendRequestToWebServiceToBeProcessed));
    }
}

这是使用生产者-消费者代码的简化形式:

private System.Collections.Concurrent.BlockingCollection<DatabaseRow> _threadSafeQueue = new System.Collections.Concurrent.BlockingCollection<DatabaseRow>();
void FunctionGetsCalledEvery2Minutes()
{
    // Synchronously loop over each database that we need to check.
    foreach (var database in databasesToCheck)
    {
        // Get the rows from this database.
        var rows = database.GetRowsFromTable();

        // Add the rows to the queue to be processed by the consumer threads.
        foreach (var row in rows)
        {
            _threadSafeQueue.Add(row);
        }
    }
}

void ConsumerCode()
{
    // Take a request off the queue and send it away to be processed.
    var request = _threadSafeQueue.Take();
    SendRequestToWebServiceToBeProcessed(request);
}

void CreateConsumerThreadsOnApplicationStartup(int numberOfConsumersToCreate)
{
    // Create the number of consumer threads specified.
    for (int i = 0; i < numberOfConsumersTo; i++)
    {
        Task.Factory.StartNew(ConsumerCode);
    }
}

在此示例中，我有一个同步生产者，但是我可以轻松地为每个要查询的数据库启动一个异步生产者线程。

这里要注意的一件事是，在Parallel.ForEach示例中，我将其限制为一次只能处理30个线程，但这仅适用于该实例。如果2分钟过去了，并且Parallel.ForEach循环仍然有10个未完成的请求，它将启动30个新线程，总共40个线程同时运行。因此，如果Web请求的超时时间为10分钟，那么我们很容易遇到同时运行150个线程的情况(10分钟/2分钟=调用函数5次*每个实例30个线程= 150)。这是一个潜在的问题，就像我增加了最大线程数，或者以小于2分钟的时间间隔开始调用该函数一样，我很快可能会同时运行数千个线程，从而消耗了服务器上比我更多的资源。想。这是一个有效的问题吗？消费者-生产者方法不存在此问题；它只会运行与我为numberOfConsumersToCreate变量指定的数量一样多的线程。

已经提到我应该为此使用TPL数据流，但是我以前从未使用过它们，也不想在此项目上花费大量时间。如果TPL数据流仍然是我想知道的最佳选择，但是我也想知道这两种方法(Parallel.ForEach与Producer-Consumer)哪种方法更适合我的情况。

希望这会提供更多的背景信息，以便我能获得更好的针对性答案。谢谢 :)

Answer 1

如果您有许多短操作而偶尔有长操作，则Parallel.ForEach将阻塞，直到所有操作完成为止。但是，当它针对一个长期的请求进行工作时，它并不会固定您所有的内核，而只是仍在工作。请记住，当正在处理许多项目时，它将尝试使用所有内核。

编辑:

使用MaxDegreeOfParallelism属性，没有理由将其设置为CPU可以运行的线程数量以上(受内核数量和超线程程度限制)。实际上，将其减少到低于该值仅是有用的。

因为阻塞不是问题Parallel.ForEach，虽然看起来很懒，但如果您的项目确实可以并发运行，则非常合适。