1、生产者消费者模型
有两个进程:一组生产者进程和一组消费者进程共享一个初始为空、固定大小为n的缓存(缓冲区)。生产者的工作是制造一段数据,只有缓冲区没满时,生产者才能把消息放入到缓冲区,否则必须等待,如此反复; 同时,只有缓冲区不空时,消费者才能从中取出消息,一次消费一段数据(即将其从缓存中移出),否则必须等待。由于缓冲区是临界资源,它只允许一个生产者放入消息,或者一个消费者从中取出消息。
生产者和消费者对缓冲区互斥访问是互斥关系,同时生产者和消费者又是一个相互协作的关系,只有生产者生产之后,消费者才能消费,他们也是同步关系。
解决思路:对于生产者,如果缓存是满的就去睡觉。消费者从缓存中取走数据后就叫醒生产者,让它再次将缓存填满。若消费者发现缓存是空的,就去睡觉了。下一轮中生产者将数据写入后就叫醒消费者。
只有生产者和消费者两个进程,正好是这两个进程存在着互斥关系和同步关系。那么需要解决的是互斥和同步PV操作的位置。
2、互斥量&条件变量
2.1 互斥量(mutex)
Linux环境下:
mutex互斥量的引入主要是因为,大部分情况,线程使用的数据都是局部变量,变量的地址空间在线程栈空间内,这种情况,变量归属单个进程,其他线程无法获得这种变量。但有时候,很多变量都需要在线程间共享,这样的变量成为共享变量,可以通过数据的共享,完成线程间的交互。多个线程并发的操作共享变量,可能会产生一些无法预期的错误,例如线程A要用的共享变量可能刚刚被线程B修改了。
引入互斥锁,可以解决上述问题:
- 代码必须有互斥行为:当代码进入临界区执行时,不允许其他线程进入该临界区。
- 如果多个线程同时要求执行临界区的代码,并且临界区没有线程在执行,那么只能允许一个线程进入该临界区。
- 如果线程不在临界区中执行,那么该线程不能阻止其他线程进入临界区。
这里写图片描述
相关头文件和API:
#include<pthread.h>
#include<errno.h>
//初始化信号量接口,如果使用默认的属性初始化互斥量, 只需把attr设为NULL.
int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restric attr);
//销毁信号量对象接口
int pthread_mutex_destroy(pthread_mutex_t *mutex);
//互斥量加锁接口--阻塞式
//说明:对共享资源的访问, 要对互斥量进行加锁, 如果互斥量已经上了锁, 调用线程会阻塞, 直到互斥量被解锁。在完成了对共享资源的访问后, 要对互斥量进行解锁。
int pthread_mutex_lock(pthread_mutex_t *mutex);
//互斥量加锁接口--非阻塞式
//说明: 这个函数是非阻塞调用模式, 也就是说, 如果互斥量没被锁住, trylock函数将把互斥量加锁, 并获得对共享资源的访问权限; 如果互斥量被锁住了, trylock函数将不会阻塞等待而直接返回EBUSY,表示共享资源处于忙状态。
int pthread_mutex_trylock(pthread_mutex_t *mutex);
//互斥量解锁接口
int pthread_mutex_unlock(pthread_mutex_t *mutex);
必须初始化互斥量:
- 方法一,静态分配: pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
- 方法二,动态分配:使用pthread_mutex_init 函数初始化(attr一般置NULL);
销毁互斥量需要注意:
- 使用PTHREAD_MUTEX_INITIALIZER初始化的互斥量不需要销毁;
- 不要销毁一个已经加锁的互斥量;
- 已经销毁的互斥量,要确保后面不会有线程在尝试加锁;
总体来讲, 有几个不成文的基本原则:
- 对共享资源操作前一定要获得锁。
- 对共享资源操作前一定要获得锁。
- 尽量短时间地占用锁。
- 如果有多锁, 如获得顺序是ABC连环扣, 释放顺序也应该是ABC。
- 线程错误返回时应该释放它所获得的锁。
各种Mutex的区别:
锁类型 | 初始化方式 | 加锁特征 | 调度特征 |
---|---|---|---|
普通锁 | PTHREAD_MUTEX_INITIALIZER | 同一线程可重复加锁,解锁一次释放锁 | 先等待锁的进程先获得锁 |
嵌套锁 | PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP | 同一线程可重复加锁,解锁同样次数才可释放锁 | 先等待锁的进程先获得锁 |
纠错锁 | PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP | 同一线程不能重复加锁,加上的锁只能由本线程解锁 | 先等待锁的进程先获得锁 |
自适应锁 | PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP | 同一线程可重加锁,解锁一次生效 | 所有等待锁的线程自由竞争 |
2.2 条件变量
条件变量:
条件变量使我们可以睡眠等待某种条件出现。
条件变量是利用线程间共享的全局变量进行同步的一种机制,主要包括两个动作:一个线程等待”条件变量的条件成立”而挂起;另一个线程使”条件成立”(给出条件成立信号)。为了防止竞争,条件变量的使用总是和一个互斥锁结合在一起。
条件变量类型为pthread_cond_;
初始化:
int pthreade_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *rest rict attr);
//cond:要初始化的变量
//arrt:NULL
销毁:
int pthread_cond_destroy(pthread_cond_t *cond);
等待条件满足:
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *testrict mutex);
//cond:要在这个条件变量上等待
//mutex:互斥量
唤醒等待:
int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);
3、实现源码
#include<stdlib.h>
#include<stdio.h>
#include<unistd.h>
#include<pthread.h>
#define CONSUMERS_COUNT 2
#define PRODUCERS_COUNT 2
typedef struct msg{
struct msg *next;
int num;
}msg;
msg *head = NULL;
pthread_cond_t cond; //条件变量
pthread_mutex_t mutex; //互斥量
pthread_t threads[CONSUMERS_COUNT+PRODUCERS_COUNT];//生产者消费者模型中的缓冲池
void *consumer(void *p){
int num = *(int *)p;
free(p);//p是在main函数里malloc的
msg *mp;
while(1){
//给互斥量mutex加锁
pthread_mutex_lock(&mutex);
while(head == NULL){
printf("%d begin wait a condition...\n",num);
pthread_cond_wait(&cond,&mutex);
}
printf("%d end wait a condition...\n",num);
printf("%d begin consume product...\n",num);
mp = head;
head = mp->next;
pthread_mutex_unlock(&mutex);
printf("Consume %d\n",mp->num);
free(mp);
printf("%d end consume product...\n",num);
sleep(rand()%5);
}
}
void *producer(void *p){
msg *mp;
int num = *(int *)p;
free(p);
while(1){
printf("%d begin produce product...\n",num);
mp = (msg*)malloc(sizeof(msg));
mp->num = rand()%1000 + 1;
printf("produce %d\n",mp->num);
pthread_mutex_lock(&mutex);
mp->next = head;
head = mp;
printf("%d end produce product...\n",num);
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
sleep(rand()%5);
}
}
int main(void){
srand(time(NULL));
//初始化条件变量cond,互斥量mutex
pthread_cond_init(&cond,NULL);
pthread_mutex_init(&mutex,NULL);
int i;
for(i=0;i<CONSUMERS_COUNT;i++){
int *p = (int *)malloc(sizeof(int));
if(NULL == p){
perror("malloc");
continue;
}
*p=i;
pthread_create(&threads[i],NULL,consumer,(void*)p);
}
for(i=0;i<PRODUCERS_COUNT;i++){
int *p = (int *)malloc(sizeof(int));
*p = i;
if(NULL == p){
perror("malloc");
continue;
}
pthread_create(&threads[CONSUMERS_COUNT+i],NULL,producer,(void*)p);
}
for(i=0;i<CONSUMERS_COUNT+PRODUCERS_COUNT;i++){
pthread_join(threads[i],NULL);
}
//销毁互斥量mutex,条件变量cond
pthread_mutex_destroy(&mutex);
pthread_cond_destroy(&cond);
}