在并行方法中抛出原始异常而不是聚合异常

Question

我在 Parallel.Invoke 调用中有两个 CPU 密集型方法：

Parallel.Invoke(
    () => { GetMaxRateDict(tradeOffObj); },
    () => { GetMinRateDict(tradeOffObj); }
);

对于 MCVE，假设：

public void GetMaxRateDict(object junk)
{
    throw new Exception("Max exception raised, do foo...");
}
public void GetMinRateDict(object moreJunk)
{
    throw new Exception("Min exception raised, do bar...")
}

我在每个内部方法中抛出不同的异常。但是，如果抛出其中之一，Parallel 包装器将抛出一个更通用的异常："One or more errors occurred"，它足够具体，可以在我的 UI 层中显示。

我能否以某种方式获取原始异常并将其抛出？

我希望并行任务在可能的情况下完全停止以引发内部异常，但如果这不可能，至少我需要在两种方法完成后能够引发它。谢谢

Answer 1

不太确定给定的示例是否会回答您的问题，但它可能会改进整体解决方案：

private static void ProcessDataInParallel(byte[] data)

{
    // use ConcurrentQueue to enable safe enqueueing from multiple threads.
    var exceptions = new ConcurrentQueue<Exception>();

    // execute the complete loop and capture all exceptions
    Parallel.ForEach(data, d =>
    {
        try
        {
            // something that might fail goes here...
        }
        // accumulate stuff, be patient ;)
        catch (Exception e) { exceptions.Enqueue(e); }
    });

    // check whether something failed?..
    if (exceptions.Count > 0) // do whatever you like ;
}

这种方法在将不同类型的异常收集到不同的队列（如有必要）或进一步 re-throwing 聚合异常方面提供了额外的自由（这样就不会冒出敏感信息，或者您可以传达特定的异常user-friendly 可能原因的描述等）。

通常，这是使用并行化进行异常管理的正确方法。不仅在 C# 中。

Answer 2

Can I grab the original exception somehow and throw it instead?

"It" 意味着只会有例外。即使这可能是正确的，因为您正在并行执行操作，所以您不能 100% 排除多个操作引发异常的可能性，即使您在第一个异常之后尝试取消其他操作也是如此。如果您同意这一点，我们可以从这样的假设出发：我们只期待一个异常，并且我们可以只捕获一个异常。（如果您允许另一个调用在一个抛出异常后继续进行，则出现两个异常的可能性会增加。）

您可以使用取消令牌。如果下面的调用之一抛出异常，它应该捕获该异常，将其放入变量或队列中，然后调用

source.Cancel;

这样做会导致整个 Parallel.Invoke 抛出一个 OperationCanceledException。您可以捕获该异常，检索已设置的异常，然后重新抛出它。

作为实践，我将采纳其他答案的 ConcurrentQueue 建议，因为我认为我们不能排除第二个线程可能抛出异常的可能性在被取消之前。

这开始看起来很小，但最终它变得如此复杂，以至于我将它分成了自己的 class。这让我怀疑我的方法是否不必要地复杂。主要目的是防止混乱的取消逻辑污染您的 GetMaxRateDict 和 GetMinRateDict 方法。

除了让您的原始方法不受污染和可测试之外，这个 class 本身也是可测试的。

我想我会从其他回复中找出这是一个不错的方法还是有更简单的方法。 我不能说我对这个解决方案特别兴奋。我只是觉得这很有趣，想写一些东西来完成你的要求。

public class ParallelInvokesMultipleInvocationsAndThrowsOneException //names are hard
{
    public void InvokeActions(params Action[] actions)
    {
        using (CancellationTokenSource source = new CancellationTokenSource())
        {
            // The invocations can put their exceptions here.
            var exceptions = new ConcurrentQueue<Exception>();

            var wrappedActions = actions
                .Select(action => new Action(() =>
                    InvokeAndCancelOthersOnException(action, source, exceptions)))
                .ToArray();
            try
            {
                Parallel.Invoke(new ParallelOptions{CancellationToken = source.Token}, 
                    wrappedActions)
            }
            // if any of the invocations throw an exception, 
            // the parallel invocation will get canceled and 
            // throw an OperationCanceledException;
            catch (OperationCanceledException ex)
            {
                Exception invocationException;
                if (exceptions.TryDequeue(out invocationException))
                {
                    //rethrow however you wish.
                    throw new Exception(ex.Message, invocationException);
                }
                // You shouldn't reach this point, but if you do, throw something else.
                // In the unlikely but possible event that you get more
                // than one exception, you'll lose all but one.
            }
        }
    }

    private void InvokeAndCancelOthersOnException(Action action,
        CancellationTokenSource cancellationTokenSource,
        ConcurrentQueue<Exception> exceptions)
    {
        // Try to invoke the action. If it throws an exception,
        // capture the exception and then cancel the entire Parallel.Invoke.
        try
        {
            action.Invoke();
        }
        catch (Exception ex)
        {
            exceptions.Enqueue(ex);
            cancellationTokenSource.Cancel();
        }
    }
}

那么用法就是

var thingThatInvokes = new ParallelInvokesMultipleInvocationsAndThrowsOneException();
thingThatInvokes.InvokeActions(
    ()=> GetMaxRateDict(tradeOffObj),
    () => GetMinRateDict(tradeOffObj));

如果它抛出异常，它将是一次调用失败的单个异常，而不是聚合异常。

在并行方法中抛出原始异常而不是聚合异常

throw original exception inside Parallel methods instead of aggregate exception

c#

parallel-processing

task-parallel-library