c# - 具有指定结果的任务并行库 WaitAny

Question

我正在尝试编写一些代码来并行调用多个不同服务器的 Web 服务，因此 TPL 似乎是显而易见的选择。

我的 Web 服务调用中只有一个会返回我想要的结果，而其他所有调用都不会。我正在尝试找到一种有效地拥有 Task.WaitAny 的方法，但只有在第一个匹配条件的 Task 返回时才解除阻塞。

我尝试使用 WaitAny 但无法确定过滤器的放置位置。我做到了这一点:

public void SearchServers()
{
    var servers = new[] {"server1", "server2", "server3", "server4"};
    var tasks = servers
                 .Select(s => Task<bool>.Factory.StartNew(server => CallServer((string)server), s))
                 .ToArray();

    Task.WaitAny(tasks); //how do I say "WaitAny where the result is true"?

    //Omitted: cancel any outstanding tasks since the correct server has been found
}

private bool CallServer(string server)
{
    //... make the call to the server and return the result ...
}

编辑:快速澄清以防万一上面有任何混淆。我正在尝试执行以下操作:

1. 对于每个服务器，启动一个Task来检查它
2. 要么，等待一个服务器返回 true(最多只有 1 个服务器返回 true)
3. 或者，等到所有服务器都返回 false，即没有匹配项。

Answer 1

我能想到的最好的方法是为每个 Task 指定一个 ContinueWith，检查结果，如果 true 则取消另一个任务。要取消任务，您可能需要使用 CancellationToken .

var tasks = servers
    .Select(s => Task.Run(...)
        .ContinueWith(t =>
            if (t.Result) {
                // cancel other threads
            }
        )
    ).ToArray();

更新:另一种解决方案是 WaitAny 直到正确的任务完成(但它有一些缺点，例如从列表中删除已完成的任务并从剩余的任务中创建一个新数组是相当繁重的操作):

List<Task<bool>> tasks = servers.Select(s => Task<bool>.Factory.StartNew(server => CallServer((string)server), s)).ToList();

bool result;
do {
    int idx = Task.WaitAny(tasks.ToArray());
    result = tasks[idx].Result;
    tasks.RemoveAt(idx);
} while (!result && tasks.Count > 0);

// cancel other tasks

更新 2:现在我会用 Rx 来做:

[Fact]
public async Task AwaitFirst()
{
    var servers = new[] { "server1", "server2", "server3", "server4" };
    var server = await servers
        .Select(s => Observable
            .FromAsync(ct => CallServer(s, ct))
            .Where(p => p)
            .Select(_ => s)
        )
        .Merge()
        .FirstAsync();
    output.WriteLine($"Got result from {server}");
}

private async Task<bool> CallServer(string server, CancellationToken ct)
{
    try
    {
        if (server == "server1")
        {
            await Task.Delay(TimeSpan.FromSeconds(1), ct);
            output.WriteLine($"{server} finished");
            return false;
        }
        if (server == "server2")
        {
            await Task.Delay(TimeSpan.FromSeconds(2), ct);
            output.WriteLine($"{server} finished");
            return false;
        }
        if (server == "server3")
        {
            await Task.Delay(TimeSpan.FromSeconds(3), ct);
            output.WriteLine($"{server} finished");
            return true;
        }
        if (server == "server4")
        {
            await Task.Delay(TimeSpan.FromSeconds(4), ct);
            output.WriteLine($"{server} finished");
            return true;
        }
    }
    catch(OperationCanceledException)
    {
        output.WriteLine($"{server} Cancelled");
        throw;
    }

    throw new ArgumentOutOfRangeException(nameof(server));
}

测试在我的机器上花费了 3.32 秒(这意味着它没有等待第 4 个服务器)并且我得到了以下输出:

server1 finished
server2 finished
server3 finished
server4 Cancelled
Got result from server3