1. 前言一个进程内部的多个线程当中因为所有的线程共享其地址空间并且进程资源大部分都会被线程共享那么当多个线程同时访问同一块资源的时候就会造成重入的现象。并且我们在前面学习线程的概念及其控制时对于代码执行结果的打印会出现如下的状况void *Print(void *args) { string name static_castconst char *(args); while(true) { cout 我是新线程: name endl; sleep(1); } return nullptr; } int main() { pthread_t tids[4]; for(int i 0; i 4; i) //创建线程打印线程名 { char *name new char[64]; snprintf(name,64,thread-%d,i1); pthread_create(tids i ,nullptr,Print,(void *)name); } for(int i 0; i 4; i) //回收线程 { pthread_join(tids[i],nullptr); } return 0; }按理来说应该时主线程打印再子线程打印这样交替但是这里会出现内容错乱的情况我们在前面讲解线程概念及其控制时讲过这是因为 Linux 采用抢占式线程调度机制pthread_create 创建线程时只会将线程加入内核就绪队列不会强制新线程马上执行主线程循环创建完所有线程后全部子线程都会处于就绪状态等待 CPU 时间片系统调度器会随机挑选就绪线程分配执行时间片不存在先创建先运行的规则要想解决这个问题就需要我们即将要学习的内容线程的同步与互斥。我们用一个例子来引入线程互斥问题#include iostream #include vector #include unistd.h using namespace std; #include Thread.hpp int tickets 10000; void GetTicket() { char name[64]; pthread_getname_np(pthread_self(),name,sizeof(name)); while(1) { if(tickets 0) { usleep(1000); printf(%s sells ticket:%d\n,name,tickets); tickets--; } else { break; } } } int main() { ThreadModule::Thread t1(GetTicket); ThreadModule::Thread t2(GetTicket); ThreadModule::Thread t3(GetTicket); ThreadModule::Thread t4(GetTicket); t1.Start(); t2.Start(); t3.Start(); t4.Start(); t1.Join(); t2.Join(); t3.Join(); t4.Join(); }我想使用这段代码来模拟多用户进行抢票的场景票数固定为一万张一共四个线程代表四个用户这段代码的执行结果是这样的大家会发现我们明明只有一万张票但当卖票的时候竟然卖到了负数张并且此时的负数还是两个相同的 -1 。分析一下我们的代码我们发现多线程访问共同资源的时候对 tickets 访问最多的是判断和对 tickets-- 的时候而上述问题本质是在判断这里导致的大家要想明白一件事情当我们执行判断语句时对变量进行条件判断对于 CPU 来说是不是运算答案是算这种运算我们称之为逻辑运算。而因为我们定义的变量 tickets 一开始是存储在内存当中的既然要进行运算就一定要将 tickets 变量读取到 CPU 当中。当判断 tickets 的数据确实大于零就会执行 tickets-- 的操作此时在CPU中对 tickets 变量的操作也会导致内存中变量的改变。对于这个过程重点是都是在一个线程内部做的。但是真实情况是不止一个线程的像我们上述代码中有四个线程其实都会做这样的工作到底是谁做这是由操作系统的调度器决定的。那么就会出现这样一种情况假如现在票还剩最后一张即 tickets 1这时候有一个线程名叫线程 1 已经把 tickets 变量读取到 CPU 中刚进入 CPU 判断完此时的 tickets 确实大于 0 即线程 1 进入了 if 条件语句线程 1 就被切换走了而此时 CPU 内的所有内容都属于线程 1 的上下文数据。我们前面讲过每个线程虽然共享进程资源但都有其独立的内容最主要的三个就是线程 ID 栈 上下文数据。那么此时线程 1 就会把它自己的上下文数据给带走并且线程 1 记住了现在的 tickets 1。然后操作系统通过调度器开始调度线程 2 因为刚刚线程 1 只是进入了 if 判断语句但还没有对 tickets 的值进行修改所以此时在内存中 tickets 照样还是 1 那线程 2 就会重复这个过程进入 if 判断语句此时线程 2 也被切换了。接下来是调度线程 3 、线程4........那么当下一次又轮到线程 1 的时候就会执行 tickets-- 的操作同样的 线程 2 、线程 3、线程4........ 这就是负数会出现的原因。而这里的本质是因为多线程并发访问多线程同时访问和修改共享变量没有同步机制保护并且因为if(tickets 0)和tickets--都不具有原子性即必须被执行无法中断的操作会被调度器干扰。解决这个问题的标准方法是对 tickets 这个共享资源加以保护而我们之前说过临界区是指多个线程 / 进程会同时访问、修改的共享资源代码段那如果我们只允许一个线程执行不允许多个线程同时执行就可以通过保护临界区来变相保护共享资源。2. 线程互斥2.1 互斥锁我们前面提到了多线程保护共享资源本质是把访问临界资源的代码保护起来这个保护的方式就是使用互斥锁。互斥锁Mutex是多线程里用来保护临界区的同步机制。 作用是保证同一时刻只有一个线程能进入临界区让临界区代码变成原子操作避免数据混乱。我们在使用互斥锁之前必须先定义锁一般有两种方式第一种是直接调用函数动态初始化pthread_mutex_t mtx; pthread_mutex_init(mtx, nullptr);第二种是静态初始化pthread_mutex_t mtx PTHREAD_MUTEX_INITIALIZER;这里的 pthread_mutex_t 本质上是一个结构体类型结构体变量你可以把它理解成一把锁的 “本体”用来描述这把锁的状态它在Linux内核中是以一个结构体的形式描述的typedef struct { // 锁的状态、所有者、等待队列... } pthread_mutex_t;而 PTHREAD_MUTEX_INITIALIZER是一个宏宏定义作用是直接给锁做 “默认初始化”就是把这把锁设置成未上锁、可用、默认属性。静态初始化最简单、最安全、最常用。锁初始化完成之后就需要上锁我们会使用 pthread_mutex_lock 函数int pthread_mutex_lock(pthread_mutex_t *mutex);该函数的作用是抢占这把锁抢到就进临界区抢不到就阻塞等待。也就意味着当函数调用到 pthread_mutex_lock(p) 时会发生两件事① 如果锁没被人占用线程成功拿到锁继续往下执行进入临界区② 如果锁已经被别人占用线程阻塞休眠直到持有锁的线程解锁醒来后重新抢锁抢到再继续。在我们代码中的具体体现是这样的int tickets 10000; pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER; void GetTicket() { char name[64]; pthread_getname_np(pthread_self(),name,sizeof(name)); while(1) { pthread_mutex_lock(mutex); if(tickets 0) { usleep(1000); printf(%s sells ticket:%d\n,name,tickets); tickets--; pthread_mutex_unlock(mutex); } else { pthread_mutex_unlock(mutex); break; } } }这里的 pthread_mutex_unlock 的逻辑和 pthread_mutex_lock 是一样的不过实现的是解锁而已。不过要注意的是不管是加锁还是解锁都是具有原子性的即不会被调度器打断一旦开始执行就必定会完整的执行完所以即使我们定义的锁是共享资源也不会像前面一样因为调度器导致资源不一的问题。下面是执行结果会发现我们最后的卖票数量停止在 ticket 1 处说明我们实现了线程的互斥。这其实就好比一个银行24小时 ATM 机每个人如果想要进行操作就必须排队等候 ATM 机空闲当有人进去了之后一定会对门进行反锁防止别的人进来引发安全问题。同样的当你的操作执行结束之后也需要将锁打开才能离开。所以不管是 if 还是 else 两个方向我们都要进行解锁操作。刚刚展示的是全局静态初始化锁我们还可以在主函数内初始化锁并使用实现栈内锁int ticket 10000; class thread_data { public: thread_data(const string n , pthread_mutex_t *p) :_name(n) ,_pmutex(p) {} string _name; pthread_mutex_t *_pmutex; }; void *route(void *args) { thread_data *td static_castthread_data *(args); while(1) { pthread_mutex_lock(td-_pmutex); if(ticket 0) { usleep(1000); printf(%s sells ticket:%d\n,td-_name.c_str(),ticket); ticket--; pthread_mutex_unlock(td-_pmutex); } else { pthread_mutex_unlock(td-_pmutex); break; } } return nullptr; } int main() { pthread_mutex_t mutex; pthread_mutex_init(mutex,nullptr); thread_data td1(thread-1,mutex); thread_data td2(thread-2,mutex); thread_data td3(thread-3,mutex); thread_data td4(thread-4,mutex); pthread_t t1,t2,t3,t4; pthread_create(t1,nullptr,route,(void*)td1); pthread_create(t2,nullptr,route,(void*)td2); pthread_create(t3,nullptr,route,(void*)td3); pthread_create(t4,nullptr,route,(void*)td4); pthread_join(t1,NULL); pthread_join(t2,NULL); pthread_join(t3,NULL); pthread_join(t4,NULL); pthread_mutex_destroy(mutex); return 0; }2.2 锁的实现原理探究为了实现互斥锁操作大多数 CPU 都提供了 swap 或 exchange 指令该指令的作用是把寄存器和内存单元的数据相交换由于只有一条指令保证了原子性即使是多处理器平台访问内存的 总线周期也有先后一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。 现在我们看一下 lock 和 unlock 的伪代码我们重点解释一下 lock 的逻辑线程先将寄存器%al置为 0再通过原子指令xchgb将寄存器值与锁变量mutex的值交换这一步同时完成了 “读取锁的当前状态” 和 “将锁标记为占用” 两个操作如果交换后%al的值大于 0说明之前锁是空闲的线程成功抢到锁并进入临界区否则说明锁已被占用线程会挂起等待之后再跳回开头重新尝试抢锁以此保证同一时间只有一个线程能进入临界区。而把内存中的变量交换到 CPU 内部的寄存器中本质上是把共享数据变成某个线程的私有数据因为如果在CPU的执行过程中该线程被切换走了该线程是会把CPU当时的上下文数据带走的因为调用了 xchgb 指令让寄存器和内存变量交换所以该 mutex 数据就被带走了。如果此时调度器切换了另一个线程CPU继续执行上述操作会发现寄存器内容还是 0 就会被挂起等待直到调度器重新调度原线程这叫做该线程竞争锁成功。当临界区代码执行完毕之后加锁的线程就会解锁实际上也只有进入临界区的线程才能解锁也是调用 movb 将 mutex 的值与 1 交换使这个锁处于空闲状态然后再唤醒等待锁的线程。在这里很有意思的点是其实互斥锁的本质就是看到底是 1 还是 0 用两个数字去表示一个资源的有和无这有点类似于信号量只不过互斥锁的信号量值为 1 而当代码处于临界区之所以要加锁就像是申请信号量一样本质都是对资源的预定机制。2.3 互斥锁的封装在C当中STL库当中也提供了关于互斥锁的容器可以使用各种接口它的本质也和我们之前封装过的vector、string等等是一个思路都是一个类所以我们可以尝试自主封装一下//Mutex.hpp #pragma once #include iostream #include pthread.h class Mutex { public: Mutex() { pthread_mutex_init(_lock,nullptr); } void Lock() { pthread_mutex_lock(_lock); } void Unlock() { pthread_mutex_unlock(_lock); } ~Mutex() { pthread_mutex_destroy(_lock); } private: pthread_mutex_t _lock; }; class LockGuard { public: LockGuard(Mutex lock) :_lockref(lock) { _lockref.Lock(); } ~LockGuard() { _lockref.Unlock(); } private: Mutex _lockref; };在这段代码当中我们先封装了 Mutex 类接着再将 Mutex 进行二次封装在主函数中的代码具体体现是这样的//Mian.cc Mutex lock; class thread_data { public: thread_data(const string n , pthread_mutex_t *p) :_name(n) ,_pmutex(p) {} string _name; pthread_mutex_t *_pmutex; }; void *route(void *args) { thread_data *td static_castthread_data *(args); while(1) { // pthread_mutex_lock(td-_pmutex); LockGuard lockguard(lock); if(ticket 0) { usleep(1000); printf(%s sells ticket:%d\n,td-_name.c_str(),ticket); ticket--; // pthread_mutex_unlock(td-_pmutex); } else { // pthread_mutex_unlock(td-_pmutex); break; } } return nullptr; }大家会发现我在函数中调用的直接是 LockGuard 类的对象因为我们将加锁与解锁封装成了 LockGuard 类的构造函数和析构函数那么在函数退出的时候会自动调用析构而不需要手动释放资源这种代码风格叫做 RAII 风格RAII Resource Acquisition Is Initialization资源获取即初始化说人话就是用对象的生命周期自动管理资源锁、内存、文件不用手动释放绝不泄漏。3. 线程同步3.1 基本概念在前面提到了线程互斥的概念我们解决了因为调度器导致的共享资源数据不一引发数据错乱的问题通过加锁限制一次只能有一个线程进入临界区但这也引发了另一个问题如果该线程持续频繁的访问临界区使得其他线程一直阻塞等待这不是会降低执行效率吗并且假如该线程已经执行完毕下一个进入临界区的线程是是谁呢如果不加以限制是不是还是会导致数据错乱呢因此我们就引发了线程同步的概念线程同步是指多个线程访问共享资源时按照预定规则有序执行避免数据错乱、冲突保证数据安全和逻辑正确即在临界资源安全的前提下让访问临界资源具有一定的顺序性。因为多线程天然是并发无序的CPU 随机切换线程若多个线程同时读写共享变量 / 文件 / 硬件就会产生竞态条件数据脏读、结果错误条件变量就可以解决这个问题。这样看上去条件变量和互斥锁看上去几乎一样都是用来解决一个线程要等另一个线程做完某件事才能继续运行的场景但实际上两者有很大区别对于互斥锁来说我占着厕所别人不能进我完事了才出来。别人必须等我 “用完” 才能进。对于条件变量来说我在厕所里但我在等外卖我不能一直占着厕所不让别人用啊 所以我先出来释放锁等外卖到了别人再叫我进去。3.2 条件变量条件变量是一个用于多线程编程的同步机制它允许一个或多个线程在某个条件不满足时进入等待状态并在其他线程改变共享数据后、使该条件满足时被唤醒并继续执行。概念性的东西其实和互斥锁的概念差不多。比如这是静态初始化pthread_cond_t cond PTHREAD_COND_INITIALIZER;其中 pthread_cond_t 也是一个和 pthread_mutex_t 一样的类型pthread_cond_t 是 POSIX 线程库中定义的条件变量数据类型用于实现线程间的条件同步。重点是需要记住条件变量必须与互斥锁配合使用。3.2 线程同步部分接口我们主要介绍三个在线程同步中使用的接口首先第一个是:pthread_cond_initint pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr);它的作用是初始化一个条件变量其中第一个参数是你定义的条件变量指针第二个参数代表的是条件变量的属性我们通常传 nullptr 代表按照条件变量的默认属性。第二个是 pthread_cond_wait int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);它的作用是让当前线程等待条件变量原子性地完成以下三个操作释放已持有的互斥锁阻塞当前线程等待条件变量被唤醒被唤醒后重新获取互斥锁其中第一个参数指向条件变量的指针第二个参数指向已加锁的互斥锁的指针。第三个是 pthread_cond_signalint pthread_cond_signal(pthread_cond_t *cond);它的作用是唤醒至少一个正在等待指定条件变量的线程。如果有多个线程在等待系统调度策略决定具体唤醒哪一个。其中的唯一一个参数代表指向条件变量的指针。3.3 生产者和消费者模型3.3.1 基本概念生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯而通过容器来进行通讯所以生产者生产完数据之后不用等待消费者处理直接扔给容器消费者不找生产者要数据而是直接从容器里取该容器就相当于一个缓冲区平衡了生产者和消费者的处理能力。这个容器就是用来给生产者和消费者解耦的。为了让大家更好的理解我们需要明确生产者、消费者之间的关系首先对于生产者和生产者来说大家都希望自己的产品能获得更多销量因此一定是处于竞争关系对于线程来说我们称之为互斥第二种是消费者和消费者的关系我们取一个极端的例子现在是世界末日全世界只剩下两个人和一桶泡面这两个人都已经快要饿死了那么此时对于这一桶泡面他俩肯定都想吃那么此时两人就处于竞争关系所以对于消费者和消费者来说也是互斥。我们在日常生活中感觉不明显只是因为产品过多资源过剩导致的第三个是生产者和消费者的关系消费者必须等待生产者制造产品之后才能去消费生产者必须等待消费者买走产品之后才能继续生成否则会滞销因此我们说生产者和消费者的关系是同步的但如果消费者不付钱直接拿产品或者生产者不制造商品直接抢消费者的钱那么此时二者就会处于竞争关系因此生产者和消费者之间的关系也是互斥的。经过这个解释之后如果以后再谈及生产者消费者模型我们可以这样总结一、 3 种关系生产者和生产者、消费者和消费者、生产者和消费者二、 2 种角色生产者和消费者线程三、 1 个交易场所通常由特定数据结构承担这三点总结起来就是 “ 3 2 1 ” 原则。3.3.2 模型中的条件变量那么这和我们前面提到的条件变量有什么关系呢我们用下面这张图去解释现在有两个人一个男孩一个女孩他俩都不能看到对方的操作他们在玩一个放苹果和拿苹果的游戏。为了保证拿和放的时候不被对方干扰于是就有了一个锁的机制当上锁时对方就不能干扰。男孩的任务是拿一个苹果然后申请锁再检查盘子里面有没有苹果如果没有的话就向盘子里放一个苹果如果有的话就自动走到铃铛处的位置去排队等候。女孩的任务是申请锁检查盘子里面有没有苹果如果有的话就把苹果拿出来然后解锁退出如果没有的话就去敲铃铛告诉男孩该放苹果了此时男孩接收到信息就去做对应的工作。在这个场景里男孩对应的是生产者女孩对应的是消费者盘子对应的是提供交互的数据结构锁对应的就是互斥锁铃铛和维护的队列queue这个整体就是条件变量。因此我们的 pthread_cond_wait 相当于是让生产者去排队等候而 pthread_cond_signal 就相当于是敲铃铛告诉生产者的动作。另外还有一个 pthread_cond_broadcast 函数因为我们的生产者并不一定只有一个人因此在铃铛处等待的可能有多个线程该函数就是一次性唤醒当前在该条件变量处等候的所有线程让他们同时去竞争锁。同样的条件变量可以去限制生产者就也可以去限制消费者当消费者发现盘子里面没有苹果时就会到铃铛处等候当生产者放入苹果后去敲铃铛消费者再去拿苹果这就更加提高了秩序性。3.3.3 基于阻塞队列的模型在多线程编程中阻塞队列Blocking Queue是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于当队列为空时从队列获取元素的操作将会被阻塞直到队列中被放入了元素当队列满时往队列里存放元素的操作也会被阻塞直到有元素被从队列中取出以上的操作都是基于不同的线程来说的线程在对阻塞队列进行操作时会被阻塞。大家看到这个描述一定会不自觉的想起来我们之前学习的管道的知识管道不也是当空间写满的时候就不能写了当读数据读到空了就不能读了。其实我们学的管道本质上就是基于字节流的一个阻塞队列。3.3.4 基于queue模拟阻塞队列的模型我先向大家直接展示代码然后一一做解释//Main.cc #include BlockQueue.hpp #include unistd.h void *ConsumerRoutine(void *args) { BlockQueueint *bq static_castBlockQueueint *(args); while(true) { int data; bq-Pop(data); cout 消费者数据 data endl; // sleep(1); } return nullptr; } void *ProductorRoutine(void *args) { BlockQueueint *bq static_castBlockQueueint *(args); int data 1; while(true) { sleep(3); bq-Enqueue(data); cout 生产者数据 data endl; // sleep(1); } } int main() { BlockQueueint *bq new BlockQueueint(); pthread_t c, p; pthread_create(c,nullptr,ConsumerRoutine,bq); pthread_create(c,nullptr,ProductorRoutine,bq); pthread_join(c,nullptr); pthread_join(p,nullptr); return 0; }//BlockQueue.hpp #ifndef __BLOCK_QUEUE_H #define __BLOCK_QUEUE_H #include iostream #include queue #include pthread.h using namespace std; const int defaultcap 5; templatetypename T class BlockQueue { public: BlockQueue(int cap defaultcap) :_cap(cap) { pthread_mutex_init(_mutex,nullptr); pthread_cond_init(_consumer_cond,nullptr); pthread_cond_init(_productor_cond,nullptr); //定义水位线 // _blockqueue_low_water _cap*1/3; // _blockqueue_high_water _cap*2/3; sleep_productor_num 0; sleep_consumer_num 0; } void Enqueue(T in) //插入 -- 生产者 { pthread_mutex_lock(_mutex); // ? while(_bq.size() _cap) { //生产者休眠数量 sleep_productor_num; pthread_cond_wait(_productor_cond,_mutex); //唤醒后在此继续执行后续代码生产者休眠数量-- sleep_productor_num--; } _bq.push(in); //水位线式唤醒 // if(_bq.size() _blockqueue_high_water) // pthread_cond_signal(_consumer_cond); //解释为什么最好在加锁和解锁之间唤醒 //休眠数量式唤醒 if(sleep_consumer_num 0) pthread_cond_signal(_consumer_cond); pthread_mutex_unlock(_mutex); } void Pop(T *out) //提取 -- 消费者 { pthread_mutex_lock(_mutex); while(_bq.empty()) //队列为空需要等待 { //等待时是在临界区内部等待如果在这里不释放锁 //别的线程就拿不到锁就会导致整个进程阻塞 //当线程在原地被唤醒时pthrad_con_wait 会让线程自动竞争锁 //在临界区中等待是因为访问临界资源必然在临界区内部访问 //判断资源是否就绪本质也是访问临界资源 //我们在访问临界资源即判断资源是否就绪之后才能决定是否要等待 //因此等待必定在临界区内部等待 //那到底为什么要判断队列是否为空如果不判断呢 //伪唤醒 sleep_consumer_num; pthread_cond_wait(_consumer_cond,_mutex); sleep_consumer_num--; } *out _bq.front(); _bq.pop(); //如果对方已经是唤醒状态此函数就会被忽略 //采用水位线策略 // if(_bq.size() _blockqueue_low_water) // pthread_cond_signal(_productor_cond); //休眠数量式唤醒 if(sleep_consumer_num 0) pthread_cond_signal(_productor_cond); pthread_mutex_unlock(_mutex); } ~BlockQueue() { pthread_mutex_destroy(_mutex); pthread_cond_destroy(_consumer_cond); pthread_cond_destroy(_productor_cond); } private: queueT _bq; int _cap; //容量上限 pthread_mutex_t _mutex; pthread_cond_t _consumer_cond; pthread_cond_t _productor_cond; //除了生产/消费后立即唤醒还可以采用低水位线的方式预警 // int _blockqueue_low_water; //消费低水位线 // int _blockqueue_high_water; //生产低水位线 int sleep_productor_num; int sleep_consumer_num; }; #endif我们以解释 BlockQueue.hpp 代码为例在开头处加上了 #ifndef 结尾跟上一个 #endif 这其实是头文件保护相当于 #pragma once 这里的意思是#ifndef __BLOCK_QUEUE_H // 如果没有定义过 __BLOCK_QUEUE_H #define __BLOCK_QUEUE_H // 就定义它并包含下面的代码 // ... 头文件的全部内容 ... #endif // 结束条件编译块我们的主要思想还是参考两个小孩玩拿苹果和放苹果的逻辑所以定义了两个条件变量 _consumer_cond 和 _productor_cond 另外还有一把互斥锁。中间承担交易场所的数据结构我们使用 queue 队列为了限制传递信息的数量我们定义了队列容量大小 defaultcap 初始化为 5 如果我们不定义该容量大小可能会引发 一个问题如果生产者生产速度太快而消费者速度很慢该队列会“无限”扩容容量达到无限大除了引发资源浪费还有可能导致系统崩溃。对于生产者它是向容器中放入数据的我们创建了一个Enqueue的函数该函数传递进来的参数对应主函数中的这个我们这里要注意的是在进入函数之后我们首先要对队列是否为满检查这符合我们说的“如果盘子里有苹果那小男孩就不去放转而去排队等候小女孩取完苹果再敲铃铛”因此在函数中我们调用 pthread_cond_wait 函数如果队列满了那就阻塞等待在这里我们用的是循环去判断循环的主要目的是应对“虚假唤醒”以及“唤醒后条件可能仍然不满足”这两种情况首先解释一下什么叫虚假唤醒在我们的常识中如果需要唤醒一个线程那就一定会调用 pthread_cond_signal 或者 pthread_cond_broadcast 函数此时线程唤醒后从 pthread_cond_wait 函数处继续向后执行那如果没有调用这两个函数线程就直接被唤醒呢这就叫做虚假唤醒。这种情况的出现是因为操作系统线程调度机制内核为简化实现、提升调度效率在等待队列重组、系统中断、外部信号触发等场景下可能无意识唤醒等待线程高并发环境下该现象更易出现。当出现虚假唤醒的情况并且 pthread_cond_wait 函数还是放在 if 语句中的此时明明消费者没有取出数据队列还是满的但因为被虚假唤醒操作系统就不会再检查队列是否为满直接向下执行就会引发栈溢出错误。此外大家还会看到注释中写了生产者休眠数量其实这是一种唤醒策略唤醒策略是指当缓冲区状态发生变化时依据预设规则判断何时、以何种方式唤醒阻塞的生产者 / 消费者线程用来精准控制线程启停平衡并发性能、减少无效唤醒与竞争。比如对于生产者来说看到队列满了就需要休眠阻塞因此需要消费者去唤醒它但该怎么唤醒呢我是取走一个数据之后马上就把你唤醒还是我稍微取几个数据之后再唤醒因为阻塞和唤醒都是需要时间的不同的策略会有不同的效率所以我们一共有三种方式1. 直接唤醒生产/消费后立即唤醒2. 采用水位线法对于队列中的数据量做一个基准如果队列中达到某数量就去唤醒对应的线程。3. 休眠数量检查法这其实是基于多生产者/消费者线程的即比如有多个生产者线程正在休眠数量达到一定值时就由消费者唤醒一个生产者线程。在唤醒之前我们需要把主函数中由生产者想要放入的数据插入到队列当中那么对于消费者只需要把队列中的数据调用 front 函数就可以取出然后再 pop 删除掉队列中的该数据表示已经被取出即可。我们在这还需要解释一个问题为什么我们选择将唤醒操作放在加锁和解锁之间明明解锁了之后我再将对方唤醒也可以啊。其实pthread_cond_signal并不是必须在加锁和解锁之间调用技术上讲你可以在解锁之后才调用signal。但是强烈推荐在加锁状态下调用这是为了避免一个经典的竞态条件称为“丢失唤醒”。假设把signal放到解锁之后会出现经典时序问题消费者线程加锁 → 判断队列为空 → 执行pthread_cond_wait因为 wait内部先释放锁线程进入休眠等待生产者线程抢到锁 → 放入数据 →解锁此时锁已释放生产者刚解锁还没执行signalCPU 再次切回消费者消费者被内核调度唤醒重新拿到锁再次判断条件队列非空直接向下执行最后生产者才执行signal但此时已经没有线程在等待唤醒信号直接作废下一轮队列再次变空消费者又执行wait进入休眠 生产者后续持续生产数据、解锁、发信号信号依旧不断丢失 消费者永远等不到有效唤醒一直卡在wait 若队列被生产至满生产者也会执行wait阻塞。最后的结果就是生产者、消费者双双阻塞所有线程停滞形成死锁然后展示一下生产/消费者线程对于唤醒的逻辑等待者 唤醒者 [持锁] 检查 condition false 准备进入 wait [等待锁] ← 被阻塞无法操作 进入 wait原子释放锁阻塞 [获得锁] ← 等待者已释放锁 condition true signal唤醒等待者 解锁 [被唤醒重新持锁] ← 在 wait 返回时 检查 condition true ← 条件满足退出循环 解锁至此Enqueue函数解释完Pop是对于消费者的代码总体逻辑大差不差再次不过多赘述。4. POSIX 信号量POSIX 信号量是由 POSIX 标准定义、基于 Linux 内核实现的计数器同步原语可用于多线程与多进程场景下的同步互斥既能实现共享资源的互斥访问以替代互斥锁也能完成生产者 - 消费者模型同步并管控可用资源数量。我们之前说过信号量的本质为一个非负整数计数器仅包含两类核心操作P 操作即 sem_wait 等待操作会将计数器数值减一若计数器值为 0 则线程或进程阻塞等待V 操作即 sem_post 释放操作会将计数器数值加一并唤醒处于阻塞等待状态的线程或进程。这在我之前的文章中的信号量的基本概念中提到过LinuxSystem V 消息队列与信号量_linux system v 信号量-CSDN博客而我们现在的目的就是再认识一个基于环形队列的使用POSIX信号量完成的生产者消费者模型。首先我们需要明确环形队列的核心矛盾与约束。队列为空时生产者的尾指针tail和消费者的头指针head指向同一个位置队列为满时这两个指针同样会重合。为了区分这两种状态我们并不依赖指针本身的数值而是依赖两个POSIX信号量所代表的“资源计数”。同时模型必须严格遵守两条铁律生产者绝不能“跑满一圈”去覆盖消费者还没来得及取走的数据即生产者不能超过消费者一圈消费者绝不能“超车”去取生产者尚未生产出来的数据即消费者不能超过生产者。在这个模型中我们将问题抽象为两种不同的“资源”。从生产者的视角看队列中的“空位”才是资源有多少个空位就能生产多少个数据从消费者的视角看队列中填充的“数据”才是资源。因此我们初始化两个信号量blank_sem其初始值为缓冲区大小 N代表当前有 N 个空位可供使用data_sem其初始值为 0代表当前有 0 个数据可供消费。当生产者需要放入数据时它首先执行P(blank_sem)即sem_wait操作。这个操作会尝试申请一个空位资源如果此时队列已满blank_sem 为 0生产者就会被阻塞在这里从而完美地阻止了生产者覆盖尚未被消费的数据即实现了“生产者不能超过消费者一圈”的约束。一旦申请成功生产者便获得了写入权限将数据存入环形队列的ring[tail]位置随后将尾指针向后移动tail并通过对 N 取模tail % N实现环形回绕。数据放置完毕后生产者必须执行V(data_sem)即sem_post释放数据资源通知消费者队列中已有新数据可供提取。与此对称消费者要取数据时必须先执行P(data_sem)。这个操作会申请一个数据资源如果当前队列为空data_sem 为 0消费者就会被阻塞在这里从而严格遵守“消费者不能超过生产者”的约束绝不会去空队列里取东西。成功申请到数据资源后消费者从ring[head]位置取出数据并将头指针后移head同样通过head % N维持环形结构。消费完成后消费者必须执行V(blank_sem)释放一个空位资源告知生产者队列中空出了一个位置可以继续放入新数据。#pragma once #include iostream #include string #include vector #include Sem.hpp using namespace std; const int defaultcap 5; templatetypename T class RingQueue { public: RingQueue(int cap defaultcap) :_cap(cap) ,_rq(cap) ,_consumer_step(0) ,_productor_step(0) ,_blank_sem(cap) ,_data_sem(0) {} void Enqueue(T in) //入队列生产者用 { //先申请信号量 _blank_sem.P(); //再找位置生产 _rq[_productor_step] in; _productor_step % _cap;//控制下标循环 //最后释放信号量 _data_sem.V(); } void Pop(T *out) //出队列消费者用 { _data_sem.P(); *out _rq[_consumer_step]; _consumer_step % _cap; _blank_sem.V(); } ~RingQueue() {} private: int _cap;//队列容量 vectorT _rq;//环形队列 int _consumer_step;//消费位置 int _productor_step;//生产位置 Sem _blank_sem; //格子资源计数器生产者关心 Sem _data_sem; //数据信号量消费者关心 };总结来说这套机制巧妙地利用两个信号量的计数来隐式地管理环形队列的空满状态。blank_sem充当着生产者的“刹车片”防止数据覆盖data_sem充当着消费者的“刹车片”防止空读。指针的移动与取模运算保证了物理上的环形复用而信号量的PV操作则通过资源计数的增减在逻辑上精确地控制了生产与消费的步调使得两者既不会相互超越也能在并发环境下高效协作在这个单对单的特定场景下也无需额外加锁。这其中的 Sem 类型也是我们对信号量库中的函数进行的类封装#ifndef __SEM_HPP #define __SEM_HPP #include iostream #include semaphore.h using namespace std; class Sem { public: Sem(int init_val) { if(init_val 0) { sem_init(_sem,0,init_val); } } void P() { int n sem_wait(_sem); } void V() { int n sem_post(_sem); } ~Sem() { int n sem_destroy(_sem); (void)n; } private: sem_t _sem; }; #endif我们可以得到这个总结相比于使用互斥锁POSIX信号量的优点是自带计数器既能互斥又能天然实现同步等待不用条件变量。5. 日志与策略模式首先我们需要明确日志在计算机系统中所扮演的角色。日志是系统运行过程中产生的、按时间序列排列的事件记录集合其核心价值在于诊断故障、监控性能以及追溯行为。一个标准的日志条目通常包含两个基本维度日志等级与日志内容。日志等级如 DEBUG、INFO、WARN、ERROR、FATAL用于标记事件的严重性或用途它决定了日志在运行时的过滤阈值日志内容则包含了时间戳、代码位置文件名/行号、核心消息体以及关键的上下文变量。然而日志系统往往面临一个棘手的挑战输出行为的多变性。在开发调试阶段我们希望日志以鲜艳的颜色直接打印在控制台上方便即时查看在正式上线后我们需要将其持久化写入磁盘文件并按照日期切割归档在复杂的微服务环境中我们甚至需要将日志格式化为 JSON 并投递至远程日志中心。如果把这些输出逻辑通过大量的if-else硬编码写在日志类内部每次新增一种输出方式都需要修改核心类这既违反了开闭原则也会导致代码急剧膨胀。为了化解这一矛盾我们可以引入策略模式Strategy Pattern。策略模式的核心思想是定义一系列算法即“策略”将它们封装成独立的类并使它们可以互相替换。在日志设计的语境下这个“算法”就是“日志的写入行为”。我们将“日志内容的拼装”与“日志输出的目的地”彻底分离。日志类本身Context上下文只负责接收原始数据、拼装通用的日志格式而对于最终的写入动作它只持有一个抽象的策略接口引用具体的写入逻辑委托给该接口的实现类来完成。下面通过代码块来清晰演示这种结构。首先是定义策略的抽象接口#include iostream #include fstream #include string #include chrono #include iomanip #include ctime #include memory using namespace std; // 抽象策略定义日志写入的契约 class LogStrategy { public: virtual ~LogStrategy() default; virtual void write(const string level, const string message) 0; };基于这个接口我们可以自由衍生出多种具体策略。例如控制台策略输出到屏幕文件策略则追加到本地磁盘// 辅助函数获取当前时间字符串 std::string getCurrentTime() { auto now std::chrono::system_clock::now(); auto time_t std::chrono::system_clock::to_time_t(now); std::stringstream ss; ss std::put_time(std::localtime(time_t), %Y-%m-%d %H:%M:%S); return ss.str(); } // 具体策略A控制台输出 class ConsoleStrategy : public LogStrategy { public: void write(const std::string level, const std::string message) override { std::cout [ getCurrentTime() ] [ level ] message std::endl; } }; // 具体策略B文件输出 class FileStrategy : public LogStrategy { private: std::string filepath_; public: explicit FileStrategy(const std::string path) :filepath_(path) {} void write(const std::string level, const std::string message) override { std::ofstream ofs(filepath_, std::ios::app); if (ofs.is_open()) { ofs [ getCurrentTime() ] [ level ] message std::endl; } } };最后我们构建日志上下文类即开发者直接使用的入口。它维护日志等级过滤并持有当前生效的策略对象通过智能指针管理class Logger { private: std::unique_ptrLogStrategy strategy_; std::string min_level_; // 为简化演示省略等级数值比较细节 public: // 构造函数注入策略 Logger(std::unique_ptrLogStrategy strategy, std::string level INFO) :strategy_(std::move(strategy)), min_level_(std::move(level)) {} // 运行时动态切换输出策略无需修改Logger内部代码 void setStrategy(std::unique_ptrLogStrategy strategy) { strategy_ std::move(strategy); } void log(const std::string level, const std::string message) { // 此处可增加等级过滤逻辑例如 if(level min_level_) strategy_-write(level, message); // 核心委托给策略 } };在实际调用中这种设计的灵活性便凸显出来int main() { // 初始化时使用文件策略 auto logger Logger(std::make_uniqueFileStrategy(app.log), WARN); logger.log(ERROR, 数据库连接超时); // 写入文件 // 运行时动态切换至控制台策略例如进入调试模式 logger.setStrategy(std::make_uniqueConsoleStrategy()); logger.log(DEBUG, 当前缓存命中率为99%); // 直接打印到屏幕 return 0; }总结来说策略模式在日志系统中的应用本质上是将“易变的输出行为”与“稳定的日志组装框架”进行了解耦。日志等级和内容格式属于数据的静态属性由Logger核心统一负责拼接而“写到哪里”和“怎样写”属于行为的动态算法由各个具体的策略类分别封装。当未来需要新增“上传至 ElasticSearch”或“加密存储”时只需新增一个继承自LogStrategy的子类即可完全无需改动现有的Logger核心代码这使得日志系统具备了极强的可扩展性与维护性。本文到此结束感谢各位读者的阅读如果有讲解的不到位或者错误的地方欢迎各位读者批评或指正。
