【Linux】二十八.线程篇五《Linux多线程编程：线程同步之条件变量》---详解

在上节内容中我们学习了线程篇五这节内容我们详细讲解线程同步.线程同步这是上节内容我们描写的是模拟抢票系统先看这段经典的多线程抢票逻辑虽然我们给它加上了互斥锁来保护数据但这就像给自习室门口安了一把只能单人进出的锁依然治不了“抢票狂魔”的问题——那个离锁最近、手速最快的线程总是能抢在别人前面一遍遍地把票扫光导致其他线程连汤都喝不上只能饿着肚子空转。这种现象在计算机里叫“饥饿问题”这个在上节内容中给大家讲过它本身没有语法错误但极其不合理就好比钥匙刚从墙上的挂钩摘下还没来得及给别人用同一个学生又顺手拿了进去反复循环只图自己方便。实际上真正合理的做法必须加上一条“排队”规则任何人用完钥匙挂回墙上后不能立刻再申请必须老老实实去队伍后面排队。把这个“按顺序办事”的规则落实到代码里来保证所有人都能公平地获取资源就是我们要引入的核心概念——线程同步。排队的本质就是在互斥保证安全的基础上赋予线程一个公平的申请与释放顺序这就是线程同步。它解决的不是数据对不对而是分配合理不合理的问题。正因如此互斥和同步总是成对出现光有互斥始终无法保证临界资源得到有序的合理访问。要说线程同步我们就要引入条件变量1.条件变量概念条件变量本质上是原生线程库提供的一个同步机制。它不是一个用来存数据的普通变量而是一个用来描述临界资源当前状态的信号器。它的作用类似于消息通知让一个线程可以挂起等待直到另一个线程改变了共享资源的状态并通过条件变量发出资源就绪的信号来唤醒它。我们在写抢票程序时很容易陷入一个误区如果发现票卖完了线程还是会拼命去申请锁、检查、再释放锁循环往复这个是错误的而是不应该再申请锁了而应该等有票了再申请。这就像中午你想去饮水机接水走过去一看桶空了你走开了过了两秒你又跑过去看又空了再走开……反复折腾。逻辑上你确实没做错但这种“轮询”不仅浪费了你的腿脚CPU计算资源还让本来想接水的人总要被你挡一下锁的争抢加剧。实际上更合理的做法是既然是没水了你就应该在饮水机旁边贴张纸条告诉送水工水来了叫我一声。然后你在旁边安心休息等送水工换好新桶敲敲你的肩膀唤醒你你再去接水。这里的“纸条”或“唤醒信号”就是用来描述临界资源状态水有没有的关键也就是我们要引入的条件变量。之所以之前我们一直在盲目地轮询申请锁根本原因就是线程之间缺乏沟通大家完全不清楚临界资源当前的状况只能靠撞大运般地去查。在张图左边负责“放苹果”右边负责“拿苹果”中间有个盘子临界资源。首先我们要明白放和拿的动作是连续的如果不加控制就会出现二义性——万一我这边还在放你那边就来拿了谁说得准这苹果到底算不算已经放好了所以锁的作用就是强制规定不管你是放还是拿必须先拿钥匙加锁把整个过程变成不可分割的“原子操作”保证盘子里的状态是明确的。但光有锁还不够。右边有个典型的轮询陷阱放苹果的人手脚麻利拿苹果的人磨磨蹭蹭。如果两人都是瞎子互相不知道对方动作放苹果的人就会反复加锁、一看苹果还在、解锁接着马上又加锁……他反反复复地瞎忙活虽然代码没错但极其浪费CPU而行动慢的拿苹果的人连抢到锁的机会都没有陷入了“饥饿”。这时图里红框标出的“铃铛”条件变量和旁边的“队列”就派上用场了。合理的做法是放苹果的人加好锁、放下苹果解锁后立刻敲一下铃铛然后去旁边挂起睡觉。这个铃铛的作用就是描述当前盘子临界资源有没有苹果的状态。另一边拿苹果的人并不是盲目地去抢锁而是先在队列等待队列里好好排队等着。一旦听到铃铛响他知道有苹果了就会从队列里被唤醒从容地去加锁、拿走苹果并敲铃铛通知对方。当⼀个线程互斥地访问某个变量时它可能发现在其它线程改变状态之前它什么也做不了。例如⼀个线程访问队列时发现队列为空它只能等待只到其它线程将⼀个节点添加到队列中。这种情况就需要⽤到条件变量。2.同步概念与竞态条件同步在保证数据安全的前提下让线程能够按照某种特定的顺序访问临界资源从⽽有效避免饥饿问题叫做同步竞态条件因为时序问题⽽导致程序异常我们称之为竞态条件。在线程场景下这种问题也不难理解详细内容上一节我都讲过了3.条件变量函数函数接口介绍初始化int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr);参数cond要初始化的条件变量attrNULL销毁int pthread_cond_destroy(pthread_cond_t *cond)等待条件满足pthread_cond_wait(cond, mutex)使调用线程在指定条件变量cond上阻塞并挂入其等待队列。关键机制阻塞挂起的瞬间系统会原子性地释放互斥锁mutex避免死锁当被唤醒后系统又会原子性地重新获取该锁之后线程才得以继续执行后文生产者消费者模型将详细拆解。pthread_cond_timedwait(cond, mutex, abstime)接口与wait完全一致仅多出超时控制参数abstime绝对时间。若指定时间内未收到唤醒信号线程将自动解除阻塞。主要用于规避无限等待的风险实际应用场景相对有限。参数cond要在这个条件变量上等待mutex互斥量后⾯详细解释唤醒等待pthread_cond_signal(cond)唤醒在指定条件变量cond的等待队列中阻塞的第一个线程。常用于资源仅能满足单个消费者的情况。pthread_cond_broadcast(cond)唤醒所有当前在该条件变量cond下阻塞等待的线程。多用于资源状态发生重大变化如一次性释放了多个资源需要通知所有等待者重新竞争应用场景相对较少。4.为什么 pthread_ cond_ wait 需要互斥量?条件等待是线程间同步的⼀种⼿段如果只有⼀个线程条件不满⾜⼀直等下去都不会满⾜所以必须要有⼀个线程通过某些操作改变共享变量使原先不满⾜的条件变得满⾜并且友好的通知等待在条件变量上的线程。条件不会⽆缘⽆故的突然变得满⾜了必然会牵扯到共享数据的变化。所以⼀定要⽤互斥锁来保护。没有互斥锁就⽆法安全的获取和修改共享数据。按照上⾯的说法我们设计出如下的代码先上锁发现条件不满⾜解锁然后等待在条件变量上不就⾏了如下代码:// 错误示范存在致命的信号丢失风险 pthread_mutex_lock(mutex); while (condition_is_false) { pthread_mutex_unlock(mutex); // 步骤1主动解锁 //解锁之后等待之前条件可能已经满足信号已经发出但是该信号可能被错过 pthread_cond_wait(cond); // 步骤2此时才开始去排队等待 pthread_mutex_lock(mutex); // 步骤3醒来后自己抢锁 } pthread_mutex_unlock(mutex);由于解锁和等待不是原⼦操作。调⽤解锁之后 pthread_cond_wait 之前如果已经有其他线程获取到互斥量摒弃条件满⾜发送了信号那么 pthread_cond_wait 将错过这个信号可能会导致线程永远阻塞在这个 pthread_cond_wait 。所以解锁和等待必须是⼀个原⼦操作。int pthread_cond_wait(pthread_cond_ t *cond,pthread_mutex_ t *mutex); 进⼊该函数后会去看条件量等于0不等于就把互斥量变成1直到cond_ wait返回把条件量改成1把互斥量恢复成原样。5.条件变量使用规范1.等待条件代码pthread_mutex_lock(mutex); // 必须使用 while 循环而不是 if来防止假的唤醒 while (条件为假) { pthread_cond_wait(cond, mutex); } // 此时条件为真可以安全地操作共享资源 修改条件; pthread_mutex_unlock(mutex);在这里为什么while是必须的因为即使被signal唤醒了也可能因为系统调度或竞争导致共享变量状态在被唤醒的瞬间又被其他线程抢走了。使用while可以确保线程醒来后重新检查一次如果条件又不满足了就继续回去等。给条件发送信号代码pthread_mutex_lock(mutex); // 修改共享变量使等待的条件变为真 设置条件为真; // 发出信号唤醒在 cond 上等待的一个线程 pthread_cond_signal(cond); pthread_mutex_unlock(mutex);修改条件和发送信号必须在同一个互斥锁的保护下完成。这样可以保证通知者写数据时等待者不会同时读数据数据一致且信号不会被错过。代码如下#include iostream #include string #include unistd.h #include pthread.h #define TNUM 4 typedef void (*func_t)(const std::string name,pthread_mutex_t *pmtx, pthread_cond_t *pcond); volatile bool quit false; // pthread_cond_t cond PTHREAD_COND_INITIALIZER; // pthread_mutex_t mtx PTHREAD_MUTEX_INITIALIZER; class ThreadData { public: ThreadData(const std::string name, func_t func, pthread_mutex_t *pmtx, pthread_cond_t *pcond) :name_(name), func_(func), pmtx_(pmtx), pcond_(pcond) {} public: std::string name_; func_t func_; pthread_mutex_t *pmtx_; pthread_cond_t *pcond_; }; void func1(const std::string name, pthread_mutex_t *pmtx, pthread_cond_t *pcond) { while(!quit) { // wait一定要在加锁和解锁之间进行wait pthread_mutex_lock(pmtx); pthread_cond_wait(pcond, pmtx); //默认该线程在执行时wait代码被执行当前线程会被立即被阻塞 std::cout name running -- 播放 std::endl; pthread_mutex_unlock(pmtx); } } void func2(const std::string name,pthread_mutex_t *pmtx, pthread_cond_t *pcond) { while(!quit) { pthread_mutex_lock(pmtx); pthread_cond_wait(pcond, pmtx); if(!quit) std::cout name running -- 下载 std::endl; pthread_mutex_unlock(pmtx); } } void func3(const std::string name,pthread_mutex_t *pmtx, pthread_cond_t *pcond) { while(!quit) { pthread_mutex_lock(pmtx); pthread_cond_wait(pcond, pmtx); std::cout name running -- 刷新 std::endl; pthread_mutex_unlock(pmtx); } } void func4(const std::string name,pthread_mutex_t *pmtx, pthread_cond_t *pcond) { while(!quit) { pthread_mutex_lock(pmtx); pthread_cond_wait(pcond, pmtx); std::cout name running -- 扫码用户信息 std::endl; pthread_mutex_unlock(pmtx); } } void *Entry(void *args) { ThreadData *td (ThreadData*)args; // td在每一个线程自己私有的栈空间中保存 td-func_(td-name_, td-pmtx_, td-pcond_); // 它是一个函数调用完成就要返回 delete td; return nullptr; } int main() { pthread_mutex_t mtx; pthread_cond_t cond; pthread_mutex_init(mtx, nullptr); pthread_cond_init(cond, nullptr); pthread_t tids[TNUM]; func_t funcs[TNUM] {func1, func2, func3, func4}; for (int i 0; i TNUM; i) { std::string name Thread ; name std::to_string(i1); ThreadData *td new ThreadData(name, funcs[i], mtx, cond); pthread_create(tids i, nullptr, Entry, (void*)td); } sleep(5); // ctrl new thread int cnt 10; while(cnt) { std::cout resume thread run code .... cnt-- std::endl; pthread_cond_signal(cond); // pthread_cond_broadcast(cond); sleep(1); } std::cout ctrl done std::endl; quit true; pthread_cond_broadcast(cond); for(int i 0; i TNUM; i) { pthread_join(tids[i], nullptr); std::cout thread: tids[i] quit std::endl; } pthread_mutex_destroy(mtx); pthread_cond_destroy(cond); return 0; }这段代码主要是利用互斥锁和条件变量实现线程的休眠与唤醒控制。程序启动之后主线程先创建四个子线程各个子线程运行后就调用 pthread_cond_wait 进入阻塞等待状态此时线程就挂在条件变量的等待队列上。主线程等待几秒之后循环调用 pthread_cond_signal 每次唤醒一个子线程被唤醒的线程拿到锁就执行对应的业务打印执行完解锁后又会再次进入等待。等唤醒循环结束把全局退出标记 quit 置为 true再通过 pthread_cond_broadcast 唤醒所有线程让各个子线程跳出循环正常结束最后主线程 join 回收线程资源销毁锁与条件变量完成整个程序的收尾。6. 条件变量的封装基于上⾯的基本认识我们已经知道条件变量如何使⽤虽然细节需要后⾯再来进⾏解释但这⾥可以做⼀下基本的封装#pragma once #include iostream #include string #include pthread.h #include Lock.hpp namespace CondModule { using namespace LockModule; class Cond { public: Cond() { int n pthread_cond_init(_cond, nullptr); (void)n; // 实际工程中此处应做 n ! 0 的异常判断或日志记录 } void Wait(Mutex *mutex) { // 传入底层互斥锁的指针 int n pthread_cond_wait(_cond, mutex-GetMutexOriginal()); (void)n; } void Notify() { int n pthread_cond_signal(_cond); (void)n; } void NotifyAll() { int n pthread_cond_broadcast(_cond); (void)n; } ~Cond() { int n pthread_cond_destroy(_cond); (void)n; // 实际工程中此处应做 n ! 0 的异常判断或日志记录 } private: pthread_cond_t _cond; }; }原来的 C 语言做法你必须在主函数里声明pthread_cond_t cond;然后调用pthread_cond_init()初始化最后程序结束前还得记得调用pthread_cond_destroy()销毁。一旦忘了销毁就会造成内存泄露。这个封装的思路把底层的 C 变量pthread_cond_t _cond包进 C 类里。构造函数里自动完成初始化析构函数里自动完成销毁。这就变成了只要这个类对象创建了条件变量就活了对象出了作用域被销毁了条件变量也就自动清理了。同时你不需要再手动去操心初始化和销毁代码放在哪了。