Parallel.ForEach 迭代中的操作超时答案

【问题标题】：Timeout for Action in Parallel.ForEach iterationParallel.ForEach 迭代中的操作超时
【发布时间】：2014-05-04 01:02:22
【问题描述】：

我的代码中有类似的东西：

Parallel.ForEach(myList, new ParallelOptions { MaxDegreeOfParallelism = 4 }, item =>
{
    Process(item);
});

问题是我在Process() 方法中做了很多事情（连接到文件共享、解析文件、保存到数据库等），我担心在此过程中可能会出现问题，从而使迭代永远不会完成...这会发生吗？

有没有办法为Process() 方法设置超时以避免最终出现僵尸线程？

更新：

我发现设置超时的最简单方法是向CancellationTokenSource 添加毫秒数或在任务上调用Wait() 方法。

选项#1

Parallel.ForEach(myList, new ParallelOptions { MaxDegreeOfParallelism = 4 }, item =>
{
    var cts = new CancellationTokenSource(2000);
    Task task = Task.Factory.StartNew(() => Process(item), cts.Token);
});

选项#2

Parallel.ForEach(myList, new ParallelOptions { MaxDegreeOfParallelism = 4 }, item =>
{
    Task task = Task.Factory.StartNew(() => Process(item));
    task.Wait(2000);
});

问题是这些选项都不能取消Process() 方法。我需要检查Process() 方法中的某些内容吗？

【问题讨论】：

“这会发生吗？”绝对...如果您有一个永远不会完成的进程（等待永远不会发生的 IO、无限循环等...），那么循环将永远不会完成。
所以我需要找到一种方法来避免无休止的迭代......也许通过用 Threading.Timer 包围 Process 方法调用？
您是坚持使用 .NET 4，还是可以迁移到 .NET 4.5？如果可以，那么您有更好的选择。
我正在使用 .NET 4.5 ...任何想法都值得赞赏！
只是想知道为什么要在 Parallel.ForEach 中创建任务？我认为 Parallel.ForEach 已经为您做到了...

标签： c# task-parallel-library c#-5.0 parallel.foreach

【解决方案1】：

考虑将CancellationToken 添加到您的代码中。这样，您可以随时正确取消所有操作。

然后，您可以使用CancelAfter() 方法。

【讨论】：

这有点多余：CancellationTokenSource 具有 CancelAfter 方法，并且在其构造函数覆盖之一中接受超时。
非常真实！答案已调整。
糟糕，您可能希望保留这两个版本，抱歉 :) 看来 OP 使用的是 .NET 4.0，它没有这两个版本。您可能还想补充一点，他需要观察并行循环内每个工作单元的取消：stackoverflow.com/a/22549046/1768303
我改变了我的问题，因为我意识到它们不够清楚。

【解决方案2】：

我最终结合了这两个选项。它有效，但我不知道这是否是正确的方法。

解决方案：

        Parallel.ForEach(myList, new ParallelOptions { MaxDegreeOfParallelism = 4 }, item =>
        {
                var tokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
                var token = tokenSource.Token;

                Task task = Task.Factory.StartNew(() => Process(item, token), token);
                task.Wait();
        });

在Process() 中，我多次检查是否取消：

    private void Process(MyItem item, CancellationToken token)
    {
        try
        {
            if (token.IsCancellationRequested)
                token.ThrowIfCancellationRequested();

            ...sentences

            if (token.IsCancellationRequested)
                token.ThrowIfCancellationRequested();

            ...more sentences

            if (token.IsCancellationRequested)
                token.ThrowIfCancellationRequested();

            ...etc
        }
        catch(Exception ex)
            Console.WriteLine("Operation cancelled");

【讨论】：

这个实现有很多我不喜欢的地方，但强制 ForEach 方法的每次迭代在单独的任务中运行是我发现最丑陋的。

【解决方案3】：

我最终得到了一个稍微不同的实现。就我而言，检查CancellationToken Process() 内由于检查之间可能长时间运行语句而无法工作。例如，如果我的超时时间是 5 秒，而单个语句需要 100 秒......直到该语句完成然后被if (token.IsCancellationRequested) 检测到，我才会知道。

这就是我最终做的事情

     Parallel.ForEach(myList, (item) =>
     {
         Task task = Task.Factory.StartNew(() =>
         {
             Process(item));
         });

         if(task.Wait(10000)) // Specify overall timeout for Process() here
             Console.WriteLine("Didn't Time out. All's good!"); 
         else
             Console.WriteLine("Timed out. Leave this one and continue with the rest"); 
     });

然后在Process() 方法中，我添加了对可能长时间运行的语句的进一步检查，以允许它优雅地处理超时（尽可能）。所以只有在最坏的情况下 Process() 必须被上面的Task.Wait() 提前停止。

    private void Process(MyItem item)
    {
      ...
        cmd.CommandTimeout = 5; // The total of timeouts within Process() were 
                                // set to be less than the total Task.Wait duration.
                                // Unfortunately some potentially long running methods 
                                // don't have a built in timeout.
      ...
    }

【讨论】：