9

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

奖励：任何实现多个发布者和单个订阅者的简单异步/等待代码？订阅者会屈服，而 pubs 会处理。（仅限核心库）

4

3 回答 3

11

这似乎是 PLINQ 的工作：

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

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

如果该ExecuteOrDownloadSomething()方法是 IO 绑定的（例如，它实际上下载了一些东西）并且您不想浪费线程，那么使用async-await可能是有意义的，但它会更复杂。

如果您想充分利用async，则不应 return IEnumerable，因为它是同步的（即，如果没有可用的项目，它会阻塞）。您需要的是某种异步收集，您可以使用ISourceBlock（特别是TransformBlock）来自 TPL Dataflow 的：

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完成之前开始处理结果），您可能需要将foreachandComplete()调用移动到单独的Task. 更好的解决方案是制作source成一个ISourceBlock<TSrc>。

于 2013-02-11T10:11:34.097 回答
1

因此，您真正想做的似乎是根据任务的完成时间对一系列任务进行排序。这并不是非常复杂：

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

所以在这里我们为每个输入任务创建一个TaskCompletionSource，然后遍历每个任务并设置一个延续，它从 a 中获取下一个完成源BlockingCollection并设置它的结果。完成的第一个任务获取返回的第一个 tcs，完成的第二个任务获取返回的第二个 tcs，依此类推。

现在您的代码变得非常简单：

var tasks = collection.Select(item => LongRunningOperationThatReturnsTask(item))
    .Order();
foreach(var task in tasks)
{
    var result = task.Result;//or you could `await` each result
    //....
}
于 2013-02-11T15:16:25.253 回答
0

在 MS 机器人团队制作的异步库中，它们具有允许使用迭代器生成异步代码的并发原语。

图书馆 (CCR) 是免费的（它以前不是免费的）。可以在这里找到一篇不错的介绍性文章：并发事务

也许您可以将这个库与 .Net 任务库一起使用，或者它会激励您“自己动手”

于 2013-02-11T08:52:01.833 回答