目录互斥为什么要有互斥实现互斥的方式-互斥锁C线程库为例互斥锁的原理C语言线程库提供的有关互斥锁的接口简单使用一下互斥锁重新理解互斥锁同步为什么要有同步条件变量实现同步C线程库为例条件变量的核心数据结构C语言线程库提供的有关条件变量的接口为什么条件变量要依赖互斥锁条件变量实现同步伪唤醒信号量实现同步C线程库为例信号量的核心数据结构C语言线程库提供的信号量的相关接口信号量实现同步生产者消费者模型基于阻塞队列的生产者-消费者模型基于环形阻塞队列的生产者消费者模型生产者消费者模型究竟为何高效互斥为什么要有互斥一些共享资源在多执行流并发访问中会出现错误为了防止错误的发生我们需要控制访问这些共享资源的执行流数量1即多个执行流互斥地访问共享资源。关于会产生什么错误这里就不再赘述相关文章有很多。实现互斥的方式-互斥锁C线程库为例当然互斥的实现方式多种多样比如中断屏蔽比如OS提供的原子操作接口而我给读者介绍的只是其中一种——互斥锁互斥锁的原理定义一个变量var表示可以访问该共享资源的执行流的数量将它的初始值设置为1。让所有想要访问共享资源的执行流都去检查var的值并规定如果var为1说明可以访问该资源本执行流将var置0后进行访问访问结束后再把var置1如果var为0说明不可以访问该资源本执行流继续检查。这样一来只有先检查到var 1 的执行流可以访问共享资源。类比先检查到var 1 的执行流会将var置为0此时共享资源如同被加锁后来的执行流无法访问它。访问结束后var又被置为1此时共享资源如同被解锁其余执行流可以争抢并访问这个资源。因此这种实现互斥的方式叫做互斥锁。C语言线程库提供的有关互斥锁的接口创建锁#includepthread.h pthread_mutex_t mutex;pthread_mutex_t是linux提供的结构体类型这里面包含有实现互斥所以用的var变量。静态初始化#includepthread.h pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER;静态初始化专用于静态存储期全局变量静态局部变量等的锁变量而局部变量或者动态创建的锁不可用这种方式初始化。动态初始化:#includepthread.h int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *mutexattr); //mutex:锁的地址 //mutexattr:可以给锁设置属性一般为nullptr //函数调用成功返回0动态初始化可以用于初始化静态存储期和非静态存储期包括堆空间栈空间等的锁变量但是使用这种方式初始化锁必须使用pthread_mutex_destroy函数销毁锁。加锁#includepthread.h int pthread_mutex_lock(pthread_mutex_t *mutex);对应于检查var变量解锁#includepthread.h int pthread_mutex_unlock(pthread_mutex_t *mutex);对应于var重置为0锁的销毁#includepthread.h int pthread_mutex_destroy(pthread_mutex_t *mutex);如果mutex是堆空间上的变量不要忘记调用delete或者free释放锁变量。简单使用一下互斥锁// 伪代码使用互斥锁保护共享数据 // 定义共享资源 int account_balance 1000; // 银行账户余额 //定义并初始化互斥锁 pthread_mutex_t balance_lock PTHREAD_MUTEX_INITIALIZER; // 存款操作线程安全 void deposit(int amount) { pthread_mutex_lock(balance_lock); // 加锁进入临界区 // 临界区开始操作共享数据 int old_balance account_balance; account_balance old_balance amount; printf(存款%d - %d\n, old_balance, account_balance); // 临界区结束 pthread_mutex_unlock(balance_lock); // 解锁离开临界区 } // 取款操作线程安全 bool withdraw(int amount) { pthread_mutex_lock(balance_lock); // 加锁 // 临界区 bool success false; if (account_balance amount) { int old_balance account_balance; account_balance old_balance - amount; printf(取款%d - %d\n, old_balance, account_balance); success true; } else { printf(取款失败余额不足\n); } pthread_mutex_unlock(balance_lock); // 解锁 return success; } // 主函数创建多个线程并发操作 int main() { // 创建多个线程同时进行存款和取款 create_thread(deposit, 100); // 线程1存100 create_thread(withdraw, 200); // 线程2取200 create_thread(deposit, 50); // 线程3存50 // 等待所有线程完成 wait_all_threads(); printf(最终余额%d\n, account_balance); // 应该是950 return 0; }重新理解互斥锁在互斥锁的原理中有两处疑点var本身也是会被多个执行流访问的共享资源所以多个执行流不能同时对var的操作。如果在某执行流访问共享资源的同时其余执行流都不断轮询检查var的值那么有很多CPU资源都被浪费了。那么实际上互斥锁的原理如下mutex相当于之前提到的varlock是加锁函数的伪代码unlock是解锁函数的伪代码xchgb是一条交换指令所有执行流都执行这条指令——将自己cpu上下文中的al0与mutex的值交换。如果发现交换后al是1说明该执行流检查到mutex为1并且将mutext置为了0实现了加锁如果发现交换后al是0该执行流检查到mutex为0。可以发现检查mutex的值和修改mutex的值仅仅通过一条指令就做到了而指令的执行是原子的不可被中断因此虽然mutex是共享资源但由于对mutex的操作都是原子的所以访问mutex不用加以保护。movb是一条赋值指令访问过共享资源的执行流执行这条指令将mutex重新置为1这就是解锁操作。同理原子操作保证访问mutex不用加以保护。加锁失败后本执行流会挂起等待。被挂起的执行流的PCB/TCB会被放到内核中的某等待队列中并且PCB/TCB的状态会被修改为睡眠所以本执行流不再被调度。当某执行流释放了锁后才会唤醒所有挂起等待的执行流所有执行流重新开始竞争。正是因此不会浪费太多CPU资源。同步为什么要有同步其一资源竞争与合理调度。在使用锁进行并发控制时持有锁的线程释放后可能立即再次参与竞争从而连续获得锁资源。这会导致其他等待锁的线程长期处于饥饿状态无法获得执行机会例如在生产-消费者模型中若消费者持续争抢锁将阻塞生产者放入数据的动作造成流程僵持与系统效率下降。因此同步机制也需考虑竞争的合理性避免单一执行流无作为地占用资源。这是共享资源与执行流之间的同步关系。其二任务依赖与顺序控制。多个执行流在协作处理任务时往往存在严格的先后依赖关系例如炒菜流程必须先完成洗菜才能进行烧油之后方可开始炒菜。若无同步机制协调执行顺序可能错乱导致任务无法正确完成。这是执行流与执行流之间的同步关系。条件变量实现同步C线程库为例条件变量的核心数据结构条件变量内维护一个等待队列执行流可以调用接口主动在该队列上睡眠等待也可以调用接口唤醒队列上等待的一个或者所有执行流。C语言线程库提供的有关条件变量的接口对条件变量进一步的叙述需要结合接口的使用来进行所以在这里先来介绍一下条件变量接口的使用希望读者不要感到突兀。创建条件变量#includepthread.h pthread_cond_t cond;静态初始化#includepthread.h pthread_cond_t cond PTHREAD_COND_INITIALIZER;静态初始化专用于静态存储期全局变量静态局部变量等的条件变量而局部变量或者动态创建的条件变量不可用这种方式初始化。动态初始化#includepthread.h int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr); //cond_attr是设置属性的参数一般设置为nullptr //初始化成功返回0动态初始化可以用于初始化静态存储期和非静态存储期包括堆空间栈空间等的条件变量但是使用这种方式初始化的条件变量必须使用pthread_cond_destroy函数销毁条件变量。销毁条件变量#includepthread.h int pthread_cond_destroy(pthread_cond_t *cond);在条件变量下睡眠等待#includepthread.h int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex); //mutex是互斥锁调用pthread_cond_wait函数时需要绑定一个互斥锁且这个互斥锁必须是加锁状态否则导致未定义行为。调用pthread_cond_wait函数后绑定的互斥锁会自动释放然后执行流阻塞等待等到执行流被唤醒的时候会自动申请加锁如果申请成功会从pthread_cond_wait语句后面继续执行如果申请失败会在锁的等待队列上等待继续申请加锁。唤醒条件变量下等待执行流#includepthread.h int pthread_cond_signal(pthread_cond_t *cond);//唤醒一个 int pthread_cond_broadcast(pthread_cond_t *cond);//唤醒所有如果使用pthread_cond_signal唤醒单个等待的执行流唤醒的是最先入队列的执行流先进先出为什么条件变量要依赖互斥锁条件变量是给不满足条件的线程一个休眠的栖息地对于单个执行流来说不会无缘无故的就被唤醒或者休眠一定是另一个线程修改了共享资源使得条件满足而共享资源的访问是一定要加锁的所以条件变量的使用模板如下lock() if(条件不成立) { //释放锁然后休眠等待被唤醒 unlock() conditio_variable.wait() }这里有两个问题如果在unlock()和wait()中间有另一个线程修改了共享资源并唤醒休眠的线程他会发现没有可以被唤醒的线程因为刚才的线程还没来得及休眠。这是一个严重的安全问题和效率问题最坏情况下线程永远不会被唤醒或者很就才被唤醒影响效率每次调用unlock很烦而条件变量的接口把释放锁和休眠动作变成了原子的解决了这个问题也省去了繁琐的操作​​​​​​​条件变量实现同步条件变量这种手段实现同步的方式就是通过判断被互斥锁保护的共享资源是否满足条件决定执行流是否在条件变量上等待。因此天然的条件变量很适合实现资源与执行流之间的同步也就是保证合理调度。这在生产者-消费者模型中有体现在缓冲区空的时候消费者不要再竞争锁而是让生产者放在缓冲区满的时候生产者不要再竞争锁而是让消费者拿。伪唤醒if(条件)//满足条件即说明共享资源没准备好该执行流需要等待 { pthread_cond_wait(cond,mutex); }有这样一种情况共享资源准备好了因此唤醒等待在条件变量下的执行流执行流1被唤醒后没有成功加锁而执行流2成功加锁等到执行流2访问共享资源结束共享资源又变成没有准备好的状态了此时执行流1才成功加锁因为执行流1是被唤醒的执行流1会从pthread_cond_wait语句后面开始执行即执行流1此时本来应该等待但却越过条件判断访问了共享资源。解决方法就是让被唤醒并成功加锁的执行流再进行条件判断如下while(条件)//满足条件即说明共享资源没准备好该执行流需要等待 { pthread_cond_wait(cond,mutex); }信号量实现同步C线程库为例信号量的核心数据结构信号量是资源计数器它表示可用资源的数目。因此它的核心数据结构就是计数器可能就是一个整型变量而且信号量本身也是需要被多执行流共同访问的共享资源因此也需要互斥保护所以信号量内部应该还有互斥锁和等待队列防止循环等待一类的数据结构当然如果可以依靠别的方式实现计数器的互斥访问就不需要使用互斥锁了比如用中断屏蔽或者系统提供的原子操作接口。总而言之信号量的核心数据结构包括计数器保证计数器互斥访问的机制等待队列。C语言线程库提供的信号量的相关接口创建信号量#include semaphore.h sem_t sem;初始化信号量#include semaphore.h int sem_init(sem_t *sem, int pshared, unsigned int value); //sem:信号量 //pthread:一般不关心设置为0 //value:信号量代表的资源数目 //成功返回0销毁信号量#include semaphore.h int sem_destroy(sem_t *sem);申请资源#include semaphore.h int sem_wait(sem_t *sem); //成功返回0申请成功信号量的资源计数器减一申请资源失败就会阻塞释放资源#includesemaphore.h int sem_post(sem_t *sem); //成功返回0释放成功信号量的资源计数器加一信号量实现同步虽然我上述提到信号量表示某种可用资源的数目听起来信号量就是为了实现资源与执行流之间的同步关系合理调度。但事实恰恰相反信号量最适合的是实现执行流与执行流之间的同步关系顺序控制。比如可以这样实现多个执行流的顺序控制//由三个同步关系设置三个信号量资源计数器为1 sem_t A sem_t B sem_t C P表示申请资源V表示释放资源 //执行流1 ---相关操作 V(A) //执行流2 ---相关操作 V(B) //执行流3 p(A) p(B) ---相关操作 V(C) //执行流4 p(C) ---相关操作此时V操作如同发送信号P操作如同等待信号只有某个事件收到了所有前驱事件发送的信号该事件才会执行。除此之外一旦资源耗尽申请资源的操作就会阻塞因此单纯使用信号量也可以实现共享资源与执行流之间的同步合理调度但是这是有局限性的不如条件变量那样适用范围广——条件变量的阻塞条件可以多种多样而信号量的阻塞条件只有资源耗尽。生产者消费者模型生产者消费者模型两个或多个执行流之间通过共享缓冲区进行数据传输。基于阻塞队列的生产者-消费者模型#includepthread.h #includequeue #includemutex #includeiostream using namespace std; /*这是一个阻塞队列它的特点是 1.队列为空取出操作阻塞队列为满插入操作阻塞。 2.线程安全 */ templateclass T class BQ { private: queueT _que; pthread_mutex_t _mutex;//保证对阻塞队列的操作线程安全的锁 pthread_cond_t _cond1;//生产者的等待队列 pthread_cond_t _cond2;//消费者的等待队列 int _cap;//阻塞队列的容量 public: bool isfull(){return _que.size() _cap;} bool isempty(){return _que.size() 0;} BQ(int capacity): _cap(capacity) { pthread_mutex_init(_mutex,nullptr); pthread_cond_init(_cond1,nullptr); pthread_cond_init(_cond2,nullptr); } ~BQ() { pthread_mutex_destroy(_mutex); pthread_cond_destroy(_cond1); pthread_cond_destroy(_cond2); } void push(const T arg) { pthread_mutex_lock(_mutex);//加锁 while(isfull())//用while而不是if,防止伪唤醒 { cout生产者等待endl; pthread_cond_wait(_cond1,_mutex);//缓冲区满了就让执行流睡眠等待 } _que.push(arg); pthread_cond_signal(_cond2);//唤醒消费者等待队列中的消费者 pthread_mutex_unlock(_mutex);//解锁 } void pop(T value) { pthread_mutex_lock(_mutex);//加锁 while(isempty())//用while而不是if,防止伪唤醒 { cout消费者等待endl; pthread_cond_wait(_cond2,_mutex);//缓冲区满了就让执行流睡眠等待 } value _que.front(); _que.pop(); pthread_cond_signal(_cond1);//唤醒消费者等待队列中的消费者 pthread_mutex_unlock(_mutex);//解锁 } }; ----------------------------------------------------------------------------------------- #includefunctional #includeblockqueue.hpp #includeunistd.h using task_t functionvoid(); void* product(void* arg)//生产者的函数 { BQtask_t * buffer static_castBQtask_t*(arg); int x 5; while(1) { sleep(1); cout生产任务endl; buffer-push([x]()-void{cout任务x执行endl;}); x; } } void* consume(void* arg)//消费者的函数 { BQtask_t* buffer static_castBQtask_t*(arg); while(1) { task_t task; buffer-pop(task); task(); } } int main() { BQtask_t buffer(5);//使用阻塞队列做缓冲区,队列中的产品是任务函数,在这种情况下生产者不断生产任务而消费者不断执行拿到的任务函数 pthread_t id1; pthread_t id2; pthread_t id3; pthread_t id4; //两个生产者 pthread_create(id1,nullptr,product,buffer); pthread_create(id2,nullptr,product,buffer); //两个消费者 pthread_create(id3,nullptr,consume,buffer); pthread_create(id4,nullptr,consume,buffer); pthread_join(id1,nullptr); pthread_join(id2,nullptr); pthread_join(id3,nullptr); pthread_join(id4,nullptr); }使用了两个条件变量利用共享资源与执行流之间的同步关系保证了合理调度即缓冲区为空的时候消费者不会争抢锁导致无效加锁以及生产者无法放入缓冲区为满时生产者不会争抢锁导致无效加锁以及消费者无法拿取。注该类型的生产者消费者模型将缓冲区当作整体使用因此同一时间访问缓冲区执行流只能有一个。此外它使用条件变量实现合理调度。基于环形阻塞队列的生产者消费者模型#includesemaphore.h #includevector #includemutex #includeiostream using namespace std; /*这是一个环形阻塞队列它的特点是 1.队列为空取出操作阻塞队列为满插入操作阻塞。 2.线程安全 */ /* 信号量如何实现同步 当队列为空时数据资源数量为0因此消费者无法消费 当队列为满时空间资源数量为0因此生产者无法生产。 由此实现合理调度 同时由于信号量控制了为空未满的时候读写不能同时操作互斥因此读写之间可以自然而然并发并行 */ templateclass T class RQ { private: vectorT _que;//数组模拟循环队列 pthread_mutex_t _mutexp;//保证多个生产者互斥访问的锁 pthread_mutex_t _mutexc;//保证多个消费者互斥访问的锁 sem_t _data;//表示可读取的资源数量的信号量 sem_t _empty;//表示可写入的资源空位置数量的信号量 int _cap;//队列的容量 int _head;//指向下一个可放数据的位置 int _tail;//指向下一个可读取数据的位置 public: RQ(int capacity): _cap(capacity), _head(0),_tail(0) { pthread_mutex_init(_mutexp,nullptr); pthread_mutex_init(_mutexc,nullptr); sem_init(_data,0,0); sem_init(_empty,0,_cap); _que.resize(_cap); } ~RQ() { pthread_mutex_destroy(_mutexp); pthread_mutex_destroy(_mutexc); sem_destroy(_data); sem_destroy(_empty); } void push(const T arg) { sem_wait(_empty);//申请空间资源 pthread_mutex_lock(_mutexp);//加锁 _que[_head] arg; _head % _cap; sem_post(_data);//增加一个数据资源 pthread_mutex_unlock(_mutexp);//解锁 } void pop(T value) { sem_wait(_data);//申请数据资源 pthread_mutex_lock(_mutexc);//加锁 value _que[_tail]; _tail%_cap; sem_post(_empty);//增加一个空间资源 pthread_mutex_unlock(_mutexc);//解锁 } }; ----------------------------------------------------------------------------------------- #includefunctional #includepthread.h #includeringqueue.hpp #includeunistd.h using task_t functionvoid(); void* product(void* arg)//生产者的函数 { RQtask_t * buffer static_castRQtask_t*(arg); int x 5; while(1) { cout生产任务endl; task_t mytask [x]()-void{cout任务x执行endl;}; buffer-push(mytask); x; } } void* consume(void* arg)//消费者的函数 { RQtask_t* buffer static_castRQtask_t*(arg); while(1) { sleep(1); task_t task; buffer-pop(task); task(); } } int main() { RQtask_t buffer(5); pthread_t id1; pthread_t id2; pthread_t id3; pthread_t id4; //两个生产者 pthread_create(id1,nullptr,product,buffer); pthread_create(id2,nullptr,product,buffer); //两个消费者 pthread_create(id3,nullptr,consume,buffer); pthread_create(id4,nullptr,consume,buffer); pthread_join(id1,nullptr); pthread_join(id2,nullptr); pthread_join(id3,nullptr); pthread_join(id4,nullptr); }注该类型的生产者消费者模型将缓冲区当作局部使用但也不是完全局部至少没有实现生产者之间的并发并行放入只实现了生产者和消费者之间的并发并行访问。因此同一时间访问缓冲区执行流可以有两个且一个生产者一个消费者。此外它使用信号量实现合理调度。生产者消费者模型究竟为何高效生产者的数据不是凭空而来也不能直接交给缓冲区因此生产者需要较长时间来获取并封装数据消费者拿到数据后也不会什么都不做而是用较长时间来处理数据。由于缓冲区的存在在生产者获取并封装数据的同时就算没有把数据交给消费者消费者也可以从缓冲区里拿取数据并处理数据而在消费者处理数据的同时就算没有处理结束生产者也可以不断获取和处理数据并放入缓冲区。那么显而易见生产者和消费者的操作获取数据封装数据处理数据是并行的整个数据流动的过程宏观上是没有丝毫停滞的对比与串行非常高效。而且对于基于环形阻塞队列的生产者消费者模型来说生产者和消费者是可以同时访问缓冲区的这样一来获取数据封装任务获取任务处理数据整个过程都是并行的效率极其之高。在基于阻塞队列的生产者消费者模型中生产者和消费者对于缓冲区的访问依然是互斥的串行的它的效率是不是就和不用该模型一样呢关于这一点做如下解释由于缓冲区的缓冲作用在生产者获取数据的同时消费者依然可以并行的处理数据因此它依然是高效的宏观上数据流动是没有停滞的。对缓冲区执行拿取和放入操作的时间相比于生产者和消费者处理数据的时间是较少的我们让大部分时间是并发并行而少部分时间串行这同样不失高效性。