一次阻止收集过程 n 个项目 - 一旦完成 1 就继续

Question

我有以下场景。

1. 我将数据库中的 50 个作业放入阻塞集合中。

2. 每个作业都是长期运行的。 （可能是）。所以我想在一个单独的线程中运行它们。 （我知道 - 最好将它们作为 Task.WhenAll 运行并让 TPL 解决 - 但我想控制同时运行的数量）

3. 假设我想同时运行 5 个（可配置）

4. 我创建了 5 个任务 (TPL)，每个任务一个并并行运行它们。

我想要做的是在第 4 步中的一个作业完成后立即在阻塞集合中选择下一个作业，并继续进行，直到完成所有 50 个作业。

我正在考虑创建一个静态blockingCollection 和一个TaskCompletionSource，它们将在作业完成时被调用，然后它可以再次调用消费者从队列中一次选择一个作业。我还想在每个作业上调用 async/await - 但除此之外 - 不确定这是否会对方法产生影响。

这是完成我想做的事情的正确方法吗？

类似于this 链接，但要注意的是，我想在前 N 项中的一项完成后立即处理下一个作业。不是在所有 N 都完成后。

更新：

好的，我有这个代码 sn-p 做我想要的，如果有人想稍后使用它。正如您在下面看到的，创建了 5 个线程，每个线程在完成当前作业后开始下一个作业。在任何给定时间只有 5 个线程处于活动状态。我知道这可能无法始终像这样 100% 工作，并且如果与一个 cpu/核心一起使用，则会出现上下文切换的性能问题。

var block = new ActionBlock<Job>(
                job => Handler.HandleJob(job), 
                    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5 });

              foreach (Job j in GetJobs())
                  block.SendAsync(j);

作业 2 开始于线程 :13。等待时间：3600000ms。时间：8/29/2014 下午 3 点 14 分 43 秒

作业 4 开始于线程 :14。等待时间：15000ms。时间：8/29/2014 下午 3 点 14 分 43 秒

作业 0 开始于线程 :7。等待时间：600000 毫秒。时间：8/29/2014 下午 3 点 14 分 43 秒

作业 1 开始于线程 :12。等待时间：900000 毫秒。时间：8/29/2014 下午 3 点 14 分 43 秒

作业 3 开始于线程 :11。等待时间：120000ms。时间：8/29/2014 下午 3 点 14 分 43 秒

作业 4 在线程 :14 上完成。 2014 年 8 月 29 日下午 3:14:58

作业 5 开始于线程 :14。等待时间：1800000ms。时间：8/29/2014 下午 3 点 14 分 58 秒

作业 3 在线程 :11 上完成。 2014 年 8 月 29 日下午 3:16:43

作业 6 开始于线程 :11。等待时间：1200000ms。时间：8/29/2014 下午 3:16:43

作业 0 在线程 :7 上完成。 2014 年 8 月 29 日下午 3:24:43

作业 7 开始于线程 :7。等待时间：30000ms。时间：8/29/2014 3:24:43 下午

作业 7 在线程 7 上完成。 2014 年 8 月 29 日下午 3:25:13

作业 8 开始于线程 :7。等待时间：100000 毫秒。时间：8/29/2014 下午 3:25:13

作业 8 在线程 7 上完成。 2014 年 8 月 29 日下午 3:26:53

作业 9 开始于线程 :7。等待时间：900000 毫秒。时间：8/29/2014 下午 3 点 26 分 53 秒

作业 1 在线程 :12 上完成。 2014 年 8 月 29 日下午 3:29:43

作业 10 开始于线程 :12。等待时间：300000 毫秒。时间：8/29/2014 下午 3 点 29 分 43 秒

作业 10 在线程 :12 上完成。 2014 年 8 月 29 日下午 3:34:43

作业 11 开始于线程 :12。等待时间：600000 毫秒。时间：8/29/2014 下午 3:34:43

作业 6 在线程 :11 上完成。 2014 年 8 月 29 日下午 3:36:43

作业 12 开始于线程 :11。等待时间：300000 毫秒。时间：8/29/2014 下午 3 点 36 分 43 秒

作业 12 在线程 :11 上完成。 2014 年 8 月 29 日下午 3:41:43

作业 13 开始于线程 :11。等待时间：100000 毫秒。时间：8/29/2014 下午 3 点 41 分 43 秒

作业 9 在线程 7 上完成。 2014 年 8 月 29 日下午 3:41:53

作业 14 开始于线程 :7。等待时间：300000 毫秒。时间：8/29/2014 下午 3 点 41 分 53 秒

作业 13 在线程 :11 上完成。 2014 年 8 月 29 日下午 3:43:23

作业 11 在线程 :12 上完成。 2014 年 8 月 29 日下午 3:44:43

作业 5 在线程 :14 上完成。 2014 年 8 月 29 日下午 3:44:58

作业 14 在线程 :7 上完成。 2014 年 8 月 29 日下午 3:46:53

作业 2 在线程 :13 上完成。 2014 年 8 月 29 日下午 4:14:43

Answer 1

您可以使用TPL Dataflow 轻松实现您的需求。

您可以做的是使用BufferBlock<T>，这是一个用于存储数据的缓冲区，并将其与ActionBlock<T> 链接在一起，当这些请求来自BufferBlock<T> 时，它将使用这些请求。

现在，这里的美妙之处在于您可以使用ExecutionDataflowBlockOptions 类指定您希望ActionBlock<T> 同时处理多少个请求。

这是一个简化的控制台版本，当他们进入时处理一堆数字，打印他们的名字和Thread.ManagedThreadID：

private static void Main(string[] args)
{
    var bufferBlock = new BufferBlock<int>();

    var actionBlock =
        new ActionBlock<int>(i => Console.WriteLine("Reading number {0} in thread {1}",
                                  i, Thread.CurrentThread.ManagedThreadId),
                             new ExecutionDataflowBlockOptions 
                                 {MaxDegreeOfParallelism = 5});

    bufferBlock.LinkTo(actionBlock);
    Produce(bufferBlock);

    Console.ReadKey();
}

private static void Produce(BufferBlock<int> bufferBlock)
{
    foreach (var num in Enumerable.Range(0, 500))
    {
        bufferBlock.Post(num);
    }
}

如果需要，您还可以使用可等待的 BufferBlock.SendAsync 异步发布它们

这样，您可以让TPL 为您处理所有限制，而无需手动执行。

Answer 2

您可以使用BlockingCollection，它可以正常工作，但它是在async-await 之前构建的，因此它会同步阻塞，在大多数情况下可能不太可扩展。

正如 Yuval Itzchakov 建议的那样，你最好使用 async 准备好的 TPL Dataflow。您所需要的只是一个ActionBlock，它可以同时处理每个项目，MaxDegreeOfParallelism 为 5，然后您将工作同步 (block.Post(item)) 或异步 (await block.SendAsync(item)) 发布到它：

private static void Main()
{
    var block = new ActionBlock<Job>(
        async job => await job.ProcessAsync(),
        new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = 5});

    for (var i = 0; i < 50; i++)
    {
        block.Post(new Job());
    }

    Console.ReadKey();
}

Answer 3

您可以使用SemaphoreSlim （如this answer）或使用ForEachAsync （如this answer）来完成此操作。