c# - 如何从 .NET 4.5 中的并行任务中产生

Question

我想将 .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()。

奖励:任何实现多个发布者和单个订阅者的简单异步/等待代码？订户会屈服，而酒吧会处理。 (仅限核心库)

Answer 1

这似乎是 PLINQ 的工作:

return source.AsParallel().Select(s => ExecuteOrDownloadSomething(s));

这将使用有限数量的线程并行执行委托(delegate)，并在完成后立即返回每个结果。

如果ExecuteOrDownloadSomething()方法是 IO 绑定(bind)的(例如它实际上下载了一些东西)并且你不想浪费线程，然后使用 async - await可能有道理，但会更复杂。

如果你想充分利用async ，你不应该返回 IEnumerable ，因为它是同步的(即，如果没有可用项目，它会阻塞)。你需要的是某种异步收集，你可以使用 ISourceBlock (特别是 TransformBlock )来自 TPL 数据流:

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

如果源“慢”(即您希望在迭代 Foo() 完成之前开始处理来自 source 的结果)，您可能需要移动 foreach和 Complete()调用一个单独的 Task .更好的解决方案是制作 source进入ISourceBlock<TSrc>也是。