Имплементация Producer/Consumer pattern

Паттерн producer/consumer достаточно часто встречается в многопоточном программировании
Его смысл состоит в том, что один или несколько потоков производят данные, и параллельно этому один или несколько потоков потребляют их.

Как правильно имплементировать этот паттерн в популярных языках программирования
Задача сама по себе нетривиальна, поскольку включает синхронизацию между потоками, и потенциальную гонку между несколькими производителями и потребителями.



Справка.

Производящий поток (или потоки) называется «производитель», «поставщик» или просто «producer», потребляющий (-ие) — «потребитель» или «consumer».

Нетривиальность проблемы заключается в том, что потенциально как создание новых данных
так и их потребление могут занимать длительное время, и хотелось бы, чтобы обработка шла без простоев, на максимально возможной скорости.

Примеры:

Произведённые данные могут представлять вычислительно интенсивное задание. В это
случае разумно иметь единственный производящий поток, и несколько выполняющих потоков (например, столько, сколько в системе ядер процессора, если узкое место обработки — вычисления).

Или производящие потоки загружают данные из сети, а по окончанию загрузки выполняющи
потоки производит разбор загруженных данных. В этом случае разумно иметь по одному производителю на сайт и, и ограничивать число производителей, если предел доступной скорости сети исчерпан.



Этот вопрос — адаптация одноименного исследования с Хэшкода.
    

Ответы

Ответ 1

Реализация на C#

