线程（四）线程的同步——条件变量

线程

线程的同步和互斥

线程同步–条件变量

• 是一个宏观概念，在微观上包含线程的相互排斥和线程先后执行的约束问题
• 解决同步方式
  • 条件变量
  • 线程信号量

什么是线程同步

举个例子：假如现在要做一款产品，在产品发行之前要进行研发和测试两个环节。现在将研发和测试看成两个线程，将产品看作两个线程操作的共享资源。对产品的测试只能是当产品研发完成以后才能进行测试，这里其实就是涉及到了线程的互斥和同步，两个线程同一时间之间只能有一个线程进行操作，而测试又只能基于研发线程操作以后的结果才能进行测试，所以这里其实包含线程先后执行的约束问题。如果这里单单是线程的互斥，那么只能保证一个同一时间内只能有一个线程执行，然后另外一个线程可以继续执行，线程的互斥并不强调线程的先后执行顺序，但是线程的同步建立在线程互斥的基础上还要注重线程先后执行的约束问题，谁先执行，谁后执行，哪个线程依赖哪一个线程执行的结果，在这个结果上继续操作，这是线程同步关心的问题。

线程同步–条件变量

• 互斥锁的缺点是它只有两种状态：锁定和非锁定

• 条件变量通过允许线程阻塞和等待另外一个线程发送信号的方法弥补了互斥锁的不足。

• 条件变量内部是一个等待队列，放值等待的线程，线程在条件变量上等待和通知，互斥锁用来保护等待队列（对等待队列进行上锁），条件变量通常和互斥锁一起使用

  关于上边案例中的同步可以这样做：首先将测试线程放到等待队列中，此时由于产品没有完成不满足条件所以线程会处于阻塞状态。当研发线程将产品开发好以后，然后给等待队列中的线程发送信号，将测试线程唤醒然后将测试线程从等待队列中删除，然后由测试线程对产品进行操作。这里有一点需要注意：既然每一个线程都可以通过系统调用将线程本身放入到等待队列中进入等待，也就是说这个等待队列也是所有的线程都可以操作的共享资源。那么对共享资源的操作就要涉及到线程安全的问题，所以这里在操作等待队列的时候要使用互斥锁对共享资源进行保护。

• 条件变量允许线程等待特定条件发生，当条件不满足时，线程通常先进入阻塞状态，等待条件发生变化。一旦其他的某个线程改变了条件，可唤醒一个或者多个阻塞的线程。

• 具体的判断条件还需用户给出

• 条件变量的数据类型

  • pthread_cond_t

条件变量的初始化和销毁

#include <pthread.h>int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *attr);
int pthread_cond_destroy(pthread_cond_t *cond);/*
功能：pthread_cond_init	对条件变量进行初始化pthread_cond_destroy	销毁一个已经初始化的条件变量，释放其占用资源参数：pthread_cond_t *cond		指向要初始化的条件变量指针pthread_condattr_t *attr	指向条件变量属性对象指针，一般设置为NULL表示默认属性返回值：成功执行返回0，失败返回错误码
*/

条件变量等待操作

#include <pthread.h>int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_timewait(pthread_cond_t *cond, pthread_mutex_t *mutex, struct timespec *timeout);/*
功能：pthread_cond_wait	阻塞当前线程，直到另外一个线程通过调用pthread_cond_signal或者pthread_cond_broadcast唤醒它pthread_cond_timewait		作用与pthread_cond_wait类似，但允许指定一个超时时间，如果条件变量在超时时间内没有被信号量唤醒，则线程会自动解除阻塞状态互斥锁的作用是对条件变量的保护参数：pthread_cond_t *cond		指向要等待其改变的条件变量的指针pthread_mutex_t *mutex		指向与条件变量关联的互斥锁的指针。在调用pthread_cond_wait之前，线程必须已经锁定这个互斥锁struct timespec *timeout	指向一个timespec结构体，表示绝对时间点。如果在这个时间点之前条件变量没有被信号量唤醒，则函数返回返回值：成功时返回0；失败时返回错误码。如果线程被条件变量唤醒，则返回值大于0。如果因为超时而返回，则返回值等于0或ETIMEDOUT。其他错误情况下返回相应的错误码。
*/ 

条件变量通知操作

#include <pthread.h>int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_boardcast(pthread_cond_t *cond);/*功能：pthread_cond_signal	当条件满足的时候通知单个线程pthread_cond_boardcast		当条件满足的时候通知在等待队列中所有的线程当在一个等待队列中有若干个等待线程时，由于pthread_cond_signal函数只能够唤醒一个线程，而等待队列中的线程又遵循系统调度，所以不知道唤醒的线程是否是你需要的线程，所以当等待队列中的线程较多时直接使用pthread_cond_broadcast唤醒所有的线程参数：pthread_cond_t *cond		指向条件变量的指针返回值：成功执行返回0，否则返回错误码
*/

