如何：实现各种制造者-使用者模式

项目
08/12/2011

本主题描述如何在您的应用程序中实现制造者-使用者模式。在此模式中，制造者向消息块发送消息，使用者从该块中读取消息。

本主题演示了两种方案。在第一个方案中，使用者必须接收制造者发送的每条消息。在第二个方案中，使用者定期轮询数据，因此不必接收每条消息。

本主题中的两个示例均使用代理、消息块和消息传递函数将消息从制造者传输给使用者。制造者代理使用 Concurrency::send 函数将消息写入 Concurrency::ITarget 对象。使用者代理使用 Concurrency::receive 函数从 Concurrency::ISource 对象读取消息。两个代理都具有 sentinel 值以协调处理结束。

有关异步代理的更多信息，请参见异步代理。有关消息块和消息传递函数的更多信息，请参见异步消息块和消息传递函数。

示例

在本示例中，制造者代理将一系列数字发送给使用者代理。使用者接收其中每个数字并计算它们的平均值。应用程序将平均值写入控制台。

本示例使用 Concurrency::unbounded_buffer 对象使制造者能够对消息进行排队。 unbounded_buffer 类实现 ITarget 和 ISource，以便制造者和使用者可以将消息发送到共享缓冲区以及从共享缓冲区接收消息。 send 和 receive 函数会对将数据从制造者传播给使用者这一任务进行协调。

// producer-consumer-average.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

// Demonstrates a basic agent that produces values.
class producer_agent : public agent
{
public:
   explicit producer_agent(ITarget<int>& target, unsigned int count, int sentinel)
      : _target(target)
      , _count(count)
      , _sentinel(sentinel)
   {
   }
protected:
   void run()
   {
      // Send the value of each loop iteration to the target buffer.
      while (_count > 0)
      {
         send(_target, static_cast<int>(_count));
         --_count;
      }
      // Send the sentinel value.
      send(_target, _sentinel);

      // Set the agent to the finished state.
      done();
   }
private:
   // The target buffer to write to.
   ITarget<int>& _target;
   // The number of values to send.
   unsigned int _count;
   // The sentinel value, which informs the consumer agent to stop processing.
   int _sentinel;
};

// Demonstrates a basic agent that consumes values.
class consumer_agent : public agent
{
public:
   explicit consumer_agent(ISource<int>& source, int sentinel)
      : _source(source)
      , _sentinel(sentinel)
   {
   }

   // Retrieves the average of all received values.
   int average()
   {
      return receive(_average);
   }
protected:
   void run()
   {
      // The sum of all values.
      int sum = 0;
      // The count of values received.
      int count = 0;

      // Read from the source block until we receive the 
      // sentinel value.
      int n;
      while ((n = receive(_source)) != _sentinel)
      {
         sum += n;
         ++count;
      }

      // Write the average to the message buffer.
      send(_average, sum / count);

      // Set the agent to the finished state.
      done();
   }
private:
   // The source buffer to read from.
   ISource<int>& _source;
   // The sentinel value, which informs the agent to stop processing.
   int _sentinel;
   // Holds the average of all received values.
   single_assignment<int> _average;
};

int wmain()
{
   // Informs the consumer agent to stop processing.
   const int sentinel = 0;
   // The number of values for the producer agent to send.
   const unsigned int count = 100;

   // A message buffer that is shared by the agents.
   unbounded_buffer<int> buffer;

   // Create and start the producer and consumer agents.
   producer_agent producer(buffer, count, sentinel);
   consumer_agent consumer(buffer, sentinel);
   producer.start();
   consumer.start();

   // Wait for the agents to finish.
   agent::wait(&producer);
   agent::wait(&consumer);

   // Print the average.
   wcout << L"The average is " << consumer.average() << L'.' << endl;
}

该示例产生下面的输出。

The average is 50.

在本示例中，制造者代理将一系列股票报价发送给使用者代理。使用者代理定期读取当前报价并将其打印到控制台。

本示例与前一个示例相似，但它使用 Concurrency::overwrite_buffer 对象来使制造者能够与使用者共享一条消息。就像在上一个示例中一样，overwrite_buffer 类实现 ITarget 和 ISource，以便制造者和使用者可以操作共享的消息缓冲区。

// producer-consumer-quotes.cpp
// compile with: /EHsc
#include <agents.h>
#include <array>
#include <algorithm>
#include <iostream>

using namespace Concurrency;
using namespace std;

// Demonstrates a basic agent that produces values.
class producer_agent : public agent
{
public:
   explicit producer_agent(ITarget<double>& target)
      : _target(target)
   {
   }
protected:
   void run()
   {
      // For illustration, create a predefined array of stock quotes. 
      // A real-world application would read these from an external source, 
      // such as a network connection or a database.
      array<double, 6> quotes = { 24.44, 24.65, 24.99, 23.76, 22.30, 25.89 };

      // Send each quote to the target buffer.
      for_each (quotes.begin(), quotes.end(), [&] (double quote) { 

         send(_target, quote);

         // Pause before sending the next quote.
         Concurrency::wait(20);
      });
      // Send a negative value to indicate the end of processing.
      send(_target, -1.0);

      // Set the agent to the finished state.
      done();
   }
private:
   // The target buffer to write to.
   ITarget<double>& _target;
};

// Demonstrates a basic agent that consumes values.
class consumer_agent : public agent
{
public:
   explicit consumer_agent(ISource<double>& source)
      : _source(source)      
   {
   }

protected:
   void run()
   {
      // Read quotes from the source buffer until we receive
      // a negative value.
      double quote;
      while ((quote = receive(_source)) >= 0.0)
      {
         // Print the quote.
         wcout.setf(ios::fixed);
         wcout.precision(2);
         wcout << L"Current quote is " << quote << L'.' << endl;

         // Pause before reading the next quote.
         Concurrency::wait(10);
      }

      // Set the agent to the finished state.
      done();
   }
private:
   // The source buffer to read from.
   ISource<double>& _source;
};

int wmain()
{
   // A message buffer that is shared by the agents.
   overwrite_buffer<double> buffer;

   // Create and start the producer and consumer agents.
   producer_agent producer(buffer);
   consumer_agent consumer(buffer);
   producer.start();
   consumer.start();

   // Wait for the agents to finish.
   agent::wait(&producer);
   agent::wait(&consumer);
}

此示例产生下面的示例输出。

Current quote is 24.44.
Current quote is 24.44.
Current quote is 24.65.
Current quote is 24.99.
Current quote is 23.76.
Current quote is 22.30.
Current quote is 25.89.

与 unbounded_buffer 对象不同，receive 函数不会从 overwrite_buffer 对象中移除消息。如果使用者在制造者覆盖该消息之前多次从消息缓冲区中进行读取，则接收者每次都将获得相同的消息。