Task.WhenAll 与 BlockingCollection 产生的无限任务的使用答案

【问题标题】：Usage of Task.WhenAll with infinite Tasks produced by BlockingCollectionTask.WhenAll 与 BlockingCollection 产生的无限任务的使用
【发布时间】：2014-08-26 21:22:20
【问题描述】：

我正在将后台任务添加到阻塞集合（在后台添加）。

我正在等待 GetConsumingEnumerable 返回的 Enumerable 上的 Task.WhenAll。

我的问题是：接收 IEnumerable 的 Task.WhenAll 的重载是否“准备好”接收无穷无尽的任务？

我只是不确定我是否可以这样做，或者它是否应该以这种方式使用？

private async Task RunAsync(TimeSpan delay, CancellationToken cancellationToken)
{
    using (BlockingCollection<Task> jobcollection = new BlockingCollection<Task>())
    {
        Task addingTask = Task.Run(async () =>
        {
            while (true)
            {
                DateTime utcNow = DateTime.UtcNow;
                var jobs = Repository.GetAllJobs();
                foreach (var job in GetRootJobsDue(jobs, utcNow))
                {
                    jobcollection.Add(Task.Run(() => RunJob(job, jobs, cancellationToken, utcNow), cancellationToken), cancellationToken);
                }

                await Task.Delay(delay, cancellationToken);
            }
        }, cancellationToken);

        await Task.WhenAll(jobcollection.GetConsumingEnumerable(cancellationToken));
    }
}

【问题讨论】：

似乎使用Parallel.Foreach(Repository.GetAllJobs,....) 会是一个更简单的解决方案。
Task.WhenAll 将阻塞调用线程直到GetConsumingEnumerable 完成
好吧，如果有无数的任务，他们永远不会全部完成，因为你永远不会得到整套。
@SriramSakthivel GetConsumingEnumerable 立即返回一个 IEnumerable 但仅在数据可用时才进行迭代......预期的行为
@rudimenter 在你回复之前我编辑了我的评论。看看这是否有意义？

标签： c# task async-await blockingcollection

【解决方案1】：

Since your goal is merely to wait until the cancellation token is cancelled，你应该这样做那个。由于其他人已经解释过的原因，在无限的任务序列上使用WhenAll 并不是解决这个问题的方法。有更简单的方法可以完成永远不会完成的任务。

await new TaskCompletionSource<bool>().Task
    .ContinueWith(t => { }, cancellationToken);

【讨论】：

看来他的目标是等到所有任务都完成后，而不是等到所有任务都创建完毕。
它是如何工作的？ TaskCompletionSource<bool>()Task 永远不会完成，这意味着 ContinueWith 永远不会执行。我假设您的意思是在任务上调用SetCancelled？
@KyleW No, it's not.
@SriramSakthivel 不，这将按原样工作。当令牌被取消时，它将停止等待永无止境的任务，这是一个继续完成的任务。你的方法会奏效，只是更多的代码行。
我明白了，这很漂亮。我忽略了它.. +1

【解决方案2】：

Task.WhenAll 无法处理无限数量的任务。它将首先（同步）等待枚举完成，然后（异步）等待它们全部完成。

如果您想以异步方式对序列做出反应，则需要使用IObservable<Task>（反应式扩展）。您可以将 TPL 数据流 BufferBlock 用作“队列”，可以使用同步或异步代码，并且可以轻松转换为 IObservable<Task>。

【讨论】：

是否同样适用于 Parallel.Foreach。还会等到枚举完成吗？
@rudimenter Parallel.Foreach 本质上是同步操作，而不是异步操作。它不仅会同步等待序列完成，还会同步等待完成所有项目的处理。
我很确定Parallel 不会在开始处理之前立即等待枚举，但是您将“丢失”一个线程来进行枚举。

【解决方案3】：

我假设Task.WhenAll 将尝试枚举集合，这意味着它本身会阻塞，直到集合完成或取消。如果没有，那么理论上代码可以在创建任务之前完成awaiting。所以那里会有一个额外的块......它将阻塞等待线程被创建，然后再次阻塞，直到任务完成。我认为这对您的代码来说并不是一件坏事，因为它仍然会一直阻塞到同一时间点。

【讨论】：