Linux|【Linux】生产者消费者模型线程|生产者-消费者

什么是生产者-消费者模型？
生产者-消费者是非常著名的一个问题。它的描述是：有一群生产者进程在生产产品，并将这些产品提供给消费者进程去消费，为使生产者与消费者能并发的执行，在两者之间设置了具有n个缓冲区的缓冲池，生产者进程将其所生产的产品放入一个缓冲区中；消费者进程可从一个缓冲区中取走产品。他们之间必须保持同步，即：不允许消费者到一个空的缓冲区中去取产品，也不允许生产者向一个已经装满产品的缓冲区中投放产品。当缓冲区为空时，消费者进程需要挂起休眠，直至生产者进程将产品放入缓冲区，消费者才能被唤醒；相反，如果缓冲区满时，消费者进程需要挂起，直至缓冲区中的产品被消费者拿走，才可以被唤醒。
举一个例子：我们把缓冲区比作平时购物的超市，生产者相当于是供货商，消费者就是去超市买东西的我们。
- 当超市的货物架上没有东西卖时，我们需要等到供货商送来货物才可以去买，超市的货物架上放满了货物，只有等到我们拿走了，才能继续上货，这是生产者和消费者之间的同步；
- 我们只有等超市摆好货物才能过去买，相反，超市也只能在我们把货物拿完才可以放；这是生产者与消费者之间的互斥（虽然在生活中这个例子不太恰当，但是在计算机中确实这样做的）；
- 假如超市中的某个产品中有一件，但是可能很多消费者都想要，这是消费者之间的竞争；
- 假如某个地区只有一个城市，那么可能会有很多生产者想把他们的产品放到超市卖，这是生产者之间的竞争。
所以，总结以上所述：

三种关系：
- 消费者和生产者：互斥与同步
- 消费者和消费者：互斥（竞争）
- 生产者和生产者：互斥（竞争）
两种角色：消费者、生产者
一个交易场所：具有存储数据的缓冲区

基于链表，使用互斥量和条件变量实现生产者-消费者
我们在Linux环境下，使用线程来模拟消费者和生产者。利用链表的插入操作模拟产生者生产，链表的删除操作来模拟消费者消费产品。
关于互斥量及条件变量的使用，可以参考之前的博客。
具体实现代码如下：

#include #include.h> #include.h> #include #includepthread_cond_t cond = PTHREAD_COND_INITIALIZER; //条件变量 pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; //互斥量typedef struct node//链表结点 { int data; struct node* next; }node, *node_p, **node_pp; node_p getnode(int data)//创建结点 { node_p ret = malloc(sizeof(node)); if(ret != NULL) { ret->data = https://www.it610.com/article/data; ret->next = NULL; return ret; } return NULL; }void initlist(node_pp pphead)//初始化链表——创建头结点 { if(pphead == NULL) exit(1); *pphead = malloc(sizeof(node)); (*pphead)->next = NULL; }int isEmpty(node_p phead)//链表判空，空返回1 { return (phead->next == NULL)?1:0; }void push(node_p phead, int data)//链表的头插 { node_p tmp = getnode(data); if(tmp != NULL) { tmp->next = phead->next; phead->next = tmp; } }void print(node_p phead)//打印链表结点的数据 { phead = phead->next; while(phead) { printf("%d", phead->data); phead = phead->next; } }void erase(node_p phead, int *x)//删除链表的数据 {if(isEmpty(phead)) return; node_p cur = phead->next; phead->next = cur->next; *x = cur->data; free(cur); }void destroy(node_p phead)//销毁链表 { while(!isEmpty(phead)) { int data; erase(phead, &data); } free(phead); }void *producer(void* arg)//生产者线程执行的函数 { node_p phead = (node_p)arg; while(1) { int data = https://www.it610.com/article/rand()%1000; pthread_mutex_lock(&mutex); push(phead, data); printf("producer done, %d\n", data); pthread_mutex_unlock(&mutex); pthread_cond_signal(&cond); sleep(1); } }void* consumer(void* arg)//消费者线程执行的函数 { node_p phead = (node_p)arg; int data = https://www.it610.com/article/0; while(1) { pthread_mutex_lock(&mutex); while(isEmpty(phead)) { printf("no product, please wait...\n"); pthread_cond_wait(&cond, &mutex); } erase(phead, &data); printf("consumer done, %d\n", data); pthread_mutex_unlock(&mutex); usleep(100000); } } int main() { node_p phead; initlist(&phead); srand((unsigned long)time(NULL)); pthread_t t1, t2; pthread_create(&t1, NULL, producer, (void*)phead); //生产者线程 pthread_create(&t2, NULL, consumer, (void*)phead); //消费者线程pthread_join(t1, NULL); //线程等待 pthread_join(t2, NULL); destroy(phead); //释放链表 pthread_mutex_destroy(&mutex); pthread_cond_destroy(&cond); return 0; }

运行结果：

文章图片

基于环形队列，使用POSIX信号量实现生产者-消费者
同样，我们在Linux环境下，使用线程模拟生产者和消费者。
首先，介绍一下POSIX信号量，它和System V信号量作用相同，达到无冲突的访问共享资源的目的，但POSIX可以用于线程间同步。
1. 初始化信号量

#include int sem_init(sem_t *sem, int pshared, unsigned int value);

参数：

pshared:0表示线程间共享，非0表示进程间共享
value:信号量的初始值

2. 销毁信号量

int sem_destroy(sem_t *sem);

3. 等待信号量

等待信号量，会将信号量的值减1，相当于P操作 int sem_wait(sem_t *sem);

【Linux|【Linux】生产者消费者模型】4. 发布信号量

发布信号量，表示资源使用完毕。将信号量值加1，相当于V操作 int sem_post(sem_t *sem);

环形队列

文章图片

具体实现代码：

#include #include #include #include#include #include #include#define SIZE 10 sem_t sem_blank; sem_t sem_data; int ring[SIZE]; void* produce(void* arg) { int i = 0; while(1) { sem_wait(&sem_blank); int data = https://www.it610.com/article/rand()%1234; ring[i] = data; i++; i %= SIZE; printf("producer done...data:%d\n", data); sem_post(&sem_data); sleep(1); } }void* consume(void* arg) { int i = 0; while(1) { sem_wait(&sem_data); int data = https://www.it610.com/article/ring[i]; i++; i %= SIZE; printf("consumemer done...data:%d\n", data); sem_post(&sem_blank); usleep(1000); } } int main() { srand((unsigned long)time(NULL)); pthread_t t1, t2; sem_init(&sem_blank, 0, SIZE); sem_init(&sem_data, 0, 0); pthread_create(&t1, NULL, produce, NULL); pthread_create(&t2, NULL, consume, NULL); pthread_join(t1, NULL); pthread_join(t2, NULL); sem_destroy(&sem_blank); sem_destroy(&sem_data); return 0; }

运行结果：