将 BufferBlock 与消费者的节流选项一起使用答案

【问题标题】：Using BufferBlock with throttling option for consumer将 BufferBlock 与消费者的节流选项一起使用
【发布时间】：2014-10-02 19:08:18
【问题描述】：

在我的生产者-消费者应用程序中，我定义了一个 BufferBlock 作为添加项目的队列。

 public static BufferBlock<AppointmentReminder> m_Queue = new BufferBlock<AppointmentReminder>();
 SemaphoreSlim seaphore = new SemaphoreSlim(4);

然后将项目添加到队列中，我有

    private static void Producer()
    {
        for (int i = 0; i < 5000; i++)
        {
            AppointmentReminder reminder = new AppointmentReminder();
            reminder.UniqueId = Guid.NewGuid();
            reminder.CallMethod = "Number";
            reminder.sString = "1234567890";
            m_Queue.Post(reminder);
        }
        for (int i = 0; i < 3000; i++)
        {
            AppointmentReminder reminder = new AppointmentReminder();
            reminder.UniqueId = Guid.NewGuid();
            reminder.CallMethod = "Letter";
            reminder.sString = "abcdefghij";
            m_Queue.Post(reminder);
        }
        for (int i = 0; i < 2000; i++)
        {
            AppointmentReminder reminder = new AppointmentReminder();
            reminder.UniqueId = Guid.NewGuid();
            reminder.CallMethod = "Mixed";
            reminder.sString = "abcd12345y";
            m_Queue.Post(reminder);
        }
        Console.WriteLine("There are {0} items in the queue.\n", m_Queue.Count);
    }

现在我必须处理消费者部分。有一种方法RunScript(AppointmentReminder callData) 可以做到这一点。这意味着如果项目可用，我们需要在消费者部分调用该方法。但是有一个节流限制。任何时候最多处理4个。

所以我有：

    private async static Task Consumer()
    {
        try
        {
            while (await m_Queue.OutputAvailableAsync())
            {
                AppointmentReminder reminder = m_Queue.Receive();
                Call d = new Call();
                d.RunScript(reminder);
            }

        }
        catch (NullReferenceException ex)
        {
            Console.WriteLine("NullReferenceException: " + ex.Message);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }

对于生产者和消费者来说，

    static void Main(string[] args)
    {
        AliveEvent = new ManualResetEvent(false);
        Producer(); 
        var consumer = Consumer();
        consumer.Wait();        
    }

我的问题是我对Task Parallel Library (TPL).不强，如何将节流约束应用于消费者？

编辑：2014 年 10 月 3 日：

基于 svick 的解决方案。消费者的代码是：

    private async static Task Consumer()
    {
        try
        {
            while (await m_Queue.OutputAvailableAsync())
            {
                var consumerBlock = new ActionBlock<AppointmentReminder>(
remainder => new Call().RunScript(remainder),
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });
                m_Queue.LinkTo(
consumerBlock, new DataflowLinkOptions { PropagateCompletion = true });
                m_Queue.Complete();
                consumerBlock.Completion.Wait();
            }
            // m_Queue is a static BufferBlock in the original code.
        }

【问题讨论】：

您的编辑越来越接近了，但还不够。首先，while (await m_Queue.OutputAvailableAsync()) 现在完全是多余的——你可以去掉它。其次，如果你的消费者是async Task，你不妨await consumerBlock.Completion 而不是用Completion.Wait() 阻塞。第三，m_Queue.Complete() 真正属于生产者（在所有Posts 之后），而不是消费者。
@KirillShlenskiy，我可以删除m_Queue.Complete() 吗？因为在现实世界的情况下，项目将在一个while循环中添加。这意味着永远保持添加项目到队列中并且永不停止。
你当然可以，但在这种情况下你必须特别小心：如果你的生产者总是以比消费者处理它们的速度更快的速度生产物品，那么你就遇到了问题。如果它确实成为一个问题，我也会限制生产者（使用m_Queue = new BufferBlock<AppointmentReminder>(new DataflowBlockOptions { BoundedCapacity = 1000 }) 并使用m_Queue.SendAsync<T>(T) 而不是Post 很容易完成。
@KirillShlenskiy，是的，在我的应用程序中，生产者比消费者快得多。另一个问题是我也可以删除consumerBlock.Completion.Wait(); 吗？我的应用程序是一个电话应用程序，它运行 365x7x24 并且永不停止。
某处必须受到限制。如果消费者不能足够快地处理商品，那么在某个时间点之后将它们排队是没有意义的——你只是在浪费内存。所以是的，你需要有限的容量和SendAsync。在不等待消费者完成方面 - 你是对的：如果没有完成，等待它没有多大意义。事实是，您的 ActionBlock 是您的消费者，因此您可以取消 Consumer() 方法调用，而只需使用 ActionBlock 初始化和 m_Queue.LinkTo 调用。

标签： c# task-parallel-library tpl-dataflow

【解决方案1】：

比自己编写消费者更好的选择是创建一个ActionBlock，它已经支持限制并行：

var consumerBlock = new ActionBlock<AppointmentReminder>(
    remainder => new Call().RunScript(remainder),
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });

然后将其链接到队列中：

queue.LinkTo(
    consumerBlock, new DataflowLinkOptions { PropagateCompletion = true });

最后，等待它完成：

queue.Complete();
consumerBlock.Completion.Wait();

【讨论】：

我想将代码组合在一起。编辑一下，你确认一下好吗？
@Love 不，那是错误的。当你像我建议的那样使用ActionBlock 时，你就不再需要循环了。
只是一个后续问题。看起来代码是针对并行问题的，我也可以将它应用于并发限制问题吗？假设最大并发数为 4。