文章目录1. 为什么有了互斥我还需要同步利用厕所坑位抢占理解互斥和同步核心工具-条件变量2. 生产者-消费者模型321原则怎么理解模型3. 基于阻塞队列实现生产者-消费者模型封装version1代码接口细节version2代码4. 基于环形队列实现生产者-消费者模型有了条件变量为什么还要信号量信号量的封装环形队列的封装5. 信号量 VS 条件变量1. 为什么有了互斥我还需要同步利用厕所坑位抢占理解互斥和同步对于互斥相当于手里有了一把锁而互斥的本质是解决竞争问题保证同一时刻只有一个线程能访问临界资源例如厕所只有一个坑位你进去之后锁上门别人就进不来这样加锁处理是为了数据安全但是只有互斥是不够的场景还是厕所如果没有同步A进去了B在门口等。A出来之后B刚要抢结果A动作快又抢到了锁进去了单纯出来透口气。B抢不到一直在门口盲目空转或者频繁申请锁失败更糟糕的情况生产者-消费者仓库满了生产者抢到了锁发现满了释放锁生产者又抢到了锁发现仓库还是满的释放锁生产者陷入死循环CPU飙升而消费者却抢不到锁A和生产者有错吗其实并没有遵守了互斥的规则但是这样不合理所以引入同步解决上述问题同步的本质解决时序问题让现场按照一定的顺序协同工作“我干完了通知你你干完了通知我”同时解锁之后排队去不能立刻再申请锁核心等待唤醒总结什么是线程同步和互斥的区别互斥为了保护共享资源不被并发破坏保证数据安全同步为了协调线程执行的先后顺序避免线程饥饿同步通常建立在互斥只是检查条件时需要保护数据核心工具-条件变量为了实现同步OS提供了条件变量它不是锁而是一个等待队列主要做两件事wait等待条件不满足在等待队列中排队等待释放CPUsignal/notify唤醒条件满足被别人唤醒2. 生产者-消费者模型321原则3种关系生产者VS生产者互斥争夺写指针消费者VS消费者互斥争夺读指针生产者VS消费者互斥保护队列 同步满的时候等消费空的时候等生产2个角色生产者线程、消费者线程1个交易场所阻塞队列怎么理解模型举个奶茶店出杯的例子阻塞队列出杯台最多放3杯奶茶生产者做奶茶的店员把杯子放到出杯台消费者取奶茶的配送员从杯台拿走杯子只要多个线程并发读写同一份共享资源并且这个读写不是原子的就必须互斥来维护不变量队列中的不变量容量任何一个槽位同一时刻只能被一个人修改占用状态读/写的移动顺序不能乱、不能重复生产者和生产者互斥两个店员同时出杯争夺槽位必须互斥否则可能会数据覆盖消费者和消费者互斥两个配送员同时取杯争夺杯子必须互斥否则可能逻辑错误一杯被拿走两次生产者和消费者互斥保护队列这个共享结构如果没有互斥P在写入C同时读取读到了P写一半的数据生产者和消费者同步互斥只保证别同时改不保证有货/有位置假如杯台满生产者不断过来检查不断跑空—应该等有空位再来假如杯台空消费者不断过来检查不断跑空—应该等有余量再来所以需要条件变量来做同步3. 基于阻塞队列实现生产者-消费者模型封装version1代码#pragmaonce#includepthread.h#includeiostream#includequeue#includeMutex.hpp#includeCond.hpp// version1templatetypenameTclassBlockQueue_v1{private:std::queueT_q;int_capacity;pthread_mutex_t _mutex;pthread_cond_t _productor_cond;// 队列满生产者在这里等pthread_cond_t _consumer_cond;// 队列空消费者在这里等int_cwait_num;int_pwait_num;boolIsFull(){return_q.size()_capacity;}boolIsEmpty(){return_q.empty();}public:BlockQueue_v1(intcap10):_capacity(cap),_cwait_num(0),_pwait_num(0){// 初始化锁和条件变量// nullptr表示默认属性pthread_mutex_init(_mutex,nullptr);// 条件变量是等待队列 不是锁pthread_cond_init(_productor_cond,nullptr);pthread_cond_init(_consumer_cond,nullptr);}// 生产者写数据入队列// 输入型参数voidEqueue(constTin){// 1. 加锁pthread_mutex_lock(_mutex);// 2. 检查条件IsFull// 为什么用while?// POSIX标准允许OS在无信号时意外唤醒线程必须通过循环进行二次状态校验// wait返回意味着重新持有了锁但不代表条件依然满足// 在收到通知到抢到锁的【时间窗口】内空位可能已被其他生产者抢先占满// 若用 if当前线程会无视满队列直接 push导致数据覆盖或溢出while(IsFull()){std::cout队列满了生产者开始等待...\n;_pwait_num;// 3. 等待// 函数底层做三件事// 1 解锁unlock允许别人进来否则会死锁// 2 线程挂起-进入_productor_cond的等待队列// 3 被唤醒后-重新抢锁lock抢不到就阻塞在这里pthread_cond_wait(_productor_cond,_mutex);_pwait_num--;std::cout生产者被唤醒准备生产...\n;}// 4. 生产// 到这里说明IsFull为0且我持有锁_q.push(in);// 5. 唤醒消费者// 只有当_cwait_num0时再去唤醒否则浪费资源if(_cwait_num0){std::cout通知消费者来取货\n;// siganl只唤醒一个// broadcast全部唤醒pthread_cond_signal(_consumer_cond);}// 6. 解锁pthread_mutex_unlock(_mutex);}// 消费者读数据出队列// 输出型参数voidDequeue(T*out){pthread_mutex_lock(_mutex);while(IsEmpty()){std::cout队列空了消费者开始等待...\n;_cwait_num;pthread_cond_wait(_consumer_cond,_mutex);_cwait_num--;std::cout消费者被唤醒...\n;}*out_q.front();_q.pop();if(_pwait_num0){std::cout通知生产者补货...\n;pthread_cond_signal(_productor_cond);}pthread_mutex_unlock(_mutex);}~BlockQueue_v1(){pthread_mutex_destroy(_mutex);pthread_cond_destroy(_productor_cond);pthread_cond_destroy(_consumer_cond);}};接口细节pthread_cond_wait(cond, mutex)一定要传锁为什么我先解锁再等待不行吗不行如果我在unlock和wait之间OS 调度切换了线程此时另一个线程发送signal。因为我已经解锁信号发送成功但我还没睡。等我回来执行wait时就错过了那个信号导致永久阻塞内核态操作wait的本质是将当前线程放入等待队列并原子性地释放锁。这两个动作必须在内核中一口气完成无法由用户代码分开完成调用后的状态调用前持有锁调用中睡眠释放锁允许其他线程进入临界区操作返回后持有锁函数内部自动帮忙抢回锁pthread_cond_singal和pthread_cond_broadcastsignal唤醒一个等待线程适合一对一broadcast唤醒所有等待线程适合读写锁写完之后所有人都能读坑如果用broadcast唤醒了10个消费者但只有1个数据10个人抢锁一个人吃到了肉剩下9个跑空只能回去继续睡浪费CPUversion2利用RAII自动释放资源代码封装锁#includeiostream#includepthread.hclassMutex{public:Mutex(){pthread_mutex_init(_lock,nullptr);}voidLock(){pthread_mutex_lock(_lock);}voidUnlock(){pthread_mutex_unlock(_lock);}~Mutex(){pthread_mutex_destroy(_lock);}// 关键条件变量需要锁要传指针pthread_mutex_t*GetLock(){return_lock;}private:pthread_mutex_t _lock;};classGuardLock{public:GuardLock(Mutexm):_mtx(m){_mtx.Lock();}~GuardLock(){_mtx.Unlock();}private:Mutex_mtx;};封装条件变量#pragmaonce#includepthread.h#includeMutex.hppclassCond{public:Cond(){pthread_cond_init(_cond,nullptr);}~Cond(){pthread_cond_destroy(_cond);}// wait需要一把锁传入封装的Mutex对象voidWait(Mutexm){// 调用底层的pthread_cond_Wait// 需要原生指针pthread_cond_wait(_cond,m.GetLock());}voidNotify(){pthread_cond_signal(_cond);}voidNotifyAll(){pthread_cond_broadcast(_cond);}private:pthread_cond_t _cond;};实现阻塞队列templatetypenameTclassBlockQueue_v2{private:int_capacity;std::queueT_q;Mutex _mutex;Cond _productor_cond;Cond _consumer_cond;int_pwait_num;int_cwait_num;boolIsFull(){return_q.size()_capacity;}boolIsEmpty(){return_q.empty();}public:// Mutex和Cond会自动调用构造初始化BlockQueue_v2(intcap10):_capacity(cap),_pwait_num(0),_cwait_num(0){}// 生产者voidEqueue(constTin){GuardLockguardLock(_mutex);while(IsFull()){std::cout生产者进入等待...\n;_pwait_num;_productor_cond.Wait(_mutex);_pwait_num--;std::cout生产者被唤醒...\n;}_q.push(in);if(_cwait_num0){std::cout唤醒消费者\n;_consumer_cond.Notify();}}// guardLock离开作用域自动析构解锁voidDequeue(T*out){GuardLockguardLock(_mutex);while(IsEmpty()){std::cout消费者进入等待...\n;_cwait_num;_consumer_cond.Wait(_mutex);_cwait_num--;std::cout消费者被唤醒...\n;}*out_q.front();_q.pop();if(_pwait_num0){std::cout唤醒生产者\n;_productor_cond.Notify();}}// 成员变量会自用调用析构销毁~BlockQueue_v2(){}};4. 基于环形队列实现生产者-消费者模型有了条件变量为什么还要信号量条件变量本质是同步机制解决了时许问题逻辑是条件不满足我就睡条件满足你叫醒我它不保存状态不知道资源具体有多少需要结合mutex和外部变量来判断信号量本质是资源计数器可以解决数量问题逻辑是票还有没有有我就拿走一张进门没有我就在门口等是对资源的一种预定机制使用条件变量实现环形队列需要加锁while(queue.full()) wait()检查满不满放数据signal/broadcast通知消费者解锁每次操作都要抢锁生产者和消费者哪怕互不干扰队列既没满也没空也要竞争同意把锁并发度低信号量解法把环形队列视为两种资源空闲空间资源_spacesem生产者关心初始值为队列容量N数据资源_datasem消费者关心初始值为0生产者只消耗空间增加数据消费者只消耗数据增加空间如果不满也不空生产者和消费者位于不同位置完全可以并行执行不互斥只有在相同位置-环形队列相同位置为空or为满只有这两种情况-互斥所以环形队列有三大约束生产者VS消费者队列为空消费者不能读队列为满生产者不能写入生产者VS生产者多个生产者不能同时写入同一个位置消费者VS消费者多个消费者不能同时读取同一个位置信号量的封装我们使用POSIX标准的信号量在semaphore.h库中常用接口初始化信号量#includesemaphore.hintsem_init(sem_t*sem,intpshared,unsignedintvalue);// 参数// pshared0表示线程间共享非0表示进程间共享// value信号量初始值销毁intsem_destroy(sem_t*sem);等待// 等待信号量将信号量的值-1intsem_Wait(sem_t*sme);// P操作发布// 发布信号量表示资源使用完毕可以归还资源了信号量值1intsem_post(sem_t*sem);// Vc操作#pragmaonce#includesemaphore.hnamespaceSemModule{intdefalutSemVal1;classSem{public:Sem(intvaluedefalutSemVal):_init_value(value){// 参数2为0表示线程间共享非0表示进程间共享intn::sem_init(_sem,0,_init_value);}// P操作proberen申请资源// 如果计数器 0原子-1返回// 如果计数器 0阻塞等待知道被唤醒voidP(){intn::sem_wait(_sem);}// V操作verhogen归还资源// 计数器原子1如果有线程在等待该资源唤醒它voidV(){intn::sem_post(_sem);}~Sem(){intn::sem_destroy(_sem);}private:sem_t _sem;int_init_value;};}环形队列的封装#pragmaonce#includeiostream#includevector#includepthread.h#includeSem.hpp#includeMutex.hppnamespaceRingBuffer{usingnamespaceSemModule;templatetypenameTclassRingBuff{private:std::vectorT_ring;int_capacity;// 生产者和消费者的索引临界资源需要保护int_p_idx;int_c_idx;// 同步机制信号量Sem _dataSem;Sem _spaceSem;// 互斥机制锁Mutex _p_lock;Mutex _c_lock;public:RingBuff(intcap):_capacity(cap),_ring(cap),_p_idx(0),_c_idx(0),_dataSem(0),_spaceSem(cap){}// 生产者voidEnqueue(constTin){// 必须先P再Lock// 先申请资源。如果先锁再P且资源不足生产者抱着锁挂起// 消费者向消费腾出空间却拿不到锁-死锁// 1. 申请空间资源-1_spaceSem.P();{// 2. 只有申请到资源后才竞争锁保护临界区GuardLockguardLock(_p_lock);_ring[_p_idx]in;_p_idx%_capacity;}// 3. 生产完毕增加数据资源唤醒消费者1_dataSem.V();}// 消费者voidDequeue(T*out){// 1. 申请资源-1 没数据就阻塞在这不占锁_dataSem.P();{// 2. 加锁访问GuardLockguardLock(_c_lock);*out_ring[_c_idx];_c_idx%_capacity;}// 3. 消费完毕归还空间资源唤醒生产者1_spaceSem.V();}~RingBuff(){}};}5. 信号量 VS 条件变量特性信号量 (Semaphore)条件变量 (Cond Var)本质资源计数器(Stateful)通知机制(Stateless)状态保存保存状态。如果P操作前已经V了信号量1P操作不会阻塞不保存。如果 wait 前已经 signal 了信号丢失wait 会死等互斥需求可以在无锁外部无Mutex情况下使用自身保证原子性必须配合 Mutex 使用适用场景这里的环形队列确定数量的资源、控制并发数复杂的同步逻辑如队列既不满也不空或者需要判断特定状态