我捕获的AggregateException没有我期望的异常

Question

我正在使用任务并行库来设置一个任务链,如下所示,但我得到一个奇怪的异常处理经验,我不明白.

我使用Parallel.ForEach并调用一个包含对以下方法的调用的Action.这个Parallel.ForEach包含在try ... catch(AggregateException)中,当发生异常时 - 就像在其中一个Parallel分支中那样 - 一个SchemaValidation异常,然后我希望在AggregateException中看到它.

但是,我得到的是"任务被取消" - TaskCanceledException.我的SchemaValidationException去了哪里？

        private static void ProcessChunk(Task<ISelectedChunk> selectionTask, 
                                     IRepository repository, 
                                     IIdentifiedExtractChunk identifiedExtractChunk, 
                                     IBatchRunConfiguration batchRunConfiguration, 
                                     IBatchRun batchRun, 
                                     ILog log, 
                                     IAuthenticationCertificate authenticationCertificate, 
                                     IFileSystem fileSystem,
                                     long batchRunRid)
    {
        var transformationTask = selectionTask.ContinueWith(TransformationFunction.Transformation(identifiedExtractChunk, batchRunConfiguration, batchRun),
                                                            TaskContinuationOptions.NotOnFaulted);

        var schemaValidationTask = transformationTask.ContinueWith(SchemaValidationFunction.SchemaValidationTask(batchRunConfiguration),
                                                                   TaskContinuationOptions.NotOnFaulted);

        var compressTask = schemaValidationTask.ContinueWith(CompressFunction.CompressTask(identifiedExtractChunk),
                                                             TaskContinuationOptions.NotOnFaulted);

        var encryptTask = compressTask.ContinueWith(EncryptionFunction.EncryptTask(authenticationCertificate),
                                                    TaskContinuationOptions.NotOnFaulted);

        var fileGenerationTask = encryptTask.ContinueWith(FileGenerationFunction.FileGenerationTask(identifiedExtractChunk, batchRunConfiguration, fileSystem),
                                                          TaskContinuationOptions.NotOnFaulted);
        // Take the time before we start the processing
        DateTime startBatchItemProcessing = DateTime.Now;

        // Start with the Selection Task
        selectionTask.Start();

        // And wait on the last task in the chain
        fileGenerationTask.Wait();

        // Take the time at the end of the processing
        DateTime endBatchItemProcessing = DateTime.Now;

        // Record all the relevant information and add it to the collection 
        IBatchChunkProcessed batchChunkProcessed = GetBatchItemProcessed(identifiedExtractChunk, batchRunRid, fileGenerationTask.Result, transformationTask.Result.Item2, startBatchItemProcessing, endBatchItemProcessing);
        BatchItemsProcessed.Add(batchChunkProcessed);

Answer 1

让我们简化一下你的代码:

var t1 = Task.Factory.StartNew(a1);
var t2 = t1.ContinueWith(a2, TaskContinuationOptions.NotOnFaulted);
var t3 = t2.ContinueWith(a3, TaskContinuationOptions.NotOnFaulted);

t3.Wait();

现在假设a1抛出一个异常.会发生什么事情会导致t1出现故障(t1.Status == TaskStatus.Faulted).因此,t2无法运行(因为NotOnFaulted)所以它将被取消.但这不是你想象的那样:t2不会出现故障,它会被取消(t2.Status == TaskStatus.Canceled).但这意味着t3可以正常运行,如果它不抛出,t3.Wait()则不会抛出任何异常.

如何解决这个问题？首先,您可能不应该使用TaskContinuationOptions.NotOnFaulted,而是使用TaskContinuationOptions.OnlyOnRanToCompletion.但这并没有解决"消失"异常的问题.为了解决这个问题,我看到两种可能性:

1. Wait()在每个延续的开始时调用,不要使用任何TaskContinuationOptions.这意味着你可能会得到一些包含在AggregateException其中的异常,它本身就包含在内AggregateException,包含在另一个中AggregateException,等等.为了解决这个问题,你可以使用Flatten()或Handle().

2. 等待所有任务,使用Task.WaitAll().WaitAll()将抛出一个AggregateException包含原始异常的内容,以及TaskCanceledException因第一个异常而被取消的每个任务.