Для современных версий языка (начиная с C# 4.0), имеет смысл не писать реализаци
вручную, а (руководствуясь советом @Flammable), воспользоваться классом BlockingCollection, представляющим нужную функциональность.

Для чтения в consumer-потоках используем просто циклы по последовательности, котору
даёт GetConsumingEnumerable(). В producer-потоках пользуемся Add, и в конце не забываем CompleteAdding, чтобы consumer-потоки смогли остановиться.

Пример:

class Program
{
    static public void Main()
    {
        new Program().Run();
    }

    BlockingCollection q = new BlockingCollection();

    void Run()
    {
        var threads = new [] { new Thread(Consumer), new Thread(Consumer) };
        foreach (var t in threads)
            t.Start();

        string s;
        while ((s = Console.ReadLine()).Length != 0)
            q.Add(s);

        q.CompleteAdding(); // останавливаем

        foreach (var t in threads)
            t.Join();
    }

    void Consumer()
    {
        foreach (var s in q.GetConsumingEnumerable())
        {
            Console.WriteLine("Processing: {0}", s);
            Thread.Sleep(2000);
            Console.WriteLine("Processed: {0}", s);
        }
    }
}


BlockingCollection позволяет ограничить количество элементов, так что попытк
добавить элемент в переполненную очередь также может быть заблокирована до освобождения места.

Заметьте, что GetConsumingEnumerable корректно работает даже в случае, когда у вас много консьюмеров. Это не так уж и очевидно.



Если вы работаете со старой версией C#, вам придётся писать нужную функциональност
вручную. Вы можете воспользоваться встроенным классом Monitor (который является аналогом mutex + condition variable из pthreads).

public class ProducerConsumer where T : class
{
    object mutex = new object();
    Queue queue = new Queue();
    bool isDead = false;

    public void Enqueue(T task)
    {
        if (task == null)
            throw new ArgumentNullException("task");
        lock (mutex)
        {
            if (isDead)
                throw new InvalidOperationException("Queue already stopped");
            queue.Enqueue(task);
            Monitor.Pulse(mutex);
        }
    }

    public T Dequeue()
    {
        lock (mutex)
        {
            while (queue.Count == 0 && !isDead)
                Monitor.Wait(mutex);

            if (queue.Count == 0)
                return null;

            return queue.Dequeue();
        }
    }

    public void Stop()
    {
        lock (mutex)
        {
            isDead = true;
            Monitor.PulseAll(mutex);
        }
    }
}


Использование (аналогичный пример):

class Program
{
    static public void Main()
    {
        new Program().Run();
    }

    ProducerConsumer q = new ProducerConsumer();

    void Run()
    {
        var threads = new [] { new Thread(Consumer), new Thread(Consumer) };
        foreach (var t in threads)
            t.Start();

        string s;
        while ((s = Console.ReadLine()).Length != 0)
            q.Enqueue(s);

        q.Stop();

        foreach (var t in threads)
            t.Join();
    }

    void Consumer()
    {
        while (true)
        {
            string s = q.Dequeue();
            if (s == null)
                break;
            Console.WriteLine("Processing: {0}", s);
            Thread.Sleep(2000);
            Console.WriteLine("Processed: {0}", s);
        }
    }
}


Ответ 2

Реализация на C с библиотекой pthreads

В C, в соответствии с духом языка, нет встроенных высокоуровневых синхронизирующихс
коллекций. Наверное самой популярной и широко используемой библиотекой, реализующей многопоточность, является pthreads. С её помощью паттерн можно реализовать так:

#include 

// объявляем структуру данных для одного задания
struct producer_consumer_queue_item {
  struct producer_consumer_queue_item *next;
  // здесь идут собственно данные. вы можете поменять этот кусок,
  // использовав структуру, более специфичную для вашей задачи
  void *data;
};

// объявляем очередь с дополнительными структурами для синхронизации.
// в этой очереди будут храниться произведённые, но ещё не потреблённые задания.
struct producer_consumer_queue {
  struct producer_consumer_queue_item *head, *tail;
                              // head == tail == 0, если очередь пуста
  pthread_mutex_t lock;       // мьютекс для всех манипуляций с очередью
  pthread_cond_t cond;        // этот cond "сигналим", когда очередь стала НЕ ПУСТОЙ
  int is_alive;               // показывает, не закончила ли очередь свою работу
};


Теперь нам нужны процедуры добавления и извлечения заданий из очереди.

void
enqueue (void *data, struct producer_consumer_queue *q)
{
  // упакуем задание в новую структуру
  struct producer_consumer_queue_item *p = (typeof(p))malloc(sizeof(*p));
  p->data = data;
  p->next = 0;

  // получим "эксклюзивный" доступ к очереди заданий
  pthread_mutex_lock(&q->lock);
  // ... и добавим новое задание туда:
  if (q->tail)
    q->tail->next = p;
  else {
    q->head = p;
    // очередь была пуста, а теперь нет -- надо разбудить потребителей
    pthread_cond_broadcast(&q->cond);
  }
  q->tail = p;

  // разрешаем доступ всем снова
  pthread_mutex_unlock(&q->lock);
}

void *
dequeue(struct producer_consumer_queue *q)
{
  // получаем эксклюзивный доступ к очереди:
  pthread_mutex_lock(&q->lock);

  while (!q->head && q->is_alive) {
    // очередь пуста, делать нечего, ждем...
    pthread_cond_wait(&q->cond, &q->lock);
    // wait разрешает доступ другим на время ожидания
  }

  // запоминаем текущий элемент или 0, если очередь умерла
  struct producer_consumer_queue_item *p = q->head;

  if (p)
  {
    // и удаляем его из очереди
    q->head = q->head->next;
    if (!q->head)
      q->tail = q->head;
  }

  // возвращаем эксклюзивный доступ другим участникам
  pthread_mutex_unlock(&q->lock);

  // отдаём данные
  void *data = p ? p->data : 0; // 0 означает, что данных больше не будет
  free(p);
  return data;
}


Ещё нужна процедура для инициализации очереди:

struct producer_consumer_queue *
producer_consumer_queue_create()
{
  struct producer_consumer_queue *q = (typeof(q))malloc(sizeof(*q));
  q->head = q->tail = 0;
  q->is_alive = 1;
  pthread_mutex_init(&q->lock, 0);
  pthread_cond_init(&q->cond, 0);

  return q;
}


И процедура для закрытия очереди:

void
producer_consumer_queue_stop(struct producer_consumer_queue *q)
{
  // для обращения к разделяемым переменным необходим эксклюзивный доступ
  pthread_mutex_lock(&q->lock);
  q->is_alive = 0;
  pthread_cond_broadcast(&q->cond);
  pthread_mutex_unlock(&q->lock);
}


Отлично, у нас есть всё, что нам надо.

Как использовать это? Нужно:


запустить несколько потоков-«производителей» и несколько «потребителей»
придумать структуру данных для задания


Пример: (производитель — главный поток, потребители — 2 потока)

// это поток-потребитель
void *
consumer_thread (void *arg)
{
  struct producer_consumer_queue *q = (typeof(q))arg;

  for (;;) {
    void *data = dequeue(q);
    // это сигнал, что очередь окончена
    if (!data)
      break; // значит, пора закрывать поток

    char *str = (char *)data;
    // тут наша обработка данных
    printf ("consuming: %s\n", str);
    sleep(2);
    printf ("consumed: %s\n", str);
    free(str);
  }
  return 0;
}

int
main ()
{
  pthread_t consumer_threads[2];

  void *res = 0;

  char *in = NULL;
  size_t sz;

  // создадим очередь:
  struct producer_consumer_queue *q = producer_consumer_queue_create();

  // и потоки-«потребители»
  pthread_create(&consumer_threads[0], 0, consumer_thread, (void *)q);
  pthread_create(&consumer_threads[1], 0, consumer_thread, (void *)q);

  // главный цикл
  // получаем данные с клавиатуры:
  while (getline(&in, &sz, stdin) > 0) {
    enqueue(in, q);
    in = NULL;
  }

  producer_consumer_queue_stop(q);
  if (pthread_join(consumer_threads[0], &res) ||
      pthread_join(consumer_threads[1], &res))
    perror("join");

  return (long)res;
}


Это реализация задачи с "бесконечной" очередью. На практике же иногда (или почт
всегда?) более полезно ограничить размер очереди и таким образом сбалансировать скорость производителей, иногда переводя их в спящее состояние, с возможностями потребителей.

Для этого немного изменим нашу producer_consumer_queue

struct producer_consumer_queue {
  struct producer_consumer_queue_item *head, *tail;
                              // head == tail == 0, если очередь пуста
  pthread_mutex_t lock;       // мьютекс для всех манипуляций с очередью
  pthread_cond_t condp;       // этот cond "сигналим", когда очередь стала НЕ ПУСТОЙ
  pthread_cond_t condc;       // этот cond "сигналим", когда в очереди ПОЯВИЛОС
СВОБОДНОЕ МЕСТО
  int is_alive;               // показывает, не закончила ли очередь свою работу
  int max, cnt,               // максимальный размер очереди и число заданий в ней
    pqcnt;                    // количество производителей, ждущих свободного места в очереди
};


Добавляем pthread_cond_t condc для "засыпания/пробуждения" потоков производителей
их счетчик в очереди на отправку сообщения и пару переменных, содержащих максимальный размер очереди и текущее количество заданий в ней.

Соответственно меняются функции для постановки задания в очередь (enqueue), выборк
его из очереди (dequeue), инициализации очереди (producer_consumer_queue_create) и ее остановки (producer_consumer_queue_stop):

void
enqueue (void *data, struct producer_consumer_queue *aq)
{
  volatile struct producer_consumer_queue *q = aq;
  // упакуем задание в новую структуру
  struct producer_consumer_queue_item *p = (typeof(p))malloc(sizeof(*p));
  p->data = data;
  p->next = 0;

  // получим "эксклюзивный" доступ к очереди заданий
  pthread_mutex_lock(&aq->lock);

  // проверим не переполнена ли она
  if (q->max <= q->cnt) {
    q->pqcnt++;
    asm volatile ("" : : : "memory");
    // зафиксируем изменения очереди в памяти
    // будем ждать пока потребители ее слегка не опустошат 
    while(q->max <= q->cnt & q->is_alive)
      pthread_cond_wait(&aq->condc, &aq->lock);
    q->pqcnt--;
    asm volatile ("" : : : "memory");
  }
  // ... и добавим новое задание туда:
  if (q->tail)
    q->tail->next = p;
  else {
    q->head = p;
    // очередь была пуста, а теперь нет -- надо разбудить потребителей
    pthread_cond_broadcast(&aq->condp);
  }
  q->tail = p;
  q->cnt++;
  asm volatile ("" : : : "memory");

  // разрешаем доступ всем снова
  pthread_mutex_unlock(&aq->lock);
}


void *
dequeue(struct producer_consumer_queue *aq)
{
  volatile struct producer_consumer_queue *q = aq;
  // получаем эксклюзивный доступ к очереди:
  pthread_mutex_lock(&aq->lock);

  if (q->pqcnt && q->max > q->cnt)
    // в очереди есть место, а кто-то спит, разбудим их
    pthread_cond_broadcast(&aq->condc);

  while (!q->head && q->is_alive) {
    // очередь пуста, делать нечего, ждем...
    pthread_cond_wait(&aq->condp, &aq->lock);
    // wait разрешает доступ другим на время ожидания
  }

  // запоминаем текущий элемент или 0, если очередь умерла
  struct producer_consumer_queue_item *p = q->head;
  if (p) {
    // и удаляем его из очереди
    q->head = q->head->next;
    if (!q->head)
      q->tail = q->head;
    q->cnt--;
    asm volatile ("" : : : "memory");
    // зафиксируем изменения очереди в памяти
    // разбудим поставщиков в их очереди
    pthread_cond_broadcast(&aq->condc);
  }

  // возвращаем эксклюзивный доступ другим участникам
  pthread_mutex_unlock(&aq->lock);

  // отдаём данные
  void *data = p ? p->data : 0; // 0 означает, что данных больше не будет
  free(p);
  return data;
}

struct producer_consumer_queue * 
producer_consumer_queue_create(int max)
{
  struct producer_consumer_queue *q = (typeof(q))malloc(sizeof(*q));
  q->head = q->tail = 0;
  q->is_alive = 1;
  q->max = max;
  q->cnt = 0;
  q->pqcnt = 0;
  pthread_mutex_init(&q->lock, 0);
  pthread_cond_init(&q->condc, 0);
  pthread_cond_init(&q->condp, 0);

  return q;
}

// И процедура для закрытия очереди:

void
producer_consumer_queue_stop(struct producer_consumer_queue *aq)
{
  volatile struct producer_consumer_queue *q = aq;
  // для обращения к разделяемым переменным необходим эксклюзивный доступ
  pthread_mutex_lock(&aq->lock);
  q->is_alive = 0;
  asm volatile ("" : : : "memory");
  pthread_cond_broadcast(&aq->condc);
  pthread_cond_broadcast(&aq->condp);
  pthread_mutex_unlock(&aq->lock);
}


Здесь же показан memory barrier (asm volatile ("" : : : "memory");), использование которого запрещает компилятору переупорядочивать операции чтения-записи из RAM.

Данная реализация не обеспечивает "упорядоченность" производителей, ожидающих свое
очереди для отправки сообщения. Т.е. поток производитель, "заснувший" первым из-за отсутствия свободного места в очереди не обязательно проснется первым. 

Если такое поведение нас не устраивает, то придется внести некоторые изменения 
наши данные, прежде всего добавив очередь поставщиков из структур producer_queue_item (которая будет частью структуры producer_consumer_queue. 

Получаем следующие структуры данных:

// объявляем структуру данных для одного задания
struct producer_consumer_queue_item {
  struct producer_consumer_queue_item *next;
  // здесь идут собственно данные. вы можете поменять этот кусок,
  // использовав структуру, более специфичную для вашей задачи
  void *data;
};

// струкура данных для спящего (ждущего свободного места) потока-производителя
struct producer_queue_item {
  struct producer_queue_item *next;
  struct producer_consumer_queue_item *item; // данные для которых нет места
  pthread_cond_t cond;  // этот cond "сигналим", когда в очереди появилось место

#if DEBUG
  pid_t tid;    // linux thread id for debug print
  int signaled; // индикатор "побудки" for debug print
#endif
};

// объявляем очередь данных с дополнительными структурами для синхронизации.
// в этой очереди будут храниться произведённые, но ещё не потреблённые задания.
struct producer_consumer_queue {
  struct producer_consumer_queue_item *head, *tail;
                              // head == tail == 0, если очередь пуста
  pthread_mutex_t lock;       // мьютекс для всех манипуляций с очередью
  pthread_cond_t cond;        // этот cond "сигналим", когда очередь стала НЕ ПУСТОЙ
  int is_alive;               // показывает, не закончила ли очередь свою работу
  int max, cnt;               // максимальный размер очереди и число заданий в ней
  // очередь  потоков-производителей, ждущих свободного места для своих данных
  struct producer_queue_item *pqhead,
    *pqtail;
};


и реализацию основных функций:

void
enqueue (void *data, struct producer_consumer_queue *q)
{
  volatile struct producer_consumer_queue *vq = q;
  // упакуем задание в новую структуру
  struct producer_consumer_queue_item *p = (typeof(p))malloc(sizeof(*p));
  p->data = data;
  p->next = 0;

  // получим "эксклюзивный" доступ к очереди заданий
  pthread_mutex_lock(&q->lock);

#if DEBUG
  printf("%ld (cnt: %d) ---> %s", (long)gettid(), vq->cnt, (char *)(p->data));
#endif

  // ... и добавим новое задание туда:
  if (vq->max <= vq->cnt || vq->pqtail) {// производитель должен ждать

#if DEBUG
    if (vq->cnt < vq->max) {
      puts("========================");
      print_queue(q, 0);
      puts("========================");
    }
#endif

    struct producer_queue_item *pq = (typeof(pq))malloc(sizeof(*pq));
    pthread_cond_init(&pq->cond, 0);  // cond по которому его разбудят
    pq->next = 0;
    pq->item = p;  // сохраним данные на время сна

#if DEBUG
    pq->tid = gettid();
#endif

    // поместим себя в очередь спящих производителей
    if (vq->pqtail)
      vq->pqtail->next = pq;
    else
      vq->pqhead = pq;
    vq->pqtail = pq;
    asm volatile ("" : : : "memory");
    // зафиксируем изменения очереди в памяти

#if DEBUG
    int at = 0; // счетчик циклов пробуждения
#endif

    do { // пойдем спать до появления свободного места в очереди данных

#if DEBUG
      printf ("%ld prod cond wait (cnt: %d  at: %d) %s", 
          (long)gettid(), vq->cnt, at++, (char *)(p->data));
      pq->signaled = 0;
#endif

      pthread_cond_wait(&pq->cond, &q->lock);
    } while(vq->max <= vq->cnt && vq->is_alive);
    // проснулись и владеем очередью

    /*
      Вот тонкий момент. Порядок активизации потоков не определен,
      а нам надо соблюдать очередность данных.
      Поэтому переустановим локальные переменные из очереди, 
      хотя это могут быть данные, положенные туда другим потоком.
    */
#if DEBUG
    if (pq != vq->pqhead) {
      printf ("BAAAD %ld (cnt: %d  at: %d) %s",
          (long)gettid(), vq->cnt, at, (char *)(p->data));
      print_queue(q, 0);
      if (vq->is_alive)
    exit(1); // совсем плохо, такого быть не должно
      else
    puts("CONTINUE");
    }
#endif

    pq = vq->pqhead;  // в любом случае берем голову очереди производителей
    if ((vq->pqhead = pq->next) == 0) // и удаляем ее
      vq->pqtail = 0;
    asm volatile ("" : : : "memory");
    p = pq->item;
    free(pq);

#if DEBUG
    printf ("%ld prod enqueued after wait (cnt: %d  at: %d) %s", 
        (long)gettid(), vq->cnt, at, (char *)(p->data));    
#endif

  }

  // вот тут реально кладем data в очередь для потребителей
  if (vq->tail)
    vq->tail->next = p;
  else {
    vq->head = p;
    // очередь была пуста, а теперь нет -- надо разбудить потребителей
    pthread_cond_broadcast(&q->cond);
  }
  vq->tail = p;
  vq->cnt++;
  asm volatile ("" : : : "memory");
  // сбросим изменения очереди в память

  // разрешаем доступ всем снова
  pthread_mutex_unlock(&q->lock);
}

#if DEBUG                   
#define cond_signal_producer(q) ({          \
      if ((q)->pqhead) {                \
      (q)->pqhead->signaled = 1;            \
      pthread_cond_signal(&(q)->pqhead->cond);  \
      }                         \
    })
