我想在并行任务/等待中使用.NET迭代器。类似于这样：
IEnumerable<TDst> Foo<TSrc, TDest>(IEnumerable<TSrc> source)
{
Parallel.ForEach(
source,
s=>
{
// Ordering is NOT important
// items can be yielded as soon as they are done
yield return ExecuteOrDownloadSomething(s);
}
}
很遗憾，.NET不能在本地处理此操作。迄今为止，@svick的最佳答案是使用AsParallel()。
额外奖励：是否有任何简单的async/await代码实现多个发布者和一个订阅者？ 订阅者将会产生，而出版物将会被处理。（仅限核心库）
return source.AsParallel().Select(s => ExecuteOrDownloadSomething(s));
这将使用有限数量的线程并行执行委托，返回每个结果完成后的结果。
如果ExecuteOrDownloadSomething()方法是I/O绑定的（例如它实际上下载了某些内容）并且您不想浪费线程，则使用async-await可能是有意义的，但这会更加复杂。
如果您想充分利用async，则不应返回IEnumerable，因为它是同步的（即如果没有可用项，则会阻塞）。您需要一些异步集合，并且可以使用TPL Dataflow中的ISourceBlock（具体来说，是TransformBlock）：
ISourceBlock<TDst> Foo<TSrc, TDest>(IEnumerable<TSrc> source)
{
var block = new TransformBlock<TSrc, TDest>(
async s => await ExecuteOrDownloadSomethingAsync(s),
new ExecutionDataflowBlockOptions
{
MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
});
foreach (var item in source)
block.Post(item);
block.Complete();
return block;
}
如果源代码“缓慢”（即在迭代source完成之前您想要开始处理来自Foo()的结果），您可能希望将foreach和Complete()调用移动到单独的Task中。更好的解决方案是将source也转换为ISourceBlock<TSrc>。
public static IEnumerable<Task<T>> Order<T>(this IEnumerable<Task<T>> tasks)
{
var input = tasks.ToList();
var output = input.Select(task => new TaskCompletionSource<T>());
var collection = new BlockingCollection<TaskCompletionSource<T>>();
foreach (var tcs in output)
collection.Add(tcs);
foreach (var task in input)
{
task.ContinueWith(t =>
{
var tcs = collection.Take();
switch (task.Status)
{
case TaskStatus.Canceled:
tcs.TrySetCanceled();
break;
case TaskStatus.Faulted:
tcs.TrySetException(task.Exception.InnerExceptions);
break;
case TaskStatus.RanToCompletion:
tcs.TrySetResult(task.Result);
break;
}
}
, CancellationToken.None
, TaskContinuationOptions.ExecuteSynchronously
, TaskScheduler.Default);
}
return output.Select(tcs => tcs.Task);
}
var tasks = collection.Select(item => LongRunningOperationThatReturnsTask(item))
.Order();
foreach(var task in tasks)
{
var result = task.Result;//or you could `await` each result
//....
}
WhenAll/WaitAll，但除此之外，我无法看出 Select 本身不能满足您的需求。您有一系列项目，并且想将其转换为任务序列，每个项目对应一个任务。Select(item=> LongRunningOperation(item)) 如何不符合您的需求，因为它返回一个任务序列？ - Servy