条件变量
条件变量本身不是锁！但它也可以造成线程阻塞。通常与互斥锁配合使用。给多线程提供一个会合的场所。
主要应用函数：
- pthread_cond_init 函数
- pthread_cond_destroy 函数
- pthread_cond_wait 函数
- pthread_cond_timedwait 函数
- pthread_cond_signal 函数
- pthread_cond_broadcast 函数
- 以上 6 个函数的返回值都是：成功返回 0， 失败直接返回错误号。
- pthread_cond_t 类型 用于定义条件变量
- pthread_cond_tcond;
pthread_cond_init 函数
初始化一个条件变量int pthread_cond_init(pthread_cond_t *restrictcond,const pthread_condattr_t *restrictattr);
参 2：attr 表条件变量属性，通常为默认值，传 NULL 即可
也可以使用静态初始化的方法，初始化条件变量：
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;
pthread_cond_destroy 函数
销毁一个条件变量
int pthread_cond_destroy(pthread_cond_t *cond);
pthread_cond_wait 函数
阻塞等待一个条件变量
int pthread_cond_wait(pthread_cond_t *restrictcond,pthread_mutex_t *restrictmutex);
函数作用：
- 阻塞等待条件变量 cond（参 1）满足
- 释放已掌握的互斥锁（解锁互斥量）相当于 pthread_mutex_unlock(&mutex);
- 1.2.两步为一个原子操作。
- 当被唤醒，pthread_cond_wait 函数返回时，解除阻塞并重新申请获取互斥锁 pthread_mutex_lock(&mutex);
pthread_cond_timedwait 函数
限时等待一个条件变量
int pthread_cond_timedwait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex, const struct timespec *restrictabstime);
参 3： 参看 mansem_timedwait 函数，查看 struct timespec 结构体。
struct timespec{
time_t tv_sec; /*seconds*/ 秒
long tv_nsec; /*nanosecondes*/ 纳秒
}
形参 abstime：绝对时间。
如：time(NULL)返回的就是绝对时间。
而 alarm(1)是相对时间，相对当前时间定时 1 秒钟。
struct timespect={1,0};
pthread_cond_timedwait(&cond,&mutex,&t); 只能定时到 1970 年 1 月 1 日 00:00:01 秒(早已经过去)
正确用法：
time_tcur=time(NULL); 获取当前时间。
structtimespect; 定义 timespec 结构体变量 t
t.tv_sec=cur+1; 定时 1 秒
pthread_cond_timedwait(&cond,&mutex,&t); 传参
setitimer 函数还有另外一种时间类型：
struct timeval
{
time_t tv_sec; /*seconds*/ 秒
suseconds_ttv_usec; /*microseconds*/ 微秒
};
pthread_cond_signal 函数
唤醒至少一个阻塞在条件变量上的线程
int pthread_cond_signal(pthread_cond_t*cond);
pthread_cond_broadcast 函数
唤醒全部阻塞在条件变量上的线程
int pthread_cond_broadcast(pthread_cond_t*cond);
生产者消费者条件变量模型
线程同步典型的案例即为生产者消费者模型，而借助条件变量来实现这一模型，是比较常见的一种方法。假定 有两个线程，一个模拟生产者行为，一个模拟消费者行为。两个线程同时操作一个共享资源（一般称之为汇聚）， 生产向其中添加产品，消费者从中消费掉产品。
/*借助条件变量模拟 生产者--消费者问题*/
#include<stdio.h>
#include<unistd.h>
#include<pthread.h>
#include<stdio.h>
#include<string.h>
/*链表作为共享数据，需要被互斥量保护*/
struct msg{
struct msg *next;
int num;
};
struct msg *head;
struct msg *mp;
/*静态初始化 一个条件变量 和一个互斥量*/
pthread_cond_t has_product = PTHREAD_COND_INITIALIZER;
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
void *consumer(void *p)
{
for(;;){
pthread_mutex_lock(&lock); //头指针为空，说明没有结点
while(head == NULL){
pthread_cond_wait(&has_product,&lock); //判断条件变量是否满足
}
mp = head;
head = mp->next; //模拟消费掉一个产品
pthread_mutex_unlock(&lock);
printf("---Consume ---%d\n",mp->num);
free(mp);
sleep(rand() % 5);
}
}
void *producer()
{
for(;;){
mp = malloc(sizeof(struct msg));
mp->num = rand() % 1000 + 1; //模拟生产一个产品
printf("--Produce ---%d\n",mp->num);
pthread_mutex_lock(&lock);
mp->next=head; //头插法
head = mp;
pthread_mutex_unlock(&lock); //释放
pthread_cond_signal(&has_product); //将等待在该条件变量上的一个线程>唤醒
sleep(rand() % 5);
}
}
int main(int argc,char *argv[])
{
pthread_t pid,cid; //pid生产者ID cid消费者ID
srand(time(NULL));
pthread_create(&pid,NULL,producer,NULL); //生产者
pthread_create(&cid,NULL,consumer,NULL); //消费者
pthread_join(pid,NULL);
pthread_join(cid,NULL);
return 0;
}
条件变量是搭配互斥锁一起使用的
- 因为条件变量实现同步只提供等待与唤醒功能，并没有提供条件判断的功能，因此条件判断需要用户实现，但是条件的操作是一个临界资源的操作，因此需要受保护，需要在条件判断之前加锁
- 如果加锁成功后，因为条件不满足而陷入休眠，就会导致卡死（因为另一方因为无法获取锁，而导致无法促使条件满足），因此需要在休眠之前解锁；并且解锁与休眠必须是原子操作
- 被唤醒之后，即将对临界资源进行操作，但是被操作前还要进行保护加锁
- 所以pthread_cond_wait集合了三步原子操作：解锁–>等待–>被唤醒后加锁
条件变量的优点
- 相较于 mutex 而言，条件变量可以减少竞争。
- 如直接使用 mutex，除了生产者、消费者之间要竞争互斥量以外，消费者之间也需要竞争互斥量，但如果汇聚 （链表）中没有数据，消费者之间竞争互斥锁是无意义的。有了条件变量机制以后，只有生产者完成生产，才会引 起消费者之间的竞争。提高了程序效率。
生产者与消费者模型（线程安全队列）
一个场所，两种角色，三种关系
功能：
- 解耦和（两个关系之间紧密）
- 支持忙闲不均
- 支持并发
三者关系
生产者–生产者：互斥
消费者–消费者：互斥
生产者–消费者：同步+互斥
/*生产者与消费者模型队列实现
* 1.实现线程安全的队列，对外提供线程安全的数据入队和出队操作
* 2.创建线程，分别作为生产者与消费者数据入队或数据出队
*/
#include<iostream>
#include<queue>
#include<pthread.h>
#define MAX_QUEUE 10
class BlockQueue
{
public:
BlockQueue(int cap = MAX_QUEUE):_capacity(cap){
//初始化队列
pthread_mutex_init(&_mutex,NULL);
pthread_cond_init(&_cond_con,NULL);
pthread_cond_init(&_cond_pro,NULL);
}
~BlockQueue(){
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond_con);
pthread_cond_destroy(&_cond_pro);
}
//入队
void QueuePush(int data){
QueueLock();
while(QueueIsFull()){ //队列满了
ProWait(); //生产者等待
}
_queue.push(data);
ConWakeUp();
QueueUnLock();
}
void QueuePop(int *data){
QueueLock();
while(QueueIsEmpty()){
ConWait();
}
*data = _queue.front();//获取队列头结点
_queue.pop();//结点出队
ProWakeUp();
QueueUnLock();
}
private:
//队列加锁
void QueueLock(){
pthread_mutex_lock(&_mutex);
}
//队列解锁
void QueueUnLock(){
pthread_mutex_unlock(&_mutex);
}
//消费者等待
void ConWait(){
pthread_cond_wait(&_cond_con,&_mutex);
}
//消费者唤醒
void ConWakeUp(){
pthread_cond_signal(&_cond_con);
}
//生产者等待
void ProWait(){
pthread_cond_wait(&_cond_pro,&_mutex);
}
//生产者唤醒
void ProWakeUp(){
pthread_cond_signal(&_cond_pro);
}
//判断队列是否为空
bool QueueIsFull(){
return (_capacity == _queue.size());
}
//队列是否是满的
bool QueueIsEmpty(){
return _queue.empty();
}
private:
std::queue<int>_queue;//创建队列
int _capacity;//队列结点最大数量
//线程安全实现成员
pthread_mutex_t _mutex;
pthread_cond_t _cond_pro;
pthread_cond_t _cond_con;
};
void *thr_consumer(void *arg){
BlockQueue *q = (BlockQueue *)arg;
while(1){
int data;
q->QueuePop(&data);
std::cout<<"consumer"<<pthread_self() <<" get data:"<< data <<std::endl;
}
return NULL;
}
int i = 0; //必须受保护
pthread_mutex_t mutex;
void *thr_productor(void *arg){
BlockQueue *q = (BlockQueue *)arg;
while(1){
pthread_mutex_lock(&mutex);
q->QueuePush(i++);
pthread_mutex_unlock(&mutex);
std::cout<<"productor:" <<pthread_self() <<"put data:"<< i <<std::endl;
}
return NULL;
}
int main(int argc,char *argv[])
{
BlockQueue q;
pthread_t ctid[4],ptid[4];
int i,ret;
pthread_mutex_init(&mutex,NULL);
for(i = 0;i < 4; i++){
ret = pthread_create(&ctid[i],NULL,thr_consumer,(void *)&q);
if(ret != 0){
std::cout<<"pthread create error\n";
return -1;
}
}
for(i = 0;i < 4; i++){
ret = pthread_create(&ptid[i],NULL,thr_productor,(void *)&q);
if(ret != 0){
std::cout<<"pthread create error\n";
return -1;
}
}
for(i = 0;i < 4; i++){
pthread_join(ctid[i],NULL);
}
for(i = 0; i < 4;i++){
pthread_join(ptid[i],NULL);
}
return 0;
}
信号量
进化版的互斥锁（1–>N）
由于互斥锁的粒度比较大，如果我们希望在多个线程间对某一对象的部分数据进行共享，使用互斥锁是没有办 法实现的，只能将整个数据对象锁住。这样虽然达到了多线程操作共享数据时保证数据正确性的目的，却无形中导 致线程的并发性下降。线程从并行执行，变成了串行执行。与直接使用单进程无异。
信号量，是相对折中的一种处理方式，既能保证同步，数据不混乱，又能提高线程并发。
计数器+等待队列+等待与唤醒功能
- 通过自身的计数器实现条件判断，当前条件满足时则直接返回并且计数-1.当条件并不满足时则阻塞
- 当产生资源后，通过信号量的唤醒功能唤醒等待并且计数+1
信号量和条件变量实现同步的区别
- 信号量的条件判断由自身来完成，而条件变量的条件判断由用户完成
- 信号量并不搭配互斥锁使用，而条件变量需要搭配互斥锁一起使用保护条件的改变
sem_init 函数
初始化一个信号量int sem_init(sem_t *sem,int pshared,unsigned int value);
参 1：sem 信号量
参 2：pshared 取 0 用于线程间；取非 0（一般为 1）用于进程间
参 3：value 指定信号量初值
sem_destroy 函数
销毁一个信号量
int sem_destroy(sem_t *sem);
sem_wait 函数
给信号量加锁 ，对计数进行判断，计数<=0则阻塞；否则立即返回流程继续，计数-1
int sem_wait(sem_t *sem);
sem_post 函数
给信号量解锁 ，对计数进行+1，并且唤醒等到的线程
int sem_post(sem_t *sem);
sem_trywait 函数
尝试对信号量加锁
(与 sem_wait 的区别类比 lock 和 trylock)
int sem_trywait(sem_t *sem);
sem_timedwait 函数
限时尝试对信号量加锁
int sem_timedwait(sem_t *sem,const struct timespec *abs_timeout);
参 2：abs_timeout 采用的是绝对时间。
定时 1 秒：
time_tcur=time(NULL); 获取当前时间。
structtimespect; 定义 timespec 结构体变量 t
t.tv_sec=cur+1; 定时 1 秒
t.tv_nsec=t.tv_sec+100;
sem_timedwait(&sem,&t); 传参
使用信号量实现生产者与消费者模型
/*使用信号量实现生产者与消费者模型
*
*/
#include<iostream>
#include<queue>
#include<pthread.h>
#include<semaphore.h>
class RingQueue
{
public:
RingQueue(int cap = 10):_capacity(cap),_queue(cap){
//1.信号量变量
//2.参数取值 0：用于线程间同步与互斥
// 非0：用于进程间同步与互斥
//3.信号量初值
sem_init(&_sem_lock,0,1);//互斥锁初始值只给1
sem_init(&_sem_data,0,0);//初始数据资源数据为0
sem_init(&_sem_space,0,cap);//初始空闲空间计数
}
~RingQueue(){
sem_destroy(&_sem_lock);
sem_destroy(&_sem_data);
sem_destroy(&_sem_space);
}
void QueuePush(int data){
// ProWait();//空闲空间计数判断是否有空闲空间，若有返回，否则等待
// 因为已经通过_sem_space的空闲空间计数知道是否有空闲空间
sem_wait(&_sem_space);//添加数据之后，空闲空间计数-1
sem_wait(&_sem_lock);//锁计数初始为1，一旦进入-1加锁
_queue[_step_write]=data;
_step_write = ( _step_write + 1) % _capacity;
sem_post(&_sem_lock);//数据添加完毕后解锁，数据资源计数+1
sem_post(&_sem_data);//数据添加完毕后，数据资源计数+1
//ConWakeUp();
}
void QueuePop(int *data){
sem_wait(&_sem_data);//取数据的时候，数据资源计数-1
sem_wait(&_sem_lock);//锁最好仅仅保护临界区
*data = _queue[_step_read];
_step_read = (_step_read + 1) % _capacity;
sem_post(&_sem_lock);
sem_post(&_sem_space);//取数据之后，空闲空间计数+!
}
private:
std::vector<int>_queue;
int _capacity; //队列最大数量
int _step_write;//当前写到哪里的下标
int _step_read;//当前读到哪里了的下标
sem_t _sem_lock;//实现互斥锁
sem_t _sem_space;//空闲空间计数
sem_t _sem_data;//数据资源计数
/*
//队列加锁
void QueueLock(){
pthread_mutex_lock(&_mutex);
}
//队列解锁
void QueueUnLock(){
pthread_mutex_unlock(&_mutex);
}
*/
};
void *thr_productor(void *arg){
RingQueue *q = (RingQueue*)arg;
int i=0;
while(1){
q->QueuePush(i);
std::cout<<"thread:"<<pthread_self()<<"put data"<<i++<<"\n";
}
return NULL;
}
void *thr_consumer(void *arg){
RingQueue *q = (RingQueue*)arg;
while(1){
int data;
q->QueuePop(&data);
std::cout<<"thread:"<<pthread_self()<<"get data"<<data<<"\n";
}
return NULL;
}
int main(int argc,char *argv[])
{
RingQueue q;
pthread_t ptid,ctid[4];
int i ,ret;
ret = pthread_create(&ptid,NULL,thr_productor,(void *)&q);
if(ret != 0){
std::cout<<"thread create error\n";
return -1;
}
for(i = 0;i < 4;i++){
ret = pthread_create(&ctid[i],NULL,thr_consumer,(void *)&q);
if(ret != 0){
std::cout<<"thread create error\n";
return -1;
}
}
for(i = 0; i < 4; i++){
pthread_join(ctid[i],NULL);
}
pthread_join(ptid,NULL);
return 0;
}