#else
#define cond_signal_producer(q) ({          \
      if ((q)->pqhead)                  \
      pthread_cond_signal(&(q)->pqhead->cond);  \
    })
#endif

void *
dequeue(struct producer_consumer_queue *q)
{
  volatile struct producer_consumer_queue *vq = q;

  // получаем эксклюзивный доступ к очереди:
  pthread_mutex_lock(&q->lock);

  // если есть спящие производители, то разбудим первого
  cond_signal_producer(vq);  
  while (!vq->head && vq->is_alive) {
    // очередь пуста, делать нечего, ждем...
    pthread_cond_wait(&q->cond, &q->lock);
    // wait разрешает доступ другим на время ожидания
  }

  // запоминаем текущий элемент или 0, если очередь умерла
  struct producer_consumer_queue_item *p = vq->head;
  if (p) {
    // и удаляем его из очереди
    vq->head = vq->head->next;
    if (!vq->head)
      vq->tail = vq->head;
    vq->cnt--;
    asm volatile ("" : : : "memory");
    // сбросим изменения очереди в память
    // разбудим первого поставщика в их очереди
    cond_signal_producer(vq);
  }

  // возвращаем эксклюзивный доступ другим участникам
  pthread_mutex_unlock(&q->lock);

  // отдаём данные
  void *data = p ? p->data : 0; // 0 означает, что данных больше не будет
  // согласно 7.20.3.2/2, можно не проверять на 0
  free(p);
  return data;
}