示例–条件变量的使用

#include "header.h"typedef struct
{int retult;int is_wait;pthread_mutex_t mutex;      //定义互斥锁类型pthread_cond_t cond;         //定义条件变量类型
}Result;void* cal_func(void *arg)
{Result *r = (Result*)arg;int i, sum = 0;//计算1-100的和然后存放到结构体中for(i = 1; i <= 100; i++)sum += i;       r->retult = sum;printf("[cal thread id]: %lx\n",pthread_self());pthread_mutex_lock(&r->mutex);while(!r->is_wait){//如果进入到这个循环中就说明等待线程还没有准备好，这时候就要释放互斥锁给等待的//线程有机会拿到互斥锁然后对共享资源进行修改，下次判断等待线程准备好就可以给等待//线程发送信号唤醒，让等待线程拿到计算结果并打印出来pthread_mutex_unlock(&r->mutex);    usleep(100);pthread_mutex_lock(&r->mutex);}pthread_mutex_unlock(&r->mutex);        //释放锁，上锁和释放锁是一一对应的pthread_cond_broadcast(&r->cond);       //唤醒所有在等待队列中的线程pthread_exit(NULL);
}void* get_func(void *arg)
{Result *r = (Result*)arg;//加锁，对共享资源进行保护pthread_mutex_lock(&r->mutex);r->is_wait = 1;pthread_cond_wait(&r->cond, &r->mutex);     //这里传互斥锁进入是为了保证等待队列这个共享资源的安全，实际上在内部做了多次加锁释放锁的操作pthread_mutex_unlock(&r->mutex);//通过阻塞等待计算线程将结果存放到结构体中，然后被另外一个线程使用broadcast唤醒printf("[get thread id:%lx] sum = %d\n",pthread_self(),r->retult);pthread_exit(NULL);
}int main(void)
{   int err = -1;pthread_t get,cal;Result r;memset(&r,'\0',sizeof(r));r.is_wait = 0;pthread_mutex_init(&r.mutex, NULL);      //以默认属性创建互斥锁pthread_cond_init(&r.cond, NULL);        //以默认属性创建条件变量//创建获取结果线程，计算结果从结构体中获取并打印出来if((err = pthread_create(&get, NULL, get_func, (void*)&r)) != 0){perror("pthread_create error");exit(EXIT_FAILURE);}//创建计算结果线程，然后将结果存放到结构体中if((err = pthread_create(&cal, NULL, cal_func, (void*)&r)) != 0){perror("pthread_create error");exit(EXIT_FAILURE);}//等待子线程退出pthread_join(get, NULL);pthread_join(cal, NULL);pthread_mutex_destroy(&r.mutex);      //销毁互斥锁pthread_cond_destroy(&r.cond);         //销毁条件变量return 0;
}

通过编译执行可以看到获取结果的线程拿到了计算线程的结果，这就是线程的同步。这里有两点比较重要：

1. 首先是is_wait变量的使用，上边这个代码中的is_wait的这个比较难以理解，详细解释一下：当cal的线程将计算结果存放到结构体后要确保get线程已经准备好了。is_wait这个变量就是用来检测get线程是否已经准备好了，由于is_wait属于一个共享资源，所以在操作的时候要使用互斥锁来进行保护共享资源的安全。在cal线程中判断is_wait的值来确保get线程已经准备好了，准备好以后就使用pthread_cond_broadcast唤醒get线程。而由于线程之间的执行顺序没办法保证，所以当cal线程先拿到锁以后检测到get线程还没有准备好的时候就要去释放锁，如果不释放锁，get线程在拿锁的时候就会阻塞，导致它一直修改不了is_wait的值，从而导致get线程陷入一个死循环。所以这里在判断get线程没有准备好的时候要立即释放锁，否则会造成死循环。在延时过后还要进行上锁的原因是可能这时候get线程还没有准备好还需要上锁进行保护。在退出循环后要释放锁，保证加锁释放锁是成对出现的。

