完成生产者-消费者后继续一个方法

Question

我在 WPF 中有一个生产者-消费者应用程序。在我点击一个按钮之后。

private async void Start_Click(object sender, RoutedEventArgs e)
{
     try
     {
        // set up data
        var producer = Producer();
        var consumer = Consumer();
        await Task.WhenAll(producer, consumer);
        // need log the results in Summary method 
        Summary();
     }
}

summary 方法是无效的；我认为这是正确的。

private void Summary(){}
async Task Producer(){ await something }
async Task Consumer(){ await something }

编辑：

我的问题是在Summary() 方法中，我必须使用任务中的计算值，但是Consumer 任务是一个长期运行的过程。程序运行Summary 很快，即使没有得到更新的值。它使用初始值。

我的想法：

await Task.WhenAll(producer, consumer);
Summary(); 

EDIT2：2014 年 11 月 5 日上午 11:08

private void Summary()
{
     myFail = 100 - mySuccess;
     _dataContext.MyFail = myFail; // update window upon property changed

 async Task Consumer()
 {
     try
     {
         Dictionary<string, string> dict = new Dictionary<string, string>();
         var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions
         {
                MaxDegreeOfParallelism = 5,
                CancellationToken = cToken
         };
         var c = new ActionBlock<T>(
          t=>
         {
              if (cToken.IsCancellationRequested)
                  return;
               dict = Do(t, cToken);
               if(dict["Success"] == "Success")
                   mySuccess++;

目前的问题是mySuccess始终是Summary方法中的初始值。

Answer 1

您可以在生产者和消费者都完成后使用ContinueWith方法执行Summary：

Task.WhenAll(producer, consumer)
    .ContinueWith(continuation => Summary());

编辑 1

您似乎滥用或使用了错误的生产者/消费者模式。

生产者应该产生值并将它们铲到通信管道的一端。在管道的另一端，消费者在值可用时使用它们。换句话说，消费者等待生产者产生一些值并将该值放入管道中并等待该值到达管道末端。

这通常涉及某种信号机制，其中生产者在创建值时向消费者发出信号（唤醒）。

在您的情况下，您没有信号机制，我强烈怀疑您的生产者只产生一个值。如果是后者，您可以只从“生产者”返回一个值。

但是，如果您的生产者创建了多个值，您可以使用 BlockingCollection<T> 类将值从生产者发送到消费者。

在您的Producer 类中，获取对管道的引用并将数据放入其中：

public class Producer
{
    private BlockingCollection<Data> _pipe;

    public void Start()
    {
        while(!done)
        {
            var value = ProduceValue();
            _pipe.Add(value);
        }
        // Signal the consumer that we're finished
        _pipe.CompleteAdding();
    }
}

在Consumer 类中等待值到达并处理每个值：

public class Consumer
{
    private BlockingCollection<Data> _pipe;

    public void Start()
    {
        foreach(var value in _pipe.GetConsumingEnumerable())
        {
            // GetConsumingEnumerable will block until a value arrives and 
            // will exit when producer calls CompleteAdding()
            Process(value);
        }
    }
}

具备上述条件后，您可以在WhenAll 方法上使用ContinueWith 或await 来运行Summary。

编辑 2

按照 cmets 中的承诺，我已经分析了您在 MSDN Forum 上发布的代码。代码有几个问题。

首先，最简单的解决方法是您没有以线程安全的方式递增计数器。递增 (value++) 不是原子操作，因此在递增共享字段时应小心。一个简单的方法是：

Interlocked.Increment(ref evenNumber);

现在，您的代码中的实际问题：

1. 正如我之前提到的，消费者不知道生产者何时完成了值的生成。因此，在生产者退出for 块后，它应该发出信号表明它已经完成。消费者等待生产者的结束信号；否则它将永远等待下一个值，但不会有。

2. 您正在将 BufferBlock 与开始执行的消费者代码链接，但您并没有等待消费者块完成 - 您只等待 0.5 秒并退出消费者方法离开消费者阻塞的工作线程徒劳地做他们的工作。

3. 由于上述原因，您的 Report 方法在处理完成之前执行，在方法执行时而不是在所有处理完成时输出 evenNumber 计数器的值。

下面是一些cmets的编辑代码：

class Program
{
    public static BufferBlock<int> m_Queue = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 1000 });
    private static int evenNumber;

    static void Main(string[] args)
    {
        var producer = Producer();
        var consumer = Consumer();

        Task.WhenAll(producer, consumer).Wait();
        Report();
    }

    static void Report()
    {
        Console.WriteLine("There are {0} even numbers", evenNumber);
        Console.Read();
    }

    static async Task Producer()
    {
        for (int i = 0; i < 500; i++)
        {
            // Send a value to the consumer and wait for the value to be processed
            await m_Queue.SendAsync(i);
        }
        // Signal the consumer that there will be no more values
        m_Queue.Complete();
    }

    static async Task Consumer()
    {
        var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 4
        };
        var consumerBlock = new ActionBlock<int>(x =>
        {
            int j = DoWork(x);
            if (j % 2 == 0) 
                // Increment the counter in a thread-safe way
                Interlocked.Increment(ref evenNumber);
        }, executionDataflowBlockOptions);

        // Link the buffer to the consumer
        using (m_Queue.LinkTo(consumerBlock, new DataflowLinkOptions { PropagateCompletion = true }))
        {
            // Wait for the consumer to finish.
            // This method will exit after all the data from the buffer was processed.
            await consumerBlock.Completion;
        }
    }

    static int DoWork(int x)
    {
        Thread.Sleep(100);
        return x;
    }
}