[C++知识库]消息传递解决生产者-消费者问题(C++)

Util.h

#pragma once
#include<string>
#include<vector>
#include<mutex>
#include<condition_variable>

#define Max 100

static const std::vector<std::string>Fruit
{
    "apple",
    "pair",
    "orange"
};

struct Message
{
    std::string message_name;
    unsigned short message_no;
    Message(const std::string& name,const unsigned short& no)
        :message_name(std::move(name)),message_no(no)
        {}
};

class Util
{
public:
    static std::mutex mu;
    static std::mutex mu_for_pop;
    static std::condition_variable cv;
};
std::mutex Util::mu{};
std::mutex Util::mu_for_pop{};
std::condition_variable Util::cv{};

工具类：提供条件变量与锁、消息的结构体

Producer.h

#pragma once
#include<iostream>
#include<cstdio>
#include<queue>
#include<ctime>
#include<memory>
#include"Util.h"

class Producer
{
public:
    Producer() = default;
public:
    void Send(std::queue<std::unique_ptr<Message>>&,const size_t&);
    std::unique_ptr<Message> Receive();
    const size_t PMailSize() const;
public:
    static std::queue<std::unique_ptr<Message>>p_buffer;
};
std::queue<std::unique_ptr<Message>> Producer::p_buffer{};

inline const size_t Producer::PMailSize() const{
    return Producer::p_buffer.size();
}

std::unique_ptr<Message> Producer::Receive(){
    std::unique_ptr<Message> message;
    if(!Producer::p_buffer.empty())
    {
        message = std::move(Producer::p_buffer.front());
        Producer::p_buffer.front().~unique_ptr();
        Producer::p_buffer.pop();
    }
    return std::move(message);
}

void Producer::Send(std::queue<std::unique_ptr<Message>>& c_mailbox,const size_t& c_mailbox_size){
    srand((unsigned)time(NULL));
    if(c_mailbox_size == Max)
    {
        std::cout<<"Producer thread will lock!\n";
        std::unique_lock<std::mutex>lock(Util::mu);
        Util::cv.wait(lock);
    }
    size_t message_num = rand()% Producer::p_buffer.size();
    for(size_t i = 0;i < message_num;++i){
        auto message = Receive();
        printf("Now Producer produced a message!\n");
        message.reset(new Message(Fruit.at(i % 3),2000 + i));
        c_mailbox.emplace(std::move(message));
    }
    printf("Producer's buffer size:%ld,Consumer's buffer size:%ld\n",p_buffer.size(),c_mailbox.size());
}

Consumer.h

#pragma once
#include<iostream>
#include<cstdio>
#include<queue>
#include<memory>
#include"Util.h"

class Consumer
{
public:
    Consumer()
    {
        for(int i = 0;i < Max;++i){
            auto empty_message = std::make_unique<Message>("null",2000+i);
            Consumer::c_buffer.emplace(std::move(empty_message));
        }
    }
public:
    void Send(std::queue<std::unique_ptr<Message>>&,const size_t&);
    std::unique_ptr<Message> Receive();
    const size_t CMailSize() const;
public:
    static std::queue<std::unique_ptr<Message>> c_buffer;
};
std::queue<std::unique_ptr<Message>> Consumer::c_buffer{};

inline const size_t Consumer::CMailSize() const{
    return Consumer::c_buffer.size();
}

std::unique_ptr<Message> Consumer::Receive(){
    std::unique_ptr<Message> message;
    if(!Consumer::c_buffer.empty())
    {
        message = std::move(Consumer::c_buffer.front());
        Consumer::c_buffer.front().~unique_ptr();
        Consumer::c_buffer.pop();
    }
    return std::move(message);
}

void Consumer::Send(std::queue<std::unique_ptr<Message>>& p_mailbox,const size_t& p_mailbox_size){
    srand((unsigned)time(NULL));
    //std::cout<<"Consumer::Send is called\n";
    if(p_mailbox_size == Max)
    {
        std::cout<<"Consumer thread will lock!\n";
        std::unique_lock<std::mutex>lock(Util::mu);
        Util::cv.wait(lock);
    }
    size_t message_num = rand()% Consumer::c_buffer.size();
    for(size_t i = 0;i < message_num;++i){
        auto message = Receive();
        printf("Now Consumer consumed a message!message_name is %s,message_no is %d\n",message->message_name.c_str(),message->message_no);
        message.reset(new Message("null",2000+i));
        p_mailbox.emplace(std::move(message));
    }
    printf("Consumer's buffer size:%ld,Peoducer's buffer size:%ld\n",c_buffer.size(),p_mailbox.size());
}

MessageTransmit.cpp

#include"Producer.h"
#include"Consumer.h"
#include<thread>
#include<unistd.h>

class MessageTransmit
{
public:
    MessageTransmit() = default;
public:
    void ProducerThread();
    void ConsumerThread();
private:
    Producer producer;
    Consumer consumer;
};

void MessageTransmit::ProducerThread(){
    //consumer.Send(Producer::p_buffer,producer.PMailSize());//to start thread
    while(true)
    {
        //std::cout<<"Producer buffer size:"<<Producer::p_buffer.size()<<std::endl;
        producer.Send(Consumer::c_buffer,consumer.CMailSize());
        if(consumer.CMailSize() > 0 && consumer.CMailSize() < Max){
            Util::cv.notify_one();
            //std::cout<<"Consumer is unlocked\n";
        }
        //sleep(1);
    }
}

void MessageTransmit::ConsumerThread(){
    while(true)
    {
        //std::cout<<"Consumer buffer size:"<<Consumer::c_buffer.size()<<std::endl;
        consumer.Send(Producer::p_buffer,producer.PMailSize());
        if(producer.PMailSize() > 0 && producer.PMailSize() < Max){
            Util::cv.notify_one();
            //std::cout<<"Producer is unlocked\n";
        }
        //sleep(1);
    }
}

int main(){
    MessageTransmit messagetransmit;
    std::thread t_producer(&MessageTransmit::ProducerThread,messagetransmit);
    std::thread t_consumer(&MessageTransmit::ConsumerThread,messagetransmit);
    t_producer.join();
    t_consumer.join();
    return 0;
}

生产者与消费者各含有一个信箱，双方都向对方的信箱发送消息。如果信箱内的消息数量达到最大值，则挂起该线程。

部分运行结果：