2. 关于pthread_cond_wait()函数内部的实现机制并不是表面看到的调用阻塞线程，等待另一个线程去唤醒它。在它内部实际上有这些步骤：

   1. 在pthread_cong_wait函数之前已经上锁，所以在函数内部首先会释放锁，来保证其他线程能够拿到锁对共享资源进行操作。
   2. 然后再获取互斥锁来保证等待队列的操作，由于等待队列也属于共享资源，所以对它的操作也要加互斥锁，加锁以后将当前线程加入到等待队列中去。
   3. 然后释放锁，等待另外的线程调用pthread_cond_signal或pthread_cond_broadcast将当前线程唤醒。
   4. 当线程被唤醒后，此时在函数内部又会获取到互斥锁，这时候获取互斥锁的作用是将线程从等待队列中删除。
   5. 最后去执行调用pthread_cond_wait()这个线程内部的代码功能。

   所以程序中的第一句上锁和最后一句的释放锁并不是对应的，而是和pthread_cond_wait()函数内部的加锁释放锁对应的。

示例–使用两个线程对同一个文件进行读写

#include "header.h"typedef struct
{int fd;						//用于获取主线程打开的文件描述符int write_done;				//用于检测是否已经写入文件char str[32];				//用于写入文件的字符串char filename[12];			//用于存储文件名字的字符串pthread_mutex_t mutex;		//创建互斥锁类型pthread_cond_t cond;		//创建条件变量类型
}OperArg;void* read_func(void *arg)
{OperArg *r = (OperArg*)arg;char buffer[32];memset(buffer, '\0', sizeof(buffer));pthread_mutex_lock(&r->mutex);	//对共享资源进行保护while(!r->write_done){pthread_cond_wait(&r->cond, &r->mutex);		//等待另外的线程唤醒，否则阻塞}//从文件中读取lseek(r->fd, 0, SEEK_SET);		//由于写入的线程将文件指针偏移到末尾，所以读取的时候要重新指向if(read(r->fd, buffer, 32) < 0){pthread_mutex_unlock(&r->mutex);perror("read error");exit(EXIT_FAILURE);}printf("[read thread id:%lx] successfully read [%s] from the [%s]\n",pthread_self(),buffer,r->filename);pthread_mutex_unlock(&r->mutex);pthread_exit(NULL);
}void* write_func(void *arg)
{OperArg *r = (OperArg*)arg;int length = strlen(r->str);pthread_mutex_lock(&r->mutex);		//加锁对共享资源进行保护if(write(r->fd, r->str, length) != length){pthread_mutex_unlock(&r->mutex);		//程序异常退出释放锁防止造成死锁perror("write error");exit(EXIT_FAILURE);}printf("[write thread id:%lx] successfully write [%s] to the [%s]\n",pthread_self(),r->str,r->filename);r->write_done = 1;		//表明已经成功向文件写入pthread_cond_broadcast(&r->cond);		//文件写入以后就可以通知读取线程进行读取了pthread_mutex_unlock(&r->mutex);		//释放互斥锁给另外的线程拿到锁pthread_exit(NULL);
}int main(int argc, char **argv)
{	if(argc < 3){fprintf(stderr,"usage: %s [filepath] [string]\n",argv[0]);exit(EXIT_FAILURE);}int fd = -1;int err = -1;pthread_t read, write;OperArg r;memset(&r, '\0', sizeof(r));//以文件的拥有者、同组人有可读可写可执行的权限打开文件，如果文件不存在就创建文件fd = open(argv[1], O_RDWR | O_CREAT | O_TRUNC, S_IRWXU | S_IRWXG);if(fd < 0){perror("open file error");exit(EXIT_FAILURE);}r.fd = fd;r.write_done = 0;						//表示还未将字符串写入到文件strcpy(r.str, argv[2]);					//将从命令行传入的字符串赋值给strstrcpy(r.filename, argv[1]);			//将文件名传给结构体pthread_mutex_init(&r.mutex, NULL);		//初始化互斥锁pthread_cond_init(&r.cond, NULL);		//初始化条件变量//创建从文件中读取的线程if((err = pthread_create(&read, NULL, read_func, (void*)&r)) != 0){perror("pthread_create error");exit(EXIT_FAILURE);}	//创建向文件写入的线程if((err = pthread_create(&write, NULL, write_func, (void*)&r)) != 0){perror("pthread_create error");exit(EXIT_FAILURE);}//等待子线程退出并回收其资源pthread_join(read, NULL);pthread_join(write, NULL);//销毁互斥锁和条件变量pthread_mutex_destroy(&r.mutex);pthread_cond_destroy(&r.cond);close(fd);return 0;
}

通过编译执行可以看到成功创建文件并且对文件进行写入和读取，在这个代码中要write_func线程先对文件进行写入后read_func线程才能读取，若两个线程的执行顺序颠倒，read()函数若在读取普通文件时若文件为空，那么read()函数就什么都没有读到。注意这里的read()函数不会被阻塞直到write()写入，因为read()函数只有在读取管道文件和套接字文件的时候才会阻塞，所以这里的read()函数在文件为空或者读到文件末尾的时候会直接返回。这里对文件读取就是使用条件变量，当文件未写入的时候read_func线程调用pthread_cond_wait()函数一直阻塞，直到write_func()线程将内容写入到文件里后才调用pthread_cond_broadcast()函数将其唤醒后才能读取。