struct producer_consumer_queue * 
producer_consumer_queue_create(int max)
{
  struct producer_consumer_queue *q = (typeof(q))malloc(sizeof(*q));
  q->head = q->tail = 0;
  q->pqhead = q->pqtail = 0;
  q->is_alive = 1;
  q->max = max;
  q->cnt = 0;
  pthread_mutex_init(&q->lock, 0);
  pthread_cond_init(&q->cond, 0);

  return q;
}

// И процедура для закрытия очереди:

void
producer_consumer_queue_stop(struct producer_consumer_queue *q)
{
  volatile struct producer_consumer_queue *vq = q;
  // для обращения к разделяемым переменным необходим эксклюзивный доступ
  pthread_mutex_lock(&q->lock);

  vq->is_alive = 0;
  pthread_cond_broadcast(&q->cond); // разбудим потребителей
  volatile struct producer_queue_item *pq;
  for (pq = vq->pqhead; pq; pq = pq->next) {
#if DEBUG
    pq->signaled = 1;
    asm volatile ("" : : : "memory");
#endif
    // будим каждого ждущего производителя
    pthread_cond_signal((pthread_cond_t *)&pq->cond);
  }

  pthread_mutex_unlock(&q->lock);
}


Все три программы (pq1.c, pq2.c и pq3.c) вместе с функцией gettid() находятся в http://pastebin.com/E23r9DZk . Для экспериментов скопируйте их в разные файлы и компилируйте, например, gcc pq3.c -pthread gettid.o


