引言生产者消费者模型是一种多进程多线程同步互斥问题有三种关系两种角色一个交易场所生产者和生产者之间的互斥关系消费者和消费者之间的互斥关系生产者和消费者之间的互斥和同步关系一个共享资源是特定结构的内存空间可以是阻塞队列或环形队列本质是多个线程高效通信做到解耦合使用queue模拟判断阻塞队列与普通队列的区别在于当队列为空时从队列获取元素操作会被阻塞知道队列放入元素当队列为满时向队列放入元素也会阻塞直到有元素被从队列取出注意STL容器本身不是线程安全的由于设计时保证高效性例如push和pop需要维护同步关系生产和消费关系使用条件变量实现多线程的同步生产者消费者使用同一把锁不同的条件变量只要生产数据就通知消费const int defaultnum 5; using namespace std; templateclass T class BlockQueue { public: BlockQueue(int dn defaultnum/*, int index 0*/):capacity_(dn)/*, index_(index)*/ { pthread_mutex_init(p_mutex_, nullptr); // pthread_mutex_init(c_mutex_, nullptr); low_water_ capacity_ / 4; high_water_ (capacity_ *2)/3; //这里 2/3 等于0所以 high_water_ 也是0 if(low_water_ 0) low_water_ 1; if(high_water_ low_water_) high_water_ capacity_ -1; } void Push(const T data) //生产者生产保证互斥 { pthread_mutex_lock(p_mutex_); while(q_.size() high_water_) //这里插入的时候 0 0 会满足条件wait等待 { pthread_cond_wait(p_cond_, p_mutex_); } q_.push(data); pthread_cond_signal(c_cond_); //唤醒comsumer要在循环外面生产一个数据再消费 pthread_mutex_unlock(p_mutex_); } T Pop() { pthread_mutex_lock(p_mutex_); while(q_.size() 0) { pthread_cond_wait(c_cond_, p_mutex_); } T data q_.front(); q_.pop(); pthread_cond_signal(p_cond_); pthread_mutex_unlock(p_mutex_); return data; } string Getname() { name_ thread ; name_ to_string(index_); return name_; } ~BlockQueue() { pthread_mutex_destroy(p_mutex_); // pthread_mutex_destroy(c_mutex_); pthread_cond_destroy(p_cond_); pthread_cond_destroy(c_cond_); } private: int capacity_; int index_; string name_; std::queueT q_; pthread_mutex_t p_mutex_; // pthread_mutex_t c_mutex_; pthread_cond_t p_cond_; pthread_cond_t c_cond_; int low_water_; int high_water_; };void *Producer(void *args) { pthread_detach(pthread_self()); BlockQueueint *bq static_castBlockQueueint *(args); // 生产数据 while (true) { int data1 rand() % 10 1; bq-Push(data1); cout produced a number : data1 endl; sleep(1); } return nullptr; } void *Consumer(void *args) { pthread_detach(pthread_self()); BlockQueueint *bq static_castBlockQueueint *(args); // 生产数据 while (true) { int ret bq-Pop(); cout Consumed a number : ret endl; sleep(1); } return nullptr; } int main() { srand(time(nullptr) ^ getpid()); // BlockQueueint *bq new BlockQueueint(); // pthread_t c, p; // pthread_create(c, nullptr, Consumer, bq); // pthread_create(p, nullptr, Producer, bq); vectorpthread_t ctids(c_num); vectorpthread_t ptids(p_num); BlockQueueint *bp new BlockQueueint(); for (int i 0; i c_num; i) { // auto* args new auto(bind(bp, 10, i)); pthread_create(ctids[i], nullptr, Consumer, bp); } for (int i 0; i p_num; i) { // auto* args new auto(bind(bp, 10, i)); pthread_create(ptids[i], nullptr, Producer, bp); } cout create success endl; // for (auto i : ctids) // { // pthread_join(i, nullptr); // } // for (auto i : ptids) // { // pthread_join(i, nullptr); // } sleep(5); }情景2指派任务Task.hppconst char *oper -*/%; enum exitcode { DivZero 1, ModZero, UnKnow }; class Task { public: Task(int a, int b, char op) : a_(a), b_(b), op_(op), code_(0) {} void Run() { switch (op_) { case : resault_ a_ b_; break; case -: resault_ a_ - b_; break; case *: resault_ a_ * b_; break; case /: if (b_ 0) { code_ DivZero; } else { resault_ a_ / b_; } break; case %: if (b_ 0) { code_ ModZero; } else { resault_ a_ % b_; } break; default: code_ UnKnow; break; } } void operator()() { Run(); } string GetTask() { string s to_string(a_); s op_; s to_string(b_); s ? ; return s; } string GetRes() { string s to_string(a_); s op_; s to_string(b_); s ; if (code_ ! 0) { switch(code_) { case DivZero: s 除0错误; break; case ModZero: s 模0错误; break; case UnKnow: s 符号错误; break; } } else { s to_string(resault_); s [; s to_string(code_); s ]; } return s; } private: int a_, b_; char op_; int resault_; int code_ 0; };main.ccusing namespace std; const int c_num 3; const int p_num 1; void *Producer(void *args) { pthread_detach(pthread_self()); int opsize 5; BlockQueueTask *bq static_castBlockQueueTask *(args); // 生产数据 while (true) { int data1 rand() % 10 1; usleep(10); int data2 rand() % 3; char op oper[rand() % opsize]; Task t(data1, data2, op); bq-Push(t); cout Produce a task : t.GetTask() endl; sleep(1); } sleep(1); return nullptr; } void *Consumer(void *args) { pthread_detach(pthread_self()); BlockQueueTask *bq static_castBlockQueueTask *(args); // 处理任务 while (true) { Task ret bq-Pop(); ret(); cout Consumer get a resualt : ret.GetRes() endl; sleep(1); } sleep(1); return nullptr; } int main() { srand(time(nullptr) ^ getpid()); // BlockQueueint *bq new BlockQueueint(); // pthread_t c, p; // pthread_create(c, nullptr, Consumer, bq); // pthread_create(p, nullptr, Producer, bq); vectorpthread_t ctids(c_num); vectorpthread_t ptids(p_num); BlockQueueTask *bp new BlockQueueTask(); for (int i 0; i c_num; i) { // auto* args new auto(bind(bp, 10, i)); pthread_create(ctids[i], nullptr, Consumer, bp); } for (int i 0; i p_num; i) { // auto* args new auto(bind(bp, 10, i)); pthread_create(ptids[i], nullptr, Producer, bp); } cout create success endl; // for (auto i : ctids) // { // pthread_join(i, nullptr); // } // for (auto i : ptids) // { // pthread_join(i, nullptr); // } sleep(50); }知识点判断条件变量使用为什么在加锁和解锁之间判断也是访问临界资源伪唤醒会向队列push错误的数据所以要用while循环调用pthread_cond_wait会自动释放锁线程进入等待队列