我试图写一些代码，这将使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 ...
}
编辑 ：快速澄清，以防万一有任何上述困惑。 我试图做到以下几点：
- 对于每个服务器，启动一个
Task来检查它 - 要么，等待服务器返回true（只有1个服务器的最大永远不会返回true）
- 或者，等到所有的服务器都返回false，即存在不匹配。
最好的是我能想到的是指定ContinueWith每个Task ，检查的结果，如果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秒我的机器上（这意味着它没有等到第四届服务器）和我得到了以下的输出：
server1 finished
server2 finished
server3 finished
server4 Cancelled
Got result from server3
您可以使用OrderByCompletion()从AsyncEx库 ，它返回的任务，因为他们完成了。 你的代码可能看起来是这样的：
var tasks = servers
.Select(s => Task.Factory.StartNew(server => CallServer((string)server), s))
.OrderByCompletion();
foreach (var task in tasks)
{
if (task.Result)
{
Console.WriteLine("found");
break;
}
Console.WriteLine("not found yet");
}
// cancel any outstanding tasks since the correct server has been found
使用Interlocked.CompareExchange将做到这一点，只有一个任务就能在serverReturedData写
public void SearchServers()
{
ResultClass serverReturnedData = null;
var servers = new[] {"server1", "server2", "server3", "server4"};
var tasks = servers.Select(s => Task<bool>.Factory.StartNew(server =>
{
var result = CallServer((string)server), s);
Interlocked.CompareExchange(ref serverReturnedData, result, null);
}).ToArray();
Task.WaitAny(tasks); //how do I say "WaitAny where the result is true"?
//
// use serverReturnedData as you want.
//
}
编辑：由于亚什德说，上面的代码可以返回变量serverReturnedData有一个有效的前值（如果服务器返回一个空值，这可能发生），以确保您可以在自定义对象包装的结果。
下面是基于svick的答案一个通用的解决方案：
public static async Task<T> GetFirstResult<T>(
this IEnumerable<Func<CancellationToken, Task<T>>> taskFactories,
Action<Exception> exceptionHandler,
Predicate<T> predicate)
{
T ret = default(T);
var cts = new CancellationTokenSource();
var proxified = taskFactories.Select(tf => tf(cts.Token)).ProxifyByCompletion();
int i;
for (i = 0; i < proxified.Length; i++)
{
try
{
ret = await proxified[i].ConfigureAwait(false);
}
catch (Exception e)
{
exceptionHandler(e);
continue;
}
if (predicate(ret))
{
break;
}
}
if (i == proxified.Length)
{
throw new InvalidOperationException("No task returned the expected value");
}
cts.Cancel(); //we have our value, so we can cancel the rest of the tasks
for (int j = i+1; j < proxified.Length; j++)
{
//observe remaining tasks to prevent process crash
proxified[j].ContinueWith(
t => exceptionHandler(t.Exception), TaskContinuationOptions.OnlyOnFaulted)
.Forget();
}
return ret;
}
凡ProxifyByCompletion被实现为：
public static Task<T>[] ProxifyByCompletion<T>(this IEnumerable<Task<T>> tasks)
{
var inputTasks = tasks.ToArray();
var buckets = new TaskCompletionSource<T>[inputTasks.Length];
var results = new Task<T>[inputTasks.Length];
for (int i = 0; i < buckets.Length; i++)
{
buckets[i] = new TaskCompletionSource<T>();
results[i] = buckets[i].Task;
}
int nextTaskIndex = -1;
foreach (var inputTask in inputTasks)
{
inputTask.ContinueWith(completed =>
{
var bucket = buckets[Interlocked.Increment(ref nextTaskIndex)];
if (completed.IsFaulted)
{
Trace.Assert(completed.Exception != null);
bucket.TrySetException(completed.Exception.InnerExceptions);
}
else if (completed.IsCanceled)
{
bucket.TrySetCanceled();
}
else
{
bucket.TrySetResult(completed.Result);
}
}, CancellationToken.None,
TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
}
return results;
}
而Forget是抑制CS4014空的方法：
public static void Forget(this Task task) //suppress CS4014
{
}
