Linux：死锁与生产者消费者模型解析

目录1.死锁2.生产者消费者模型3.基于BlockingQueue的生产者消费者模型1.死锁死锁是指在一组进程中的各个进程均占有不会释放的资源但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态。上图就是发生了死锁当然一个线程一把锁也可能发生死锁例如一个线程申请同一把锁两次第二次就会把自己阻塞等待发生死锁。死锁四个必要条件都满足才会产生可参考上图互斥条件一个资源每次只能被一个执行流使用请求与保持条件一个执行流因请求资源而阻塞时对已获得的资源保持不放不剥夺条件一个执行流已获得的资源在末使用完之前不能强行剥夺即其他前程不能直接调用 pthread_mutex_unlock() 剥夺然后给自己加锁。循环等待条件若干执行流之间形成一种头尾相接的循环等待资源的关系避免死锁破坏死锁的四个必要条件。破坏互斥条件即不用锁就不会有死锁。破环请求与保持就是申请第二把锁失败后把第一把也释放掉回退到没申请锁之前。循环等待就是一直想要对方的锁资源所以建议按照同样的次序一次申请完锁即第二四点。加锁顺序一致避免锁未释放的场景资源一次性分配对一个临界资源不要重复加锁解锁一次访问完避免死锁算法死锁检测算法(了解)银行家算法了解2.生产者消费者模型互斥能保证资源的安全性但如果当前线程竞争资源的能力较强不断地加锁解锁可能导致其他线程永远访问不到资源所以需要同步来较为高效地使用资源。仓库相当于特定空间的数据结构或者容器。1个交易场所内存空间2种角色生产者与消费者这个模型中还存在3种关系生产者与生产者 - 互斥消费者与消费者 - 互斥生产者与消费者 - 互斥与同步生产者消费者模型的本质是进行线程之间的数据传输的。可以让生产数据和处理数据解耦仓库相当于一段缓存生产者生产好放进去可以继续执行其他操作。消费者也可以执行自己的操作需要数据时再读支持并发。生产者与消费者处理速度不同也没事支持忙闲不均提高处理数据的效率可以看后面代码。条件变量为什么需要条件变量比如我们上面学习到的抢票系统和生产者与消费者模型当消费者不断申请锁访问临界资源票时可能导致生产者一直无法访问临界资源提供票这种问题就需要线程同步让他们有一定的顺序条件变量可以解决这一问题。条件变量是 pthread 库提供的一种一个线程向另一个线程进行事件通知的方式条件变量的类型是pthread_cond_t。条件变量初始化这里也分全局条件变量初始化和局部条件变量初始化全局条件变量程序结束会自动释放局部条件变量需要执行后面的销毁函数。全局定义 pthread_cond_t cond PTHREAD_COND_INITIALIZER; 局部定义 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)等待条件满足调用该函数的进程在该条件变量下等待int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex); 参数 cond要在这个条件变量上等待 mutex互斥量后面详细解释 pthread 的接口成功的时候返回0失败的时候返回真唤醒等待broadcast广播 是条件变量成立唤醒所有线程 int pthread_cond_broadcast(pthread_cond_t *cond); signal 使指定的条件变量成立并唤醒一个线程 int pthread_cond_signal(pthread_cond_t *cond);不引入临界区单纯使用条件变量让多个线程按照顺序执行。pthread_cond_signal 从等待条件变量队列中唤醒一个线程。当然我们控制不了具体执行顺序因为无法判断第一轮的线程谁先申请到锁但后续他们执行顺序都是一样的。显示器打印东西出现混乱本质是显示器也是文件资源而我们对资源没有进行保护显示器也是临界资源。pthread_cond_broadcast 也就是之前学习到的线程一起争抢锁。对于为什么等待条件变量需要传入锁我们后面讲。pthread_cond_t cond PTHREAD_COND_INITIALIZER; pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER; void* threadRoution(void* argv) { std::string name static_castconst char*(argv); while(true) { sleep(1); pthread_mutex_lock(mutex);//可以保证显示器打印不出现混乱显示器 //等待主线程使条件变量成立 pthread_cond_wait(cond,mutex);//????? std::cout I am a new thread : name std::endl; pthread_mutex_unlock(mutex); } } int main() { pthread_t t1,t2,t3; pthread_create(t1,nullptr,threadRoution,(void*)thread-1); pthread_create(t2,nullptr,threadRoution,(void*)thread-2); pthread_create(t3,nullptr,threadRoution,(void*)thread-3); sleep(5);//5s之后开始让条件变量成立 while (true) { pthread_cond_signal(cond);//唤醒一个线程 //pthread_cond_broadcast(cond);//全部唤醒之后 没有顺序 sleep(1); } pthread_join(t1,nullptr); pthread_join(t2,nullptr); pthread_join(t3,nullptr); return 0; }线程访问临界资源首先需要锁再判断有没有资源。如何保证顺序在加锁和解锁中等待条件变量成立不成立会在一个等待条件变量队列中等待释放锁其他线程因此会得到锁按顺序进来再排队阻塞直到条件变量成立依次唤醒醒来时依旧需要申请锁。pthread_cond_t cond PTHREAD_COND_INITIALIZER; pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER; int tickets 10000; void* threadRoution(void* argv) { std::string name static_castconst char*(argv); while(true) { pthread_mutex_lock(mutex);//临界资源:显示器和tickets if(tickets 0) { std::cout name I am a new thread : tickets std::endl; tickets--; usleep(1000); } else { std::cout 没有票了 name std::endl; pthread_cond_wait(cond,mutex);//????? } pthread_mutex_unlock(mutex); 就是每一个线程在大量地申请锁释放锁浪费锁资源 } }每一个线程在大量地申请锁释放锁访问临界资源浪费锁资源就有可能导致生产者得到锁无法提供资源所以我们应该让这些消费者线程去等条件变量不是锁等有票了再去访问临界资源就会阻塞这些线程。pthread_cond_wait(cond,mutex)我们在主线程提供加票之后加的1000张票每次被一个线程6秒内抢完但总体他们有顺序int main() { pthread_t t1,t2,t3; pthread_create(t1,nullptr,threadRoution,(void*)thread-1); pthread_create(t2,nullptr,threadRoution,(void*)thread-2); pthread_create(t3,nullptr,threadRoution,(void*)thread-3); sleep(5);//开始执行唤醒 while (true) { sleep(6); pthread_mutex_lock(mutex); tickets 1000;// pthread_mutex_unlock(mutex); 这里会一个全部抢完因为6秒内其他线程在队列中还没有被唤醒 pthread_cond_signal(cond); //pthread_cond_broadcast(cond);//全部唤醒之后 没有顺序 } pthread_join(t1,nullptr); pthread_join(t2,nullptr); pthread_join(t3,nullptr); return 0; }所以单纯的互斥能保证数据安全不一定合理或者高效也需要同步pthread_cond_wait可以让线程在等待的时候会自动释放锁其他线程可以申请锁所以需要传入一把锁。线程被唤醒的时候是在临界区被唤醒的当线程被唤醒在 pthread_cond_wait 返回的时候要重新申请并持有锁。当线程被唤醒的时候重新申请并持有锁本质也是要重新参与锁的竞争所以 pthread_cond_broadcast 唤醒后是乱序的。3.基于BlockingQueue的生产者消费者模型在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于当队列为空时消费线程不能再消费当队列满时生产线程不能再生成。伪唤醒问题对应条件不足但是线程却被唤醒了。例如我们采用 pthread_cond_broadcast 唤醒策略当有一个线程消费完唤醒其他消费者释放锁其他消费者都可能会逐渐按照 pthread_mutex_lock 获得锁没有条件变量唤醒时调用 pthreadread_cond_wait 的方式释放锁都在 pthread_cond_wait 处阻塞当只有最后一个资源被消费后其他线程就不会在 pthread_cond_wait 处继续阻塞了但此时已经没有资源了所以会发生问题需要将 if 改成 while因为while本身也具有条件判断的能力。生产者也是同样的道理我们可以看下面代码。自己实现的BlockingQueue.hpp:#pragma once #include iostream #include queue #include pthread.h const int defaultcap 5; templateclass T struct BlockQueue { public: BlockQueue(int cap defaultcap):_capacity(cap) { pthread_mutex_init(_mutex,nullptr); pthread_cond_init(_p_cond,nullptr); pthread_cond_init(_c_cond,nullptr); } bool Push(const Tin)//生产者 { pthread_mutex_lock(_mutex); while(IsFull()) { //阻塞等待 pthread_cond_wait(_p_cond,_mutex); } //不加else也行 _q.push(in); //if(_q.size() _productor_water_line) pthread_cond_signal(_c_cond);//有产品唤醒对方消费 //当对方已经醒着的时候会忽略这个信号 //在锁内外唤醒都行因为消费者消费完等待时会释放锁。 pthread_mutex_unlock(_mutex); return true; } bool Pop(T* out)//消费者 { pthread_mutex_lock(_mutex); while(IsEmpty()) { //阻塞等待 pthread_cond_wait(_c_cond,_mutex); } //不加else也行 *out _q.front(); _q.pop(); pthread_cond_signal(_p_cond);//消费产品了唤醒对方生产 pthread_mutex_unlock(_mutex); return true; } bool IsFull() { return _q.size() _capacity; } bool IsEmpty() { return _q.size() 0; } ~BlockQueue() { pthread_mutex_destroy(_mutex); pthread_cond_destroy(_p_cond); pthread_cond_destroy(_c_cond); } private: std::queueT _q;//要有容量上限STL int _capacity;//_q.size() _capacity 满了 pthread_mutex_t _mutex; //多线程访问需要对queue保护 //防止一方线程不断地申请访问锁另一方申请不到例如满了之后消费者无法消费 pthread_cond_t _p_cond;//没了之后 让生产者等待 消费者给生产者 pthread_cond_t _c_cond;//满了之后 让消费者等待 生产者给消费者 //两个线程互相唤醒 也可以定制唤醒策略比如达到一定程度再唤醒 //int _productor_water_line _capacity / 3; };重点是对临界区加锁然后同步通过双方互相 pthread_cond_signal 发送条件变量唤醒。对已经唤醒的线程发送唤醒信号是会被忽略的不影响。LockGuard也可以在这里使用不用调用解锁的函数了及时 pthread_cond_signal 中解锁了唤醒时还是会上锁的不用担心。测试代码Main.cc一个生产者一个消费者。#include BlockQueue.hpp #include iostream #include unistd.h #include pthread.h #include time.h #include cstdint void* consumer(void* argv) { //生产者与消费者同时看到了一份公共资源 BlockQueueint* bq static_castBlockQueueint*(argv); while(true) { int data 0; bq-Pop(data); std::cout consumer data: data std::endl; //消费者没有sleep sleep(1); } } void* productor(void* argv) { BlockQueueint* bq static_castBlockQueueint*(argv); while(true) { int data rand() % 10 1; //[1,10] bq-Push(data); std::cout prodector data: data std::endl; //sleep(1); } } int main() { srand((uint16_t)time(nullptr)^getpid()^pthread_self());//更随机 BlockQueueint *bq new BlockQueueint(); pthread_t c, q; pthread_create(c,nullptr,consumer,bq); pthread_create(q,nullptr,productor,bq); pthread_join(c,nullptr); pthread_join(q,nullptr); return 0; }因为阻塞队列是队列所以这里也是 FIFO。生产者与消费者模型是如何提高效率的一定不要忽略生产前和消费后还需要处理其他事情也要时间比如后面要学习网络我们从网络中取数据也需要时间这个时间消费者就可以同时处理之前从网络获取的数据并发就体现出来了。实现多生产者多消费者我们写的 BlockQueue 已经可以实现多生产者多消费者只需要多创建几个生产者线程和消费者线程。我们也可以通过传入结构体的方式传入更多信息多传入一个线程名字。Main.cc#include BlockQueue.hpp #include iostream #include unistd.h #include pthread.h #include time.h #include cstdint #include Task.hpp class ThreadData { public: BlockQueueTask* bq; std::string name; }; void* consumer(void* argv) { //生产者与消费者同时看到了一份公共资源 ThreadData* ctd static_castThreadData*(argv); BlockQueueTask* bq ctd-bq;//阻塞队列 std::string name ctd-name; while(true) { //sleep(1); Task t; bq-Pop(t); //t.Run(); t();//仿函数 std::cout consumer data: t.PrintResult() name std::endl; } } void* productor(void* argv) { ThreadData* ptd static_castThreadData*(argv); BlockQueueTask* bq ptd-bq; std::string name ptd-name; while(true) { int data1 rand() % 10; //[1,10] usleep(rand()%123);//让随机数变化一下 int data2 rand() % 10; usleep(rand()%123); char oper rand() % opers.size(); Task t(data1,data2,opers[oper]); //for debug std::cout prodector task: t.PrintTask() name std::endl; bq-Push(t); sleep(1); } } int main() { srand((uint16_t)time(nullptr)^getpid()^pthread_self());//更随机 BlockQueueTask *bq new BlockQueueTask();//不能直接传入可以扩容的queue因为不是线程安全的 pthread_t c[3], p[2]; //3个消费者 ThreadData* ctd new ThreadData; ctd-bq bq; ctd-name c thread - 1 ; pthread_create(c[0],nullptr,consumer,ctd); ThreadData* ctd1 new ThreadData; ctd1-bq bq; ctd1-name c thread - 2 ; pthread_create(c[1],nullptr,consumer,ctd1); ThreadData* ctd2 new ThreadData; ctd2-bq bq; ctd2-name c thread - 3 ; pthread_create(c[2],nullptr,consumer,ctd2); ThreadData* ptd1 new ThreadData; ptd1-bq bq; ptd1-name p thread - 1 ; pthread_create(p[0],nullptr,productor,ptd1); ThreadData* ptd2 new ThreadData; ptd2-bq bq; ptd2-name p thread - 2 ; pthread_create(p[0],nullptr,productor,ptd1); pthread_join(c[0],nullptr); pthread_join(p[0],nullptr); pthread_join(c[1],nullptr); pthread_join(p[1],nullptr); pthread_join(c[2],nullptr); return 0; }Task.cc:#pragma once #include string const int defaultvalue 0; enum { ok 0, div_zero, mod_zero, unknow }; const std::string opers -*/%(); class Task { public: Task(){}; Task(int x,int y,char op) :data_x(x),data_y(y),oper(op),result(defaultvalue),code(ok) {} void Run() { switch (oper) { case : result data_x data_y; break; case -: result data_x - data_y; break; case *: result data_x * data_y; break; case /: { if (data_y 0) code div_zero; else result data_x / data_y; } break; case %: { if (data_y 0) code mod_zero; else result data_x / data_y; } break; default: code unknow; break; } } void operator()() { Run(); } std::string PrintTask() { std::string s; s std::to_string(data_x); s oper; s std::to_string(data_y); s ?; return s; } std::string PrintResult() { std::string s; s std::to_string(data_x); s oper; s std::to_string(data_y); s ; s std::to_string(result); s [; s std::to_string(code); s]; return s; } int data_x; int data_y; char oper; int result; int code; };输出结果为什么这里消费者线程调度会有一定的顺序生产者线程调度没有呢因为最开始我们的 BlockQueue是空的我们无论哪个消费者进入都会在 pthread_cond_wait 出排队等待释放锁所以当有资源是会 FIFO 调度而生产就没有这个排队限制会竞争锁。如果 BlockQueue 不为空没有条件变量的限制消费者能直接执行完释放锁也是无序的。本篇结束