生产者消费者模型Linux C 实现

生产者消费者问题是同步问题中的一种常见情况，借用一下维基百科的话

生产者消费者问题（英语：Producer-consumer problem），也称有限缓冲问题（英语：Bounded-buffer problem），是一个多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

信号量配合互斥锁实现

信号量：
信号量的特性如下：信号量是一个非负整数（车位数），所有通过它的线程/进程（车辆）都会将该整数减一（通过它当然是为了使用资源），当该整数值为零时，所有试图通过它的线程都将处于等待状态。在信号量上我们定义两种操作： Wait（等待）和 Release（释放）。当一个线程调用Wait操作时，它要么得到资源然后将信号量减一，要么一直等下去（指放入阻塞队列），直到信号量大于等于一时。Release（释放）实际上是在信号量上执行加操作，对应于车辆离开停车场，该操作之所以叫做“释放”是因为释放了由信号量守护的资源。

wait, release在Linux下对应的函数为：
int sem_wait(sem_t * sem);

int sem_post(sem_t * sem);
设定两个信号量，empty用来表示空槽的个数，full用来表示占有的个数。
生产者在向任务队列里放资源时，调用sem_wait(&full)来检查队列是否已满，如果满的话，就阻塞，直到有消费者从里面取资源再苏醒，如果不满，就放资源，并通知消费者来取。

消费者在从任务队列里取资源时，调用sem_wait(&empty)来检查队列是否为空，如果空的话，就阻塞，直到有生产者向里面放资源再苏醒，如果不空，就取资源，并通知生产者来放。

而互斥锁仅仅是为了防止多个线程同时对队列进行操作，造成未知的结果。

#include <stdio.h>
#include <pthread.h>
#include <semaphore.h>

#define MAX 5  //队列长度

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
sem_t full;     //填充的个数
sem_t empty;    //空槽的个数

int top = 0;     //队尾
int bottom = 0;  //队头

void* produce(void* arg)
{
    int i;
    for ( i = 0; i < MAX*2; i++)
    {
        printf("producer is preparing data\n");
        sem_wait(&empty);//若空槽个数低于0阻塞
        
        pthread_mutex_lock(&mutex);
        
        top = (top+1) % MAX;
        printf("now top is %d\n", top);
        
        pthread_mutex_unlock(&mutex);
        
        sem_post(&full);
    }
    return (void*)1;
}

void* consume(void* arg)
{
    int i;
    for ( i = 0; i < MAX*2; i++)
    {
        printf("consumer is preparing data\n");
        sem_wait(&full);//若填充个数低于0阻塞
        
        pthread_mutex_lock(&mutex);
        
        bottom = (bottom+1) % MAX;
        printf("now bottom is %d\n", bottom);
        
        pthread_mutex_unlock(&mutex);
        
        sem_post(&empty);
    }
    
    return (void*)2;
}

int main(int argc, char *argv[])
{
    pthread_t thid1;
    pthread_t thid2;
    pthread_t thid3;
    pthread_t thid4;
    
    int  ret1;
    int  ret2;
    int  ret3;
    int  ret4;
    
    sem_init(&full, 0, 0);
    sem_init(&empty, 0, MAX);
    
    pthread_create(&thid1, NULL, produce, NULL);
    pthread_create(&thid2, NULL, consume, NULL);
    pthread_create(&thid3, NULL, produce, NULL);
    pthread_create(&thid4, NULL, consume, NULL);
    
    pthread_join(thid1, (void**)&ret1);
    pthread_join(thid2, (void**)&ret2);
    pthread_join(thid3, (void**)&ret3);
    pthread_join(thid4, (void**)&ret4);
    
    return 0;
}

注：如果把sem_wait()和sem_post()放到pthread_mutex_lock()与pthread_mutex_unlock()之间会如何呢？
答案是：死锁，因为我们不能预知线程进入共享区顺序，如果消费者线程先对mutex加锁，并进入，sem_wait()发现队列为空，阻塞，而生产者在对mutex加锁时，发现已上锁也阻塞，双方永远无法唤醒对方。

条件变量与互斥锁

条件变量的常见用法是在不满足某些条件时，阻塞自己，直到有线程通知自己醒来。

而互斥量在这里的作用依然还是防止多线程对共享资源同时操作，造成未知结果。

生产者消费者的行为与之前相同，只不过原来只调用sem_wait()可以完成两步，1是检查条件，2是阻塞，现在条件变量需要我们自己来设定条件(所以说条件变量配合互斥锁比信号量的功能更强大，因为它可以自定义休眠条件，但是这对使用者的要求也提高了，必须理清逻辑关系避免死锁)

#include <stdio.h>  
#include <pthread.h>  
  
#define MAX 5  
  
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;  
pthread_cond_t notfull = PTHREAD_COND_INITIALIZER;  //是否队满  
pthread_cond_t notempty = PTHREAD_COND_INITIALIZER;     //是否队空  
  
int top = 0;  
int bottom = 0;  
  
void* produce(void* arg)  
{  
    int i;  
    for ( i = 0; i < MAX*2; i++)  
    {  
        pthread_mutex_lock(&mutex);  
        while ((top+1)%MAX == bottom)  
        {  
            printf("full! producer is waiting\n");  
            pthread_cond_wait(&notfull, &mutex);//等待队不满  
        }  
  
        top = (top+1) % MAX;  
        printf("now top is %d\n", top);  
        pthread_cond_signal(&notempty);//发出队非空的消息  
  
        pthread_mutex_unlock(&mutex);  
    }  
    return (void*)1;  
}  
  
