在计时器上发送 TPL 数据流缓冲区块消息

Question

我想以this 问题为基础。

到目前为止，我已经找到了一种并行处理作业的方法。我在控制台应用程序中运行它。我从 db 得到了 50 个工作，使用TPL DataFlow 处理它们，到目前为止一切都很好。但我意识到，如果有一个工作需要一个小时来处理，而其余的工作在 15 分钟内完成，那么控制台应用程序将继续运行一个小时而不处理任何进一步的工作。我无法将其更改为 Windows 服务，因此我必须让控制台应用程序处理新的工作，可能每 15 分钟检查一次。

我可以启动一个计时器，每 15 分钟检查一次新作业。如果数据库中有任何新工作，我需要添加到buffer block 以便actionblock 可以处理它。问题是，在添加前 50 个作业后，您必须调用 complete 和 completion.wait 以获得缓冲区和操作块。所以我不能再向现有缓冲区添加新作业了。

我可以检查当前的actionblock's isCompleted 属性，然后动态创建buffer/actionblock 的另一个组合。基本上条件是如果当前的actionblock 仍在运行，请检查timer 上的新工作并创建一个新的buffer/actionblock 组合。这就是我打算做的。但在我走上这条路之前，有没有其他方法可以解决这个问题？

Answer 1

如果我理解正确并且您想要的只是要同时执行的作业的恒定“流”，但等待执行的作业不超过 50 个，您可以使用相同的 ActionBlock 和 BoundedCapacity 并添加可以的话：

private static Task ProcessJobsAsync(CancellationToken cancellationToken)
{
    var block = new ActionBlock<Job>(
        job => job.Process(),
        new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = Environment.ProcessorCount, // Or any other value that fits
            BoundedCapacity = 50,
        });
    cancellationToken.Register(block.Complete);
    var producer = Task.Run(async () =>
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            foreach (var job in await GetJobsAsync())
            {
                await block.SendAsync(job,cancellationToken);
            }
        }
    });

    return Task.WhenAll(producer, block.Completion);
}

如果块很慢，并达到其容量wait block.SendAsync(job,cancellationToken); 将异步等待，直到队列中的空间被清除以用于另一个作业。这样你总是有工作要执行。当您想关闭应用程序（或取消操作）时，您可以使用CancellationToken 发出信号