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

Question

我正在编写一个服务，它结合了来自各种互联网来源的数据，并即时生成响应。速度比完整性更重要，所以我想在一些（不是所有）互联网资源响应后立即生成我的响应。通常，我的服务会创建 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)}");

预期输出：

结果：1​​0、20

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

Answer 1

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

这也可能是一种处理任务的糟糕方式和/或不是线程安全的，和/或只是一个糟糕的想法。但我希望有人会指出这一点。

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();
}

Answer 2

我正在为这个问题添加一个更多的解决方案，不是因为stuartd's answer 不够，而是为了多样化。此实现使用Unwrap technique 来返回包含所有异常的Task，与内置Task.WhenAll 方法传播所有异常的方式完全相同。

public static Task<TResult[]> WhenSome<TResult>(int atLeast, params Task<TResult>[] tasks)
{
    if (tasks == null) throw new ArgumentNullException(nameof(tasks));
    if (atLeast < 1 || atLeast > tasks.Length)
        throw new ArgumentOutOfRangeException(nameof(atLeast));

    var cts = new CancellationTokenSource();
    int successfulCount = 0;
    var continuationAction = new Action<Task<TResult>>(task =>
    {
        if (task.IsCompletedSuccessfully)
            if (Interlocked.Increment(ref successfulCount) == atLeast) cts.Cancel();
    });
    var continuations = tasks.Select(task => task.ContinueWith(continuationAction,
        cts.Token, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default));

    return Task.WhenAll(continuations).ContinueWith(_ =>
    {
        cts.Dispose();
        if (successfulCount >= atLeast) // Success
            return Task.WhenAll(tasks.Where(task => task.IsCompletedSuccessfully));
        else
            return Task.WhenAll(tasks); // Failure
    }, TaskScheduler.Default).Unwrap();
}

continuations 不会传播tasks 的结果或异常。这些是可取消的延续，当达到指定数量的成功任务时，它们会被集体取消。

注意：此实现可能传播的结果可能不止atLeast。如果您想要准确的结果数量，您可以在.Where LINQ 操作符之后链接.Take(atLeast)。