Linux下的生产者消费者模型模型

概念引入

linux下的生产者消费者模型举例说明就像我们在超市买东西，那么我们就是消费者，我们所消费商品的供货商显而易见就是生产者了，而其中将消费者和生产者联系起来的超市就是一个交易场所。

将上述场景类比到我们实际的软件开发过程中，代码的某个模块负责生产数据（供货商），而生产出来的数据却不得不交给另一模块（消费者）来对其进行处理，在这之间我们必须要有一个类似上述超市的东西来存储数据（超市），这就抽象成了我们的生产者/消费者模型。

其中，产生数据的模块，就形象地称为生产者；而处理数据的模块，就称为消费者；生产者和消费者之间的中介就叫做缓冲区。

1.为什么要使用生产者消费者模型

归根结底来说，生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这种生产消费能力不均衡的问题，所以便有了生产者和消费者模式。

2.生产者消费者模型的优点

(1).解耦—即降低生产者和消费者之间的依赖关系。

(2).支持并发, 即生产者和消费者可以是两个独立的并发主体，互不干扰的运行。

(3).支持忙闲不均，如果制造数据的速度时快时慢，缓冲区可以对其进行适当缓冲。当数据制造快的时候，消费者来不及处理，未处理的数据可以暂时存在缓冲区中。等生产者的制造速度慢下来，消费者再慢慢处理掉。

为了方便记忆，我们有个 “三二一” 原则

三种关系

１.生产者与生产者（互斥）

２.生产者与消费者（同步与互斥）

３.消费者与消费者（互斥）

两种角色

１.生产者

２.消费者

一个场所

１个缓冲区

3.三种模型

我们将以三种模型来了解生产者和消费者模型，分别是基于单链表的生产者消费者模型，基于环形队列的生产者消费者模型和基于环形队列的多生产者多消费者模型。实现之前我们先来了解下需要用到的两个背景知识。

互斥锁

就是维持我们上边说到的互斥关系的一种机制。下边是所用到的函数

1.初始化互斥锁

(1)

int pthread_mutex_init(pthread_mutex_t *mp, const pthread_mutexattr_t *mattr);

(2)

int pthread_mutex_init(pthread_mutex_t *mp,  const pthread_mutexattr_t *mattr);

返回值：成功返回0，任何其他返回值都表示错误。

当输入参数mattr为空指针时，pthread_mutex_init()函数以默认值初始化由参数mp指定的互斥锁。当输入参数mattr指向一个互斥锁属性对象时，pthread_mutex_init()函数被用来创建一个指定属性的互斥锁，其属性为参数mattr指向的互斥锁属性对象的属性。初始化一个互斥锁以后，这个互斥锁处在解锁状态。

2.锁定互斥锁

int pthread_mutex_lock(pthread_mutex_t *mutex); //阻塞式
int pthread_mutex_trylock(pthread_mutex_t *mutex); //非阻塞式

返回值：函数成功返回0；任何其他返回值都表示错误。

3.解锁互斥锁

int pthread_mutex_unlock(pthread_mutex_t *mutex);

返回值：成功返回0，任何其他返回值都表示错误。

解开已锁定的互斥锁，具体操作根据互斥锁的类型的不同而不同。

条件变量

条件变量是用来描述临界资源状态的，它具有原子性。

之所以要有条件变量是因为 如果消费者在消费的时候，缓存区没数据，那么消费就会失败，所以我们要创建一个条件变量来表示消费者在什么情况才能消费，如果条件变量的条件不满足那么线程就会进行等待，直到有人唤醒它为止。来看看相关函数：

1.初始化条件变量

类似上边的互斥锁的初始化。

2.等待

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);

3.唤醒

int pthread_cond_broadcast(pthread_cond_t *cond);//唤醒多个线程

int pthread_cond_signal(pthread_cond_t *cond);//唤醒单个线程

其实在实际应用中互斥锁和条件变量总是配合使用才能更好的发挥他们的作用。下来让我们用代码来实现

1.基于单链表的模型

基于单链表的生产者消费者模型需要先创建一条单链表，生产者往这条单链表放入结点，而消费者从这条单链表中取出结点。

代码 mycp.c