Ответ 3

Реализация на C#, библиотека Dataflow

Ещё одной альтернативой является использование Майкрософтовской библиотеки Dataflow
которая собственно и создана для того, чтобы управлять потоками данных. Для использовани
кода в примерах, вам нужно подключить NuGet-пакет Microsoft.Tpl.Dataflow. Класс BufferBlock практически является готовым producer/consumer'ом, но, в отличие от блокирующего интерфейса BlockingCollection, он имеет async-интерфейс!

Для асинхронного добавления задания поставщик может воспользоваться SendAsync. Асинхронно
добавление нужно, так как очередь может быть ограниченной длины, а значит, добавление должно будет дожидаться наличия свободного места! По окончанию добавления нужно вызвать Complete.

async Task ProduceSingle(ITargetBlock queue, int howmuch)
{
    Random r = new Random();
    while (howmuch-- > 0)
    {
        // эмулируем длительную работу по подготовке следующего задания
        // длительность выбираем случайно, чтобы задания приходили в
        // непредсказуемые моменты времени
        await Task.Delay(1000 * r.Next(1, 3));
        var v = string.Format("automatic {0}", r.Next(1, 10));
        await queue.SendAsync(v);
    }
    queue.Complete();
}


Если у вас несколько поставщиков, закрывать очередь нужно лишь когда они все отработают:

