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

🕗 发布于 2024-10-10 10:29 linux Linux系统编程

文章目录

线程
- 线程的同步和互斥
- - 线程同步--条件变量

线程

线程的同步和互斥

线程同步–条件变量

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

什么是线程同步

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

线程同步–条件变量

互斥锁的缺点是它只有两种状态：锁定和非锁定
条件变量通过允许线程阻塞和等待另外一个线程发送信号的方法弥补了互斥锁的不足。
条件变量内部是一个等待队列，放值等待的线程，线程在条件变量上等待和通知，互斥锁用来保护等待队列（对等待队列进行上锁），条件变量通常和互斥锁一起使用

关于上边案例中的同步可以这样做：首先将测试线程放到等待队列中，此时由于产品没有完成不满足条件所以线程会处于阻塞状态。当研发线程将产品开发好以后，然后给等待队列中的线程发送信号，将测试线程唤醒然后将测试线程从等待队列中删除，然后由测试线程对产品进行操作。这里有一点需要注意：既然每一个线程都可以通过系统调用将线程本身放入到等待队列中进入等待，也就是说这个等待队列也是所有的线程都可以操作的共享资源。那么对共享资源的操作就要涉及到线程安全的问题，所以这里在操作等待队列的时候要使用互斥锁对共享资源进行保护。
条件变量允许线程等待特定条件发生，当条件不满足时，线程通常先进入阻塞状态，等待条件发生变化。一旦其他的某个线程改变了条件，可唤醒一个或者多个阻塞的线程。
具体的判断条件还需用户给出
条件变量的数据类型
- 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;
}

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

首先是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线程还没有准备好还需要上锁进行保护。在退出循环后要释放锁，保证加锁释放锁是成对出现的。
关于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]);//将从命令行传入的字符串赋值给str
strcpy(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;
}

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

原文地址：https://blog.csdn.net/m0_52867657/article/details/142797002

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：12- Cesium 中动态处理与四个圆形渐变过渡材质相关的属性
下一篇：集合论（ZFC）之极限序数（Limit Ordinal）

开发指南072-图片热点
/处理跳转： area.getAttribute('href');平台支持使用图像导航界面，例如展示如下一张图，用户点击对应位置触发对应动作。热点数据通过后台接口获取（注意处理权限，没有权限的热点不生
阅读更多2024-10-13
使用机器学习边缘设备的快速目标检测
这项机器学习研究探讨了一种低成本的边缘设备，该设备与具有计算机视觉功能的嵌入式系统集成，以提高目标检测和分类的推理时间和精度。研究的主要目标是减少推理时间并降低功耗，以支持一个竞技型类人机器人的嵌入式
阅读更多2024-10-13
【Windows】【DevOps】Windows Server 2022 安装ansible，基于powershell实现远程自动化运维部署入门到放弃！
文件URL：https://www.python.org/ftp/python/3.13.0/python-3.13.0-amd64.exe。直接拿linux主机测试ansible连接windows
阅读更多2024-10-13
C# 中循环的应用说明
一循环的概念说明二、循环类型三、循环控制语句四、无限循环
阅读更多2024-10-13
Linux `vmstat` 命令详解
vmstat（Virtual Memory Statistics）是 Linux 系统中的一个监控工具，用于报告系统的虚拟内存、进程、CPU 活动等信息。它能帮助用户了解系统的整体性能状况，尤其是内存
阅读更多2024-10-13
Linux下多任务编程（网络编程2）
本文介绍解决accpet和recv相互阻塞的问题，可以用多线程并发外也可以用epoll I/O多路复用的方式解决。
阅读更多2024-10-13
[单master节点k8s部署]37.微服务（一)springCloud 微服务
微服务架构的一个重要特点是，它与开发中使用的具体或无关。每个微服务都可以使用最适合其功能需求的语言或技术来实现。例如，一个微服务可以用Java编写，另一个微服务可以用Python、Go、Node.js
阅读更多2024-10-13
Zynq(3)使用外设MIO/EMIO
使用MIO/EMIO实现流水灯，着重介绍Zynq IP核的配置，解读vitis中的c语言程序，介绍MIO与EMIO的区别。
阅读更多2024-10-13
笔试算法总结
思路很简单，但是当时做题提交的时候，通过率总是18%。不知道为啥，后面我改成了Long类型，然后就通过了全部用例。（易错1：第一次提交没考虑0的情况）使用 StringBuilder 模拟栈的行为，通
阅读更多2024-10-13
快速学习一个算法，Transformer模型架构
它的主要思想是在同一时间通过多个独立的注意力头（Attention Head）来关注序列中不同部分的信息，然后将这些信息综合起来，生成更丰富的表示。自注意力机制的目的是对输入序列中的每个元素计算一个输
阅读更多2024-10-13

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

文章目录

线程

线程的同步和互斥

线程同步–条件变量

什么是线程同步

示例–条件变量的使用

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

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

相关文章