1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <pthread.h>
4
5 pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;//初始化互斥锁
6 pthread_cond_t cond = PTHREAD_COND_INITIALIZER;//初始化条件变量
7
8 typedef struct _node
9 {
10     int _data;
11     struct _node *_next;
12 }node_t, *node_p, **node_pp;//创建链表结构体
13
14 node_p head;//创建head链表
15
16 static node_p alloc_node(int data) //初始化链表
17 {
18     node_p tmp = (node_p)malloc(sizeof(node_t));
19     if(!tmp)
20     {
21         perror("malloc");
22         exit(1);
23     }
24     tmp->_data = data;
25     tmp->_next = NULL;
26 }
27
28 void initList(node_pp _h)
29 {
30     *_h = alloc_node(0);
31 }
32
33 void pushFront(node_p h, int data)//头插
34 {
35     node_p tmp = alloc_node(data);
36     tmp->_next = h->_next;
37     h->_next = tmp;
38 }
39
40 static void free_node(node_p _n)//释放节点
41 {
42     if(_n)
43     {
44         free(_n);
45     }
46 }
47
48 int isEmpty(node_p h)//判空
49 {
50     return h->_next == NULL ? 1 : 0;
51 }
52
53 void popFront(node_p h, int *out)//头删
54 {
55     if(!isEmpty(h))
56     {
57         node_p tmp = h->_next;
58         h->_next = tmp->_next;
59         *out = tmp->_data;
60         free_node(tmp);
61     }
62 }
63
64 void showList(node_p h)//打印链表
65 {
66     node_p start = h->_next;
67     while(start)
68     {
69         printf("%d ", start->_data);
70         start = start->_next;
71     }
72     printf("\n");
73 }
74
75 void destoryList(node_p h)//销毁链表
76 {
77     int out;
78     while(!isEmpty(h))
79     {
80         popFront(h, &out);
81     }
82     free_node(h);
83 }
84
85 void* product(void *arg)//生产者
86 {
87     while(1)
88     {
89         int data = rand()%1234;//产生随机数
90         pthread_mutex_lock(&lock);//加锁
91         pushFront(head, data);
92         pthread_mutex_unlock(&lock);//解锁
93         pthread_cond_signal(&cond);//生产完成后唤醒消费者
94         printf("The product success:%d\n", data);
95         sleep(1);
96     }
97 }
98
99 void* consume(void *arg)
100 {
101     while(1)
102     {
103         int data = -1;//消费失败返回-1
104         pthread_mutex_lock(&lock);
105         while(isEmpty(head))
106         {
107             printf("no data, The consumer wait...\n");
108             pthread_cond_wait(&cond, &lock);//链表中无节点等待
109         }
110         popFront(head, &data);
111         pthread_mutex_unlock(&lock);
112         printf("The consume success:%d\n", data);
113     }
114 }
115
116 int main()
117 {
118     pthread_t p;//定义线程p
119     pthread_t c;//定义线程c
120
121     pthread_create(&p, NULL, product, NULL);
122     pthread_create(&c, NULL, consume, NULL);
123     initList(&head);
124     destoryList(head);
125     pthread_join(p, NULL);
126     pthread_join(c, NULL);
127     return 0;
128 }

运行结果：



通过运行结果可以看到我们模拟出了生产者消费者的场景，线生产后消费，没有数据，生产者等待，有了数据生产者被唤醒在消费，实现这种机制互斥和条件变量的使用很重要。

2.基于环形队列的模型

什么是环形队列呢，我们知道队列是先进先出，那么环形队列就是把队列的首尾连接起来，我们看图



就像一个环形管道，数据在环中循环。举个例子，就像超市，货架上的东西摆满了供货商就会停止摆放货物，而消费者在货架上没有货物的时候就不能消费，总结来说在环形队列中消费者和生产者需要满足以下几个条件：

（1）生产者生产速度要快于消费者消费的速度；

（2）生产者不能将消费者套圈；

（3）应该以生产者先生产开始，当数据满溢时，则生产者等待消费者消费后再生产。

需要用到的关于信号量（semaphore）的函数

int sem_init(sem_t *sem, int pshared, unsigned int value); //初始化信号量

value：可用资源的数量；

pshared：为0表示信号量用于同一进程的线程。

sem_destroy()   //释放相关资源。
sem_wait()      //(P操作)   获得资源使semaphore的值减1；已经为0则挂起等待
sem_post()      //(V操作)   释放资源使semaphore的值加1；同时唤醒挂起线程

代码：ring.c

1 #include <stdio.h>
2 #include <pthread.h>
3 #include <semaphore.h>
4 #define SIZE 64
5 int ring[SIZE];//创建数组，存放数据
6
7 sem_t blank_sem;//定义信号量，表示格子量
8 sem_t data_sem;//定义信号量，表示格子中含有的数据个数
9
10 void* product(void *arg)
11 {
12     int data = 0;
13     int step = 0;
14     while(1)
15     {
16         int data = rand()%1234;
17         sem_wait(&blank_sem);//空格字数减1
18         ring[step] = data;//放入数据
19         sem_post(&data_sem);//格子中含有数据加一
20         printf("The product done: %d\n", data);
21         step++;
22         step %= SIZE;
23         sleep(1);
24     }
25 }
26
27 void* consume(void* arg)
28 {
29     int step = 0;
30     while(1)
31     {
32         int data = -1;//消费失败输出-1
33         sem_wait(&data_sem);//格子中含有的数据个数减一
34         data = ring[step];//拿出数据
35         sem_post(&blank_sem);//空格子数加一
36         printf("The consume done :%d\n",data);
37         step++;
21         step++;
22         step %= SIZE;
23         sleep(1);
24     }
25 }
26
27 void* consume(void* arg)
28 {
29     int step = 0;
30     while(1)
31     {
32         int data = -1;//消费失败输出-1
33         sem_wait(&data_sem);//格子中含有的数据个数减一
34         data = ring[step];//拿出数据
35         sem_post(&blank_sem);//空格子数加一
36         printf("The consume done :%d\n",data);
37         step++;
38         step %= SIZE;
39     }
40 }

