异步等待某些任务完成(Task.WhenSome)

问题描述

我正在编写一个服务，它结合来自各种互联网来源的数据，并在飞翔上生成响应。速度比完整性更重要，因此我希望在部分(不是所有)互联网资源响应后立即生成我的响应。通常，我的服务会创建10个并发Web请求，并且应该在其中5个请求完成后停止等待并开始处理。NET Framework和我所知道的任何第三方库都没有提供此功能，因此我可能不得不自己编写它。我尝试实现的方法具有以下签名：

public static Task<TResult[]> WhenSome<TResult>(int atLeast, params Task<TResult>[] tasks)
{
    // TODO
}

与Task.WhenAny的工作方式相反，如果已经获取了所需数量的结果，则应该接受异常。但是，如果在完成所有任务后，没有足够的收集结果，则应抛出AggregateException以传播所有异常。

使用示例：

var tasks = new Task<int>[]
{
    Task.Delay(100).ContinueWith<int>(_ => throw new ApplicationException("Oops!")),
    Task.Delay(200).ContinueWith(_ => 10),
    Task.Delay(Timeout.Infinite).ContinueWith(_ => 0,
        new CancellationTokenSource(300).Token),
    Task.Delay(400).ContinueWith(_ => 20),
    Task.Delay(500).ContinueWith(_ => 30),
};
var results = await WhenSome(2, tasks);
Console.WriteLine($"Results: {String.Join(", ", results)}");

预期输出：

在本例中，返回值30的最后一个任务应该被忽略(甚至不应该等待)，因为我们已经获得了想要的结果数(2个结果)。出于同样的原因，也应该忽略出错和取消的任务。

推荐答案

这是一些笨重的代码，我认为它满足了您的要求。这可能是一个起点。

这也可能是一个糟糕的处理任务和/或不是ThreadSafe的方式，和/或仅仅是一个糟糕的想法。但我希望如果是这样的话，会有人指出这一点。

async Task<TResult[]> WhenSome<TResult>(int atLeast, List<Task<TResult>> tasks)
{
    List<Task<TResult>> completedTasks = new List<System.Threading.Tasks.Task<TResult>>();
    int completed = 0;
    List<Exception> exceptions = new List<Exception>();

    while (completed < atLeast && tasks.Any()) {
        var completedTask = await Task.WhenAny(tasks);
        tasks.Remove(completedTask);

        if (completedTask.IsCanceled)
        {
            continue;
        }

        if (completedTask.IsFaulted)
        {
            exceptions.Add(completedTask.Exception);
            continue;
        }

        completed++;
        completedTasks.Add(completedTask);
    }

    if (completed >= atLeast)
    {
        return completedTasks.Select(t => t.Result).ToArray();
    }

    throw new AggregateException(exceptions).Flatten();
}

这篇关于异步等待某些任务完成(Task.WhenSome)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！