同步1个生产者和多个消费者(多线程)

Question

我想实现生产者-消费者场景，其中有一个生产者线程和多个消费者线程。实际上，生产者线程需要以固定的时间间隔(例如，5秒)创建一组对象，而消费者线程需要使用这些对象。

我不确定如何定期创建一组对象，以及如何同步多个使用者。

提前感谢

Answer 1

你到底有什么问题？传统上，工作队列受到条件锁的保护；如果多个“使用者”正在等待同一个锁，那么只有一个人会成功地获得它。但是，您应该只从队列中抓取一个对象，将其从队列中移除，并在处理作业之前释放锁，而不是获取整个队列。

至于定期创建作业，这是您正在使用的任何库中的timer类的作业。如果您正在等待输入，则select()和poll()调用都有一个超时值；如果您什么都没做，可以只调用usleep()。

Answer 2

我不会详细说明所有代码，但您可以为此使用互斥和条件变量(以及链表)。基本模式是生产者执行以下操作：

loop_forever
    wait_until_interval_has_elapsed
    lock_the_mutex
        append_an_item_to_the_list
        signal_the_condvar // can be outside the mutex
    unlock_the_mutex

每个消费者都会这样做：

loop_forever
    lock_the_mutex
        while(list_is_empty)
            wait_the_condvar
        remove_an_item_from_the_list
    unlock_the_mutex
    process_the_item

您也可以使用互斥和信号量来完成此操作：

loop_forever
    wait_until_interval_has_elapsed
    lock_the_mutex
        append_an_item_to_the_list
    unlock_the_mutex
    post_the_semaphore

loop_forever
    wait_the_semaphore
    lock_the_mutex
        remove_an_item_from_the_list
    unlock_the_mutex
    process_the_item

这稍微简单一点，但是一旦您决定添加一种机制来告诉消费者完成他们正在做的事情并退出，您可能会更好地使用条件变量，因为您可以“广播”它。信号量很容易被误用，这就是为什么C++11没有信号量的原因。不过，Posix是这样的，所以在linux上您可以选择。公平地说，几乎所有其他的多线程操作系统也是如此。

在线程具有不同优先级的情况下，互斥/condvar对也可能提供比信号量更好的行为。与互斥锁不同的是，信号量没有“所有者”，因此优先级继承之类的技术是不可能的。

对于wait_until_interval_has_elapsed，使用计时器或休眠函数--请注意，除非您在实时系统上，否则您永远不能确保能够在特定时间运行，只能在特定时间被唤醒。

Answer 3

您需要一种方法来告诉使用者线程有新的对象可用。下面是一个例子(我没有包括#include)：

#define NUM_CONSUMERS 2

static int objects[4]; //The place for produced objects
static pthread_mutex_t cond_mut = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

void* producer_func(void* arg);
void* consumer_func(void* arg);

int main(int argc, char** argv) {
  pthread_t producer;
  pthread_t consumers[NUM_CONSUMERS];
  int res;
  void* zero = 0;
  res = pthread_create(&producer, NULL, producer_func, zero); 
  for(int i = 0; i < NUM_CONSUMERS; i++) {
    res = pthread_create(&consumers[i], NULL, consumer_func, zero);
  }
}

void* producer(void* arg) {
  while(1) {
    objects[0] = 0;
    objects[1] = 1;
    objects[2] = 2;
    objects[3] = 3;
    pthread_mutex_lock(&cond_mut);
    pthread_cond_signal(&cond);
    pthread_mutex_unlock(&cond_mut);
    sleep(5);
  }
}

void* consumer(void* arg) {
  while(1) {
    pthread_mutex_lock(&cond_mut);
    pthread_cond_wait(&cond);
    pthread_mutex_unlock(&cond_mut);
    //Process objects here 
  }
}