示例–一个读者一个写者使用条件变量来实现同步

#include "header.h"typedef struct
{int value;int rd_wait;							//判断读者线程运行条件int wr_wait;							//判断写者线程运行条件pthread_mutex_t rd_mutex;				pthread_mutex_t wr_mutex;					pthread_cond_t rd_cond;pthread_cond_t wr_cond;					//定义两种互斥锁和条件变量
}Storage;void* read_func(void *arg)
{Storage *s = (Storage*)arg;int i = 1;for(; i <= 10; i++){pthread_mutex_lock(&s->rd_mutex);			//对共享资源进行保护s->rd_wait = 1;								//表示读者线程已经准备好了pthread_cond_wait(&s->rd_cond, &s->rd_mutex);		//将自己加入等待队列等待另外一个线程唤醒printf("[read thread id:%lx] read [%d] from the structure\n",pthread_self(),s->value);pthread_mutex_unlock(&s->rd_mutex);pthread_mutex_lock(&s->wr_mutex);			//加锁，判断写者线程是否准备好了while(!s->wr_wait){pthread_mutex_unlock(&s->wr_mutex);		//如果没有准备好就要释放互斥锁给另外的线程修改wr_wait的机会sleep(1);pthread_mutex_lock(&s->wr_mutex);			//再次上锁判断wr_wait的值检测写者线程是否准备好了		}s->wr_wait = 0;pthread_cond_signal(&s->wr_cond);pthread_mutex_unlock(&s->wr_mutex);}pthread_exit(NULL);
}void* write_func(void *arg)
{Storage *s = (Storage*)arg;int i = 1;for(; i <= 10; i++){pthread_mutex_lock(&s->rd_mutex);		//对共享资源进行加锁，保证共享资源的安全性s->value = i + 10;						//写者线程赋值给结构体中的成员在读者线程中读取printf("[write thread id:%lx] write [%d] to the structure\n",pthread_self(),s->value);while(!s->rd_wait){//若此线程先运行，那么读者线程还没有修改rd_wait的值以此证明读者//线程已经准备好了，所以判断它没有准备好就要释放互斥锁让读者线程//能够拿到锁对rd_wait进行修改pthread_mutex_unlock(&s->rd_mutex);		sleep(1);pthread_mutex_lock(&s->rd_mutex);	//当读者线程修改完后还需要写着线程去判断rd_wait的值，所以还需要再次上锁	}//若退出循环则说明写着线程已经准备好了，把rd_wait初始化等待下一次循环判断s->rd_wait = 0;						//当写者线程准备好的时候就要给它发信号唤醒它pthread_cond_signal(&s->rd_cond);		pthread_mutex_unlock(&s->rd_mutex);//写者加锁来修改写者线程的条件变量pthread_mutex_lock(&s->wr_mutex);s->wr_wait = 1;					//表示写者线程已经准备好了pthread_cond_wait(&s->wr_cond, &s->wr_mutex);	//写者线程等待读者线程调用signal唤醒pthread_mutex_unlock(&s->wr_mutex);}pthread_exit(NULL);
}int main(void)
{int err = -1;pthread_t read, write;Storage s;memset(&s, '\0', sizeof(s));//初始化互斥锁和条件变量pthread_mutex_init(&s.rd_mutex, NULL);pthread_mutex_init(&s.wr_mutex, NULL);pthread_cond_init(&s.rd_cond, NULL);pthread_cond_init(&s.wr_cond, NULL);//创建读者线程用于从结构体中读取数据if((err = pthread_create(&read, NULL, read_func, (void*)&s)) != 0){perror("pthread_create error");exit(EXIT_FAILURE);}//创建写者线程用于向结构体中写入数据if((err = pthread_create(&write, NULL, write_func, (void*)&s)) != 0){perror("pthread_create error");exit(EXIT_FAILURE);}//等待子线程退出并回收其资源pthread_join(read, NULL);pthread_join(write, NULL);//销毁互斥锁和条件变量pthread_mutex_destroy(&s.rd_mutex);pthread_mutex_destroy(&s.wr_mutex);pthread_cond_destroy(&s.rd_cond);pthread_cond_destroy(&s.wr_cond);return 0;
}

通过编译结果可以看代码的执行流程是写者先运行向结构体中写入数据，然后读者线程从结构体中读取数据，然后两个线程交替运行。通过条件变量来控制两个线程的执行顺序来实现一个线程写入一个线程读取，使用互斥锁来保护共享资源的安全。