为Parallel.ForEach中的项设置超时

Question

我正在使用Parallel.ForEach处理并发字典集合。ConcurrentDictionary包含Key和Byte字段。

concurrentDictionary-这里是Concurrent Dictionary集合。

Parallel.ForEach(concurentDictionary, (item) =>
{
    var mailObject = dbContext.MailObjects.Where(x=> x.MailObjectId == item.key).Select(y=> y);
    mailObject.MailBody = ConvertToPdf(item.Value.MailBody);
} 
dbContext.SaveChanges();

我想设置超时，如果任何特定项目需要设置的最长时间，比如60秒。如果花费超过60秒，我想退出，不保存该特定项目。其余项目应使用dbContext.SaveChanges()方法保存。如何在Parallel.ForEach中实现这一点？

Answer 1

这样的东西可能与您想要的差不多。

CancellationTokenSource cts = new CancellationTokenSource();
ParallelOptions parallelOptions = new ParallelOptions();
parallelOptions.CancellationToken = cts.Token;
parallelOptions.MaxDegreeOfParallelism = System.Environment.ProcessorCount;
Timer timer = new Timer(callback => { cts.Cancel(); }, null, 60*1000, Timeout.Infinite);
try
{
    Parallel.ForEach(concurentDictionary, parallelOptions, (item) =>
    {
        var mailObject = dbContext.MailObjects.Where(x => x.MailObjectId == item.key).Select(y => y);
        mailObject.MailBody = ConvertToPdf(item.Value.MailBody);
        parallelOptions.CancellationToken.ThrowIfCancellationRequested();
    });
}
catch (OperationCanceledException e)
{
    // Log the cancellation event...
}
finally
{
    dbContext.SaveChanges();
    cts.Dispose();
}

以下是解决方案的完整修订版，它为每次迭代确定了计时器的范围。它不是取消令牌，而是利用loopstate在一次（或多次）超过时间限制时停止迭代。这并不依赖于异常来退出循环。然而，在编写本文时，任何异常都会触发所有循环停止。请注意，这不会立即停止所有循环，并将完成现有循环的执行，直到它们也完成或超时。

{
    if (Parallel.ForEach(concurentDictionary, (item, loopState) =>
        {
            Timer timer = new Timer(callback => { loopState.Stop(); }, null, 60 * 1000, Timeout.Infinite);
            try
            {
                if (loopState.ShouldExitCurrentIteration || loopState.IsExceptional)
                    loopState.Stop();
                var mailObject = dbContext.MailObjects.Where(x => x.MailObjectId == item.key).Select(y => y);
                mailObject.MailBody = ConvertToPdf(item.Value.MailBody);
                if (loopState.ShouldExitCurrentIteration || loopState.IsExceptional)
                    loopState.Stop();
            }
            catch (Exception)
            {
                loopState.Stop();
                throw;
            }
            finally
            {
                timer.Dispose();
            }
        }
    ).IsCompleted)
    {
        // All events complete
    }
}
catch (AggregateException)
{
    // Execution will not get to this point until all of the iterations have completed (or one 
    // has failed, and all that were running when that failure occurred complete).
}