运行结果：



从结果可以看出环形队列模型和链表模型的区别在于环形队列模型我们设定了条件后，生产者总是早于消费者，P V 操作保持了环形队列中格子中的数据原子性。

3.基于环形队列的多生产者多消费者模型。

环形队列的多生产者多消费者模型是在单生产者和消费者的基础上实现的，我们增加了生产者和消费者，需要注意的是我们的三二一原则里边的三种关系。下边我们来看代码，以三个生产者三个消费者为例。

代码 :ings.c

1 #include <stdio.h>
2 #include <pthread.h>
3 #include <semaphore.h>
4
5 #define SIZE 64
6 #define CONSUME 3 //定义生产者的个数
7 #define PRODUCT 3 //定义消费者的个数
8
9 int ring[SIZE];
10
11 pthread_mutex_t lock1 = PTHREAD_MUTEX_INITIALIZER;
12 pthread_mutex_t lock2 = PTHREAD_MUTEX_INITIALIZER;
13
14 sem_t blank_sem;
15 sem_t data_sem;
16
17 void* product(void *arg)
18 {
19     int data = 0;
20     static int step = 0;
21     while(1)
22     {
23         pthread_mutex_lock(&lock1);//加锁
24         int data = rand()%1234;
25         sem_wait(&blank_sem);//P操作
26         ring[step] = data;//产生数据
27         sem_post(&data_sem);//V操作
28         printf("The product done: %d\n", data);
29         step++;
30         step %= SIZE;
31         pthread_mutex_unlock(&lock1);//解锁
32         sleep(2);
33     }
11 pthread_mutex_t lock1 = PTHREAD_MUTEX_INITIALIZER;
12 pthread_mutex_t lock2 = PTHREAD_MUTEX_INITIALIZER;
13
14 sem_t blank_sem;
15 sem_t data_sem;
16
17 void* product(void *arg)
18 {
19     int data = 0;
20     static int step = 0;
21     while(1)
22     {
23         pthread_mutex_lock(&lock1);//加锁
24         int data = rand()%1234;
25         sem_wait(&blank_sem);//P操作
26         ring[step] = data;//产生数据
27         sem_post(&data_sem);//V操作
28         printf("The product done: %d\n", data);
29         step++;
30         step %= SIZE;
31         pthread_mutex_unlock(&lock1);//解锁
32         sleep(2);
33     }
34 }
35
36 void* consume(void *arg)
37 {
38     static int step = 0;
39     while(1)
40     {
41         pthread_mutex_lock(&lock2)//加锁;
42         int data = -1;
43         sem_wait(&data_sem);
44         data = ring[step];
45         sem_post(&blank_sem);
46         printf("The consume done: %d\n", data);
47         step++;
48         step %= SIZE;
49         pthread_mutex_unlock(&lock2);//解锁
50         sleep(2);
51     }
52 }
53
54 int main()
55 {
56     pthread_t p[PRODUCT];//创建生产者数组
57     pthread_t c[CONSUME];//创建消费者数组
58
59     int i = 0;
60     for(i = 0; i < PRODUCT; i++)//创建多生产者线程
61     {
62         pthread_create(&p[i], NULL, product, NULL);
63     }
64     for(i = 0; i < CONSUME; i++)//创建多消费者线程
65     {
66         pthread_create(&c[i], NULL, consume, NULL);
67     }
68
69     sem_init(&blank_sem, 0, SIZE);
70     sem_init(&data_sem, 0, 0);
71      for(i = 0; i < PRODUCT; i++)//生产者线程等待
72      {
45         sem_post(&blank_sem);
46         printf("The consume done: %d\n", data);
47         step++;
48         step %= SIZE;
49         pthread_mutex_unlock(&lock2);//解锁
50         sleep(2);
51     }
52 }
53
54 int main()
55 {
56     pthread_t p[PRODUCT];//创建生产者数组
57     pthread_t c[CONSUME];//创建消费者数组
58
59     int i = 0;
60     for(i = 0; i < PRODUCT; i++)//创建多生产者线程
61     {
62         pthread_create(&p[i], NULL, product, NULL);
63     }
64     for(i = 0; i < CONSUME; i++)//创建多消费者线程
65     {
66         pthread_create(&c[i], NULL, consume, NULL);
67     }
68
69     sem_init(&blank_sem, 0, SIZE);
70     sem_init(&data_sem, 0, 0);
71      for(i = 0; i < PRODUCT; i++)//生产者线程等待
72      {
73          pthread_join(p[i], NULL);
74      }
75      for(i = 0; i < CONSUME; i++)//消费者线程等待
76      {
77          pthread_join(c[i], NULL);
78      }
79
80      sem_destroy(&blank_sem);
81      sem_destroy(&data_sem);
82
83      pthread_mutex_destroy(&lock1);
84      pthread_mutex_destroy(&lock2);
85
86      return 0;
87 }

运行结果：