async Task Produce1(ITargetBlock queue, int howmuch)
{
    Random r = new Random();
    while (howmuch-- > 0)
    {
        await Task.Delay(1000 * r.Next(1, 3));
        var v = string.Format("automatic {0}", r.Next(1, 10));
        await queue.SendAsync(v);
    }
}

// функция Console.ReadLine() -- блокирующая, поэтому выполняем её асинхронно
// (иначе она заблокирует вызывающий поток)
// у Console нет async-интерфейса.
Task ReadConsole()
{
    // блокирующую функцию выгружаем в thread pool
    return Task.Run(() => Console.ReadLine());
}

async Task Produce2(ITargetBlock queue)
{
    string s;
    while ((s = await ReadConsole()).Length != 0)
        await queue.SendAsync("manual " + s);
}

async Task ProduceAll(ITargetBlock queue)
{
    var p1 = Produce1(queue, 20);
    var p2 = Produce2(queue);
    await Task.WhenAll(p1, p2);
    queue.Complete();
}


Теперь, потребитель. Если потребитель лишь один, всё просто:

async Task ConsumeSingle(ISourceBlock queue)
{
    while (await queue.OutputAvailableAsync())
        Console.WriteLine(await queue.ReceiveAsync());
}


Для случая нескольких потребителей использовать ReceiveAsync — неверно, так как задани
может быть взято из очереди другим потребителем! Функции TryReceiveAsync тоже нету, поэтому после асинхронного выяснения того, что очередь не пуста, используем TryReceive:

async Task ConsumeCooperative(IReceivableSourceBlock queue, int number)
{
    Random r = new Random();
    while (await queue.OutputAvailableAsync())
    {
        string v;
        // в этой точке данные могут быть уже уйти другому потребителю
        if (!queue.TryReceive(out v))
            continue; // продолжаем ждать

        // цветной вывод и прочие плюшки
        // мне лень синхронизировать вывод на консоль, хотя конечно это разделяемый ресурс
        if (Console.CursorLeft != 0)
            Console.WriteLine();
        var savedColor = Console.ForegroundColor;
        Console.ForegroundColor = (ConsoleColor)(number + 8);
        Console.WriteLine(string.Format("{0}[{1}]: {2}",
                          new string(' ', number * 4), number, v));
        Console.ForegroundColor = savedColor;

        // симулируем длительную обработку результата клиентом
        await Task.Delay(1000 * r.Next(1, 3));
    }
}

Task ConsumeAll(IReceivableSourceBlock queue)
{
    var c1 = ConsumeCooperative(queue, 1);
    var c2 = ConsumeCooperative(queue, 2);
    return Task.WhenAll(c1, c2);
}


Осталась обвязка:

class Program
{
    static void Main(string[] args)
    {
        new Program().RunAll().Wait();
    }

    async Task RunAll()
    {
        BufferBlock queue = new BufferBlock();
        var p = ProduceAll(queue);
        var c = ConsumeAll(queue);
        await Task.WhenAll(p, c, queue.Completion);
    }

    // остальные методы
}




В своей статье Async Producer/Consumer Queue using Dataflow Stephen Cleary предлагае
другой подход, более в духе библиотеки Dataflow. В ней заложена симметрия между блоками-источникам
(ISourceBlock), блоками-приёмниками (ITargetBlock), и блоками-преобразователями (IPropagatorBlock). В соответствии с этой идеологией, мы применяем для поставщика блок-приёмник ActionBlock:

Task Consume2(ISourceBlock queue, int number)
{
    var consumerOptions = new ExecutionDataflowBlockOptions { BoundedCapacity = 1 };
    var consumer = new ActionBlock(v => ConsumeImpl2(v, number), consumerOptions);
    var linkOptions = new DataflowLinkOptions { PropagateCompletion = true };
    queue.LinkTo(consumer, linkOptions);
    return consumer.Completion;
}

void ConsumeImpl2(string v, int number)
{
    Console.WriteLine(string.Format("[{0}]: {1}", number, v));
    Thread.Sleep(1500);
}

Task ConsumeAll2(ISourceBlock queue)
{
    var c1 = Consume2(queue, 1);
    var c2 = Consume2(queue, 2);
    return Task.WhenAll(c1, c2);
}


Для чего нужно BoundedCapacity = 1? Дел в том, что по умолчанию ActionBlock имее
«неограниченную» ёмкость, и таким образом за раз потребит все данные из очереди. Таким образом, если мы ввели ограничение на объём очереди

queue = new BufferBlock(new DataflowBlockOptions { BoundedCapacity = 20 });


то данные будут всё равно накапливаться в ActionBlock'е. Чтобы роль хранилища исполнялас
BufferBlock'ом, а потребителя — ActionBlock, и нужно ограничить его объём. Заметьте также, что ограничение ёмкости ActionBlock'а позволяет библиотеке Dataflow балансировать нагрузку, отправляя данные свободному блоку.

Заметьте, что в этом случае то, в каком контексте (пул потоков? выделенный поток?
исполняется ActionBlock, управляется не через стандартный механизм async/await, а посредством TaskScheduler'а в настройках.

Подход со стыковкой блоков выполнения может быстро стать слишком тяжеловесным, поэтому я советовал бы использовать его лишь для достаточно сложных задач.


Ответ 4

Реализация на C#, библиотека Dataflow ("инверсный" алгоритм)

Отличие данного алгоритма от приведенного VladD "прямого" - в том, что используется очередь потребителей вместо очереди элементов.

Это позволяет избавиться от цикла приема в потребителе - ценой появления знания списк
потребителей поставщиком данных. Иными словами, в такой конфигурации потребители выходят не активными - а пассивными.

Кроме того, сам алгоритм получается очень простым.

Потребитель в таком алгоритме не имеет никакого алгоритма, это лишь класс (или интерфейс либо вовсе делегат), в котором можно вызвать метод:

class Consumer
{
  public void Consume(string str)
  {
    Console.WriteLine(str);
  }
}


На стороне поставщика алгоритм немного сложнее:

BufferBlock consumers = new BufferBlock();

public async void SendToConsumer(string str)
{
  var consumer = await consumers.ReceiveAsync();
  try
  {
    consumer.Consume(str);
  }
  finally
  {
    consumers.Post(consumer);
  }
}

async Task Produce(int howmuch)
{
    Random r = new Random();
    while (howmuch-- > 0)
    {
        // эмулируем длительную работу по подготовке следующего задания
        // длительность выбираем случайно, чтобы задания приходили в
        // непредсказуемые моменты времени
        await Task.Delay(1000 * r.Next(1, 3));
        var v = string.Format("automatic {0}", r.Next(1, 10));
        SendToConsumer(v);
    }
}


К сожалению, это еще не все. Дело в том, что в таком виде любое исключение, возникшее в методе Consume, обрушит программу.

Поэтому надо сделать одну из двух вещей:


добавить в метод контракт класса Consumer требование обязательной обработки исключений;
или же установить глобальный обработчик неперехваченных исключений.


А еще лучше - воспользоваться сразу обоими вариантами. 

И еще один важный момент. Пассивность потребителя означает, в частности, что потребител
будет запущен в контексте синхронизации поставщика. Иногда это может быть нежелательным - в таком случае в код SendToConsumer следует добавить вызов Task.Run.

Иногда же, напротив, производителю нужно дождаться полной обработки порции данных
В таком случае надо сменить возвращаемое значение у метода SendToConsumer с void на Task.