用条件变量来解决生产者消费者问题

在上一篇文章中，我使用了匿名信号量

什么是条件变量？
我们可以设想一个场景：线程A需要某个条件成立才能继续执行，否则一直等待下去，而线程B执行过程中使线程的执行条件成立，并且唤醒A。
举个例子，在生产者消费者模型中，消费者如果看到缓冲区为空时，就等待，而生产者往缓冲区添加完数据后，唤醒消费者。
其中这个场景中，可以POSIX的条件变量来实现。
以下是条件变量所需要的函数

pthread_cond_init
pthread_cond_destroy
pthread_cond_wait
pthread_cond_signal
pthread_cond_broadcast

这里大家就百度怎么用，或者在Linux中用man 3 pthread_cond_***的方法也行。
其中重点讲一下的是 pthread_cond_wait这个函数。
条件变量是要和互斥量配合使用的，而pthread_cond_wait的搭配互斥量后是执行者三步操作。

第一，条件不成立时，释放互斥量（将互斥量解锁）。
第二，阻塞自己，让别的线程开始执行。
第三，当被其他的线程唤醒时，重新获得互斥量（重新将互斥量加锁），并执行后续代码。

掌握了这个之后，就来说下条件变量的使用规范。
第一，等待条件代码

pthread_mutex_lock(&mutex);
while (条件为假）
    pthread_cond_wait(cond, mutex);
修改条件;
pthread_mutex_unlock(&mutex);

第二，修改条件代码

pthread_mutex_lock(&mutex);
设置条件为真
pthread_cond_signal(cond);
pthread_mutex_unlock(&mutex);

在这里的一个难点是，第一段条件等待代码为什么不用if而用while呢？在man 3 pthread_cond_wait 之后发现这样一段话

If a signal is delivered to a thread waiting for a condition variable, upon return from the signal handler the thread resumes waiting for the condition variable as if it was not interrupted, or it shall return zero due to spurious wakeup.

也就是说，唤醒条件的信号，可以唤醒多个线程，但是只能允许一个信号访问，也就是说，因此等待线程需要不断的用while轮询一直到达到条件了才行。

好了，讲了这么多我们用它来做个生产者消费者实例吧。

#include <unistd.h>
#include <sys/types.h>
#include <pthread.h>
#include <semaphore.h>

#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <string.h>

#define ERR_EXIT(m) \
        do \
        { \
                perror(m); \
                exit(EXIT_FAILURE); \
        } while(0)

#define CONSUMERS_COUNT 2
#define PRODUCERS_COUNT 1

pthread_mutex_t g_mutex;
pthread_cond_t g_cond;

pthread_t g_thread[CONSUMERS_COUNT + PRODUCERS_COUNT];

int nready = 0;

void *consume(void *arg)
{
    int num = (int)arg;
    while (1)
    {
        pthread_mutex_lock(&g_mutex);
        while (nready == 0)
        {
            printf("%d begin wait a condtion ...\n", num);
            pthread_cond_wait(&g_cond, &g_mutex);
        }

        printf("%d end wait a condtion ...\n", num);
        printf("%d begin consume product ...\n", num);
        --nready;
        printf("%d end consume product ...\n", num);
        pthread_mutex_unlock(&g_mutex);
        sleep(1);
    }
    return NULL;
}

void *produce(void *arg)
{
    int num = (int)arg;
    while (1)
    {
        pthread_mutex_lock(&g_mutex);
        printf("%d begin produce product ...\n", num);
        ++nready;
        printf("%d end produce product ...\n", num);
        pthread_cond_signal(&g_cond);
        printf("%d signal ...\n", num);
        pthread_mutex_unlock(&g_mutex);
        sleep(1);
    }
    return NULL;
}

int main(void)
{
    int i;

    pthread_mutex_init(&g_mutex, NULL);
    pthread_cond_init(&g_cond, NULL);


    for (i = 0; i < CONSUMERS_COUNT; i++)
        pthread_create(&g_thread[i], NULL, consume, (void *)i);

    sleep(1);

    for (i = 0; i < PRODUCERS_COUNT; i++)
        pthread_create(&g_thread[CONSUMERS_COUNT + i], NULL, produce, (void *)i);

    for (i = 0; i < CONSUMERS_COUNT + PRODUCERS_COUNT; i++)
        pthread_join(g_thread[i], NULL);

    pthread_mutex_destroy(&g_mutex);
    pthread_cond_destroy(&g_cond);

    return 0;
}

最后，下图为代码执行结果。