void* consume(void* arg)  
{  
    int i;  
    for ( i = 0; i < MAX*2; i++)  
    {  
        pthread_mutex_lock(&mutex);  
        while ( top%MAX == bottom)  
        {  
            printf("empty! consumer is waiting\n");  
            pthread_cond_wait(&notempty, &mutex);//等待队不空  
        }  
        bottom = (bottom+1) % MAX;  
        printf("now bottom is %d\n", bottom);  
        pthread_cond_signal(&notfull);//发出队不满的消息  
  
        pthread_mutex_unlock(&mutex);  
    }  
  
    return (void*)2;  
}  
  
int main(int argc, char *argv[])  
{  
    pthread_t thid1;  
    pthread_t thid2;  
    pthread_t thid3;  
    pthread_t thid4;  
  
    int  ret1;  
    int  ret2;  
    int  ret3;  
    int  ret4;  
  
    pthread_create(&thid1, NULL, produce, NULL);  
    pthread_create(&thid2, NULL, consume, NULL);  
    pthread_create(&thid3, NULL, produce, NULL);  
    pthread_create(&thid4, NULL, consume, NULL);  
  
    pthread_join(thid1, (void**)&ret1);  
    pthread_join(thid2, (void**)&ret2);  
    pthread_join(thid3, (void**)&ret3);  
    pthread_join(thid4, (void**)&ret4);  
  
    return 0;  
}

当然生产者消费者模型远没有如此简单，可以参考UNIX网络编程卷2中关于多生产者-单个消费者多个生产者-多个消费者模型的介绍。

关于条件变量与互斥量之间的关系：

1.线程在准备检查共享变量状态时锁定互斥量
2.检查共享变量状态（上面例子中的top%MAX == bottom）
3.如果共享变量未处于预期状态，线程应在等待条件变量并进入休眠前解锁互斥量，以便其它线程能够访问共享变量。
4.当线程因为条件变量的通知而被唤醒时，必须对互斥量再次加锁，因为在典型情况下，线程会立即访问共享变量。
关于上面第4条的解释：以生产者代码解释，如果缓冲区满了，线程在等待条件变量并进入休眠前解锁互斥量，消费者消耗缓冲区，生产者收到通知被唤醒后，重新对互斥量进行加锁。然后处理完共享变量后进行解锁。

信号量、互斥锁、条件变量差异

互斥锁是用在多线程多任务互斥的，一个线程占用了某一个资源，那么别的线程就无法访问，直到这个线程unlock，其他的线程才开始可以利用这个资源。比如对全局变量的访问，有时要加锁，操作完了，再解锁。在第一种生产者消费者模型实现中我们使用了信号量与互斥锁。
信号量不一定是锁定某一个资源，而是流程上的概念，比如：有A,B两个线程，B线程要等A线程完成某一任务以后再进行自己下面的步骤，这个任务并不一定是锁定某一资源，还可以是进行一些计算或者数据处理之类。而线程互斥量则是“锁住某一资源”的概念，在锁定期间内，其他线程无法对被保护的数据进行操作。
主要区别：

互斥锁必须是谁上锁就由谁来解锁，而信号量的wait和post操作不必由同一个线程执行。
互斥锁要么被锁住，要么被解开，和二值信号量类似
sem_post是各种同步技巧中，唯一一个能在信号处理程序中安全调用的函数
互斥锁是为上锁而优化的；条件变量是为等待而优化的；信号量既可用于上锁，也可用于等待，因此会有更多的开销和更高的复杂性
互斥锁，条件变量都只用于同一个进程的各线程间，而信号量（有名信号量）可用于不同进程间的同步。当信号量用于进程间同步时，要求信号量建立在共享内存区。
信号量有计数值，每次信号量post操作都会被记录，而条件变量在发送信号时，如果没有线程在等待该条件变量，那么信号将丢失。

自旋锁与睡眠

如果临界区可能包含引起睡眠的代码则不能使用自旋锁，否则可能引起死锁：
那么为什么信号量保护的代码可以睡眠而自旋锁会死锁呢？
解释:spin lock的自动禁止抢占的，进程A如果拿到锁以后，内核的抢占暂时被禁止。然后它休眠了，切换到另一个进程B（注意，这不是抢占，是进程自己放弃CPU）。等到进程B想要获得这个锁时发生了死锁，尽管B的时间片会被用完，但由于内核抢占被禁止了，所以B不会被调度出去。
更糟的情况是，如果进程A用irq save方式来得到这个spin lock，那中断是被禁止的，时钟中断不会被响应，进程B的时间片根本不会被更新。
总结下自旋锁的特点：
单CPU非抢占内核下：自旋锁会在编译时被忽略（因为单CPU且非抢占模式情况下，不可能发生进程切换，时钟只有一个进程处于临界区（自旋锁实际没什么用了）
单CPU抢占内核下：自选锁仅仅当作一个设置抢占的开关（因为单CPU不可能有并发访问临界区的情况，禁止抢占就可以保证临街区唯一被拥有）
多CPU下：此时才能完全发挥自旋锁的作用，自旋锁在内核中主要用来防止多处理器中并发访问临界区，防止内核抢占造成的竞争。
使用自旋锁时要注意：
由于自旋时不释放CPU，因而持有自旋锁的线程应该尽快释放自旋锁，否则等待该自旋锁的线程会一直在哪里自旋，这就会浪费CPU时间。
持有自旋锁的线程在sleep之前应该释放自旋锁以便其他线程可以获得该自旋锁。内核编程中，如果持有自旋锁的代码sleep了就可能导致整个系统挂起。