C++实现生产者与消费者:疯狂的项目经理与不知疲倦的程序猿

前言:HR与程序员的一天

每天到了上班时间,公司里的每个项目经理为完成自己的业绩疯狂,疯狂为公司揽活,交给程序猿完成,程序猿也是不知疲倦的牛马,完成一个活就接着干下一个活,直到HR不再揽活并且HR已经揽的活全都干完才下班
实现这个生产者消费者问题需要确定:

• 某个时间内公司最多能存在多少个商单
• 公司里有几个牛马程序员
• 公司内有多少个项目经理
• 每个项目经理每天要揽多少活
  弄清楚上述几个问题后,明确设计几个类,类中的数据结构都是什么
  画出类图

类图

并生成代码

#ifndef _TASK_H
#define _TASK_H
//项目经理要接的各种任务
class Task {
public: 
virtual ~Task()=0;
void virtual process()=0;
};
class TaskA:public Task{
public:
	~TaskA()override;
    void process() override; // 声明
};


#endif //_TASK_H

//公司某个时刻的任务队列
#ifndef _TASKQUEUE_H
#define _TASKQUEUE_H
#include "Task.h"
#include <iostream>
#include <condition_variable>
#include <queue>
#include <mutex>
using std::queue;
using std::condition_variable;
using std::mutex;
class TaskQueue {
public: 
    size_t m_capacity;
    queue<Task*> m_que;
    mutex m_mtx;
    condition_variable m_queueEmpty;
    condition_variable m_queueFull;
    bool m_accept;

TaskQueue(size_t capa); 
~TaskQueue();
void wakeup();
void push(Task* tasl);  
Task*pop();
bool empty();
bool full();
};
#endif //_TASKQUEUE_H

//项目经理
class Produce{
public:
    void produce(TaskQueue & tq);
}

//牛马程序员
class Consume{
public:
    void consume(TaskQueue &tq);
};

流程图

代码的实现

代码解释,在main()函数中,设置了TaskQueue为3,即公司某个时刻内最多存在3个任务,创建了两个生产者(项目经理)和两个消费者(程序猿),每个项目经理要完成6个商单,程序猿则抢活干

#include "Task.h"
#include <iostream>
using std::cout;
using std::endl;
void TaskA:: process(){
    cout<<"Task A 的做法和内容是 布拉布拉不拉不拉"<<endl;
}
Task::~Task(){
    cout<<"完成了一个任务"<<endl;
}
TaskA::~TaskA (){//在头文件.h中的析构函数中加入override
        cout<<"活A被完成了~~(*^▽^*)"<<endl;         //这里才不会报错
    };

#include "TaskQueue.h"
#include <thread>
#include <functional>
using std::cout;
using std::endl;
using std::unique_lock;
using std::thread;
using std::bind;
using std::lock_guard;
TaskQueue::TaskQueue(size_t capa):
    m_capacity(capa),
    m_que(),
    m_mtx(),
    m_queueEmpty(),
    m_queueFull(),
    m_accept(true)
{}
TaskQueue::~TaskQueue() {
        cout<<"今天的任务全部都派发完了,大伙做完手头的任务就可以下班了"<<endl;
}
void TaskQueue::push(Task* task) {
        unique_lock<mutex> lt(m_mtx);
        while(this->full()&&m_accept){
        m_queueFull.wait(lt);
        }
        if(!m_accept){ //下班后,getOffWork公司不再接受新任务
            delete task;
            return;
        }
    m_que.push(task);
    m_queueEmpty.notify_one();
      //代码结束时,lt析构会自动执行 lt.unlock();
}

Task * TaskQueue::pop() {
            unique_lock<mutex> lt(m_mtx);
           // 使用while防止虚假唤醒
    while (empty() && m_accept) {
        m_queueEmpty.wait(lt);
    }
    // 明确终止条件
    if (!m_accept && empty()) {
        return nullptr;
    }
        Task*tmp=( m_que.front());
        m_que.pop();
        m_queueFull.notify_one();
        //lt.unlock();
    return tmp;
}
bool TaskQueue::empty() {

        return m_que.size()==0;
}
bool TaskQueue::full() {
    return m_que.size()==m_capacity;
}
void TaskQueue::wakeup(){
        m_accept=false;
        m_queueFull.notify_all();
        m_queueEmpty.notify_all();
}
 mutex coutMutex;
class Produce{
public:
    void produce(TaskQueue & tq){
        int cnt=6;
        while(cnt--){
        // tq.m_que.push(new TaskA());错误,会绕过上锁的过程!!
                tq.push(new TaskA());
        {   lock_guard<mutex> lg(coutMutex);
        cout<<"这是生成的第:"<<cnt<<"条命令"<<endl;
        }

        }
    }
    //生成Task,并且加入到任务队列中;
};
class Consume{
public:
    void consume(TaskQueue &tq){
                int completed = 0; // 完成计数器
        while(true) {
            // 核心修改：通过pop()自动处理等待逻辑
            Task* task = tq.pop();
            
            // 终止条件：收到空指针表示队列已关闭且空
            if (!task) break;
            
            // 处理任务
            task->process();
            {
                std::lock_guard<std::mutex> lg(coutMutex);
                std::cout << "✅ 已执行第" << ++completed << "条任务" << std::endl;
            }
            delete task;
        }
        // 最终状态报告
        std::lock_guard<std::mutex> lg(coutMutex);
        std::cout << "消费者退出，共处理" << completed << "个任务" << std::endl;
    }

};
int main(){
    TaskQueue queue(3);
    Consume cs,cs2;
    Produce pd,pd2;
    thread tc(&Consume::consume,&cs,std::ref(queue));
    thread tc2(&Consume::consume,&cs2,std::ref(queue));
    thread tp(&Produce::produce,&pd,std::ref(queue));
    thread tp2(&Produce::produce,&pd2,std::ref(queue));
    tp.join();
    tp2.join();
    queue.wakeup();
    tc.join();
    tc2.join();
}