如何从异步方法返回AggregateException [英] How to return AggregateException from async method

问题描述

我有一个异步方法，就像增强型Task.WhenAll一样工作.它需要执行一系列任务，并在所有任务完成后返回.

I got an async method working like an enhanced Task.WhenAll. It takes a bunch of tasks and returns when all are completed.

public async Task MyWhenAll(Task[] tasks) {
    ...
    await Something();
    ...

    // all tasks are completed
    if (someTasksFailed)
        throw ??
}

我的问题是，当一个或多个任务失败时，如何获得一种类似于Task.WhenAll返回的任务的方法?

My question is how do I get the method to return a Task looking like the one returned from Task.WhenAll when one or more tasks has failed?

如果我收集异常并抛出AggregateException，它将被包装在另一个AggregateException中.

If I collect the exceptions and throw an AggregateException it will be wrapped in another AggregateException.

完整示例

async Task Main() {
    try {
        Task.WhenAll(Throw(1), Throw(2)).Wait();
    }
    catch (Exception ex) {
        ex.Dump();
    }

    try {
        MyWhenAll(Throw(1), Throw(2)).Wait();
    }
    catch (Exception ex) {
        ex.Dump();
    }
}

public async Task MyWhenAll(Task t1, Task t2) {
    await Task.Delay(TimeSpan.FromMilliseconds(100));
    try {
        await Task.WhenAll(t1, t2);
    }
    catch {
        throw new AggregateException(new[] { t1.Exception, t2.Exception });
    }
}
public async Task Throw(int id) {
    await Task.Delay(TimeSpan.FromMilliseconds(100));
    throw new InvalidOperationException("Inner" + id);
}

对于Task.WhenAll，例外是AggregateException，其中有两个内部例外.

For Task.WhenAll the exception is AggregateException with 2 inner exceptions.

对于MyWhenAll，例外是AggregateException，其中一个内部AggregateException具有2个内部例外.

For MyWhenAll the exception is AggregateException with one inner AggregateException with 2 inner exceptions.

我为什么要这样做

我经常需要调用页面调度API:s，并希望限制同时连接的数量.

I often need to call paging API:s and want to limit number of simultaneous connections.

实际的方法签名是

public static async Task<TResult[]> AsParallelAsync<TResult>(this IEnumerable<Task<TResult>> source, int maxParallel)
public static async Task<TResult[]> AsParallelUntilAsync<TResult>(this IEnumerable<Task<TResult>> source, int maxParallel, Func<Task<TResult>, bool> predicate)

这意味着我可以像这样进行分页

It means I can do paging like this

var pagedRecords = await Enumerable.Range(1, int.MaxValue)
                                   .Select(x => GetRecordsAsync(pageSize: 1000, pageNumber: x)
                                   .AsParallelUntilAsync(maxParallel: 5, x => x.Result.Count < 1000);
var records = pagedRecords.SelectMany(x => x).ToList();

一切正常，汇总中的汇总只是一个小麻烦.

It all works fine, the aggregate within aggregate is just a minor inconvenience.

推荐答案

async方法被设计为仅对每个集合最多对返回的任务执行单个异常，而不是多个.

async methods are designed to only every set at most a single exception on the returned task, not multiple.

这为您提供了两个选择，您不能使用async方法作为开始，而是依靠其他方法来执行您的方法:

This leaves you with two options, you can either not use an async method to start with, instead relying on other means of performing your method:

public Task MyWhenAll(Task t1, Task t2)
{
    return Task.Delay(TimeSpan.FromMilliseconds(100))
        .ContinueWith(_ => Task.WhenAll(t1, t2))
        .Unwrap();
}

如果您有一个更复杂的方法而又不使用await则很难编写，那么您将需要解开嵌套的聚合异常，这样做虽然很繁琐，但也不太繁琐:

If you have a more complex method that would be harder to write without using await, then you'll need to unwrap the nested aggregate exceptions, which is tedious, although not overly complex, to do:

    public static Task UnwrapAggregateException(this Task taskToUnwrap)
    {
        var tcs = new TaskCompletionSource<bool>();

        taskToUnwrap.ContinueWith(task =>
        {
            if (task.IsCanceled)
                tcs.SetCanceled();
            else if (task.IsFaulted)
            {
                if (task.Exception is AggregateException aggregateException)
                    tcs.SetException(Flatten(aggregateException));
                else
                    tcs.SetException(task.Exception);
            }
            else //successful
                tcs.SetResult(true);
        });

        IEnumerable<Exception> Flatten(AggregateException exception)
        {
            var stack = new Stack<AggregateException>();
            stack.Push(exception);
            while (stack.Any())
            {
                var next = stack.Pop();
                foreach (Exception inner in next.InnerExceptions)
                {
                    if (inner is AggregateException innerAggregate)
                        stack.Push(innerAggregate);
                    else
                        yield return inner;
                }
            }
        }

        return tcs.Task;
    }

这篇关于如何从异步方法返回AggregateException的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！