C++11 實現生產者消費者雙緩沖

基礎的生產者消費者模型，生產者向公共緩存區寫入數據，消費者從公共緩存區讀取數據進行處理，兩個線程訪問公共資源，加鎖實現數據的一致性。

通過加鎖來實現

class Produce_1 {  2 public:  3     Produce_1(std::queue<int> * que_, std::mutex * mt_) {  4         m_mt = mt_;  5         m_que = que_;  6         m_stop = false;  7  }  8     void runProduce() {  9         while (!m_stop) { 10             std::this_thread::sleep_for(std::chrono::seconds(1)); 11             std::lock_guard<std::mutex> lgd(*m_mt); 12             m_que->push(1); 13             std::cout << "Produce_1 produce 1" << std::endl; 14  } 15  } 16     void join() { 17         m_trd->join(); 18  m_trd.reset(); 19  } 20     void start() { 21         m_trd.reset(new std::thread(std::bind(std::mem_fun(&Produce_1::runProduce), this))); 22  } 23     void stop() { 24         m_stop = true; 25  } 26 private: 27     std::mutex * m_mt; 28     std::queue<int> * m_que; 29     volatile bool m_stop; 30     std::shared_ptr<std::thread> m_trd; 31 }; 32 

/*
*單緩沖一個同步隊列 效率較低 36 */
class Consume_1 { 38 public: 39     Consume_1(std::queue<int> * que_, std::mutex * mt_) { 40         m_mt = mt_; 41         m_que = que_; 42         m_stop = false; 43  } 44 
    void runConsume() { 46         while (!m_stop) { 47             std::this_thread::sleep_for(std::chrono::seconds(1)); 48             std::lock_guard<std::mutex> lgd(*m_mt); 49             if (!m_que->empty()) { 50                 m_que->pop(); 51  } 52             std::cout << "Consume_1 consume" << std::endl; 53  } 54  } 55     void join() { 56         m_trd->join(); 57  m_trd.reset(); 58  } 59     void start() { 60         m_trd.reset(new std::thread(std::bind(std::mem_fun(&Consume_1::runConsume), this))); 61  } 62     void stop() { 63         m_stop = true; 64  } 65 private: 66     std::mutex * m_mt; 67     std::queue<int> * m_que; 68     volatile bool m_stop; 69     std::shared_ptr<std::thread> m_trd; 70 };

 

通過條件變量來實現

typedef struct Mutex_Condition{  2  std::mutex mt;  3  std::condition_variable cv;  4 }Mutex_Condition;  5 
class Produce {  7 public:  8     Produce(std::queue<int> * que_, Mutex_Condition * mc_) {  9         m_que = que_; 10         m_mc = mc_; 11         m_stop = false; 12  } 13     void join() { 14         m_trd->join(); 15  m_trd.reset(); 16  } 17     void produce(int enter) { 18         std::lock_guard<std::mutex> lgd(m_mc->mt); 19         m_que->push(enter); 20         m_mc->cv.notify_one(); 21  } 22 
    void runProduce() { 24         while (!m_stop) { 25             std::this_thread::sleep_for(std::chrono::seconds(1)); 26             produce(1); 27             std::cout << "Produce Thread produce 1 " << std::endl; 28  } 29  } 30 
    void start() { 32         m_trd.reset(new std::thread(std::bind(std::mem_fun(&Produce::runProduce), this))); 33  } 34     void stop() { 35         m_stop = true; 36  } 37 
private: 39     std::queue<int> * m_que; 40     Mutex_Condition * m_mc; 41     std::shared_ptr<std::thread> m_trd; 42     volatile bool m_stop; 43 }; 44 

class Consume { 47 public: 48     Consume(std::queue<int> * que_, Mutex_Condition * mc_) { 49         m_que = que_; 50         m_mc = mc_; 51         m_stop = false; 52  } 53     void join() { 54         m_trd->join(); 55  m_trd.reset(); 56  } 57     void consume() { 58         std::unique_lock<std::mutex> lgd(m_mc->mt); 59         while (m_que->empty()) { 60             int i = 0; 61             m_mc->cv.wait(lgd); 62  } 63         m_que->pop(); 64         std::cout << "Consume Thread consume " << std::endl; 65  } 66     void runConsume() { 67         while (!m_stop) { 68             std::this_thread::sleep_for(std::chrono::seconds(1)); 69  consume(); 70  } 71  } 72     void start() { 73         m_trd.reset(new std::thread(std::bind(std::mem_fun(&Consume::runConsume), this))); 74  } 75     void stop() { 76         m_stop = true; 77  } 78 
private: 80     std::queue<int> * m_que; 81     Mutex_Condition * m_mc; 82     std::shared_ptr<std::thread> m_trd; 83     volatile bool m_stop; 84 
};

 

二、生產者消費者-雙緩沖

一個公共緩存區，由於多線程訪問的鎖沖突較大，可以采取雙緩沖手段來解決鎖的沖突

雙緩沖的關鍵：雙緩沖隊列的數據交換

1）生產者線程不斷的向生產者隊列A寫入數據，當隊列中有數據時，進行數據的交換，交換開始啟動時通過條件變量通知交換線程來處理最先的數據交換。

2）數據交換完成后，通過條件變量通知消費者處理數據，此時交換線程阻塞到消費者數據處理完成時通知的條件變量上。

3）消費者收到數據交換后的通知后，進行數據的處理，數據處理完成后，通知交換線程進行下一輪的雙緩沖區的數據交換。

要點：

生產者除了在數據交換時，其余時刻都在不停的生產數據。

數據交換隊列需要等待消費者處理數據完成的通知，以進行下一輪交換。

消費者處理數據時，不進行數據交換，生產者同時會不斷的生產數據，消費者需要等待數據交換完成的通知，並且發送消費完成的通知給交換線程

 

 使用條件變量的版本實現

 

typedef struct Mutex_Condition{
    std::mutex mt;
    std::condition_variable cv;
}Mutex_Condition;

class Produce_1 {
public:
    Produce_1(std::queue<int> * que_1, std::queue<int> * que_2, Mutex_Condition * mc_1 , Mutex_Condition * mc_2) {
        m_read_que   = que_1;
        m_writer_que = que_2;
        m_read_mc    = mc_1;
        m_writer_mc  = mc_2;
        m_stop       = false;

    }
    void runProduce() {
        while (!m_stop) {
            std::this_thread::sleep_for(std::chrono::microseconds(20 * 1000));
            std::lock_guard<std::mutex> lgd(m_writer_mc->mt);
            m_writer_que->push(1);
            m_writer_mc->cv.notify_one();
            std::cout << "m_writer push" << std::endl;
        }
        
    }
    void join() {
        m_trd->join();
        m_trd.reset();
    }
    void start() {
        m_trd.reset(new std::thread(std::bind(std::mem_fun(&Produce_1::runProduce), this)));
    }
    void stop() {
        m_stop = true;
    }
private:
    Mutex_Condition * m_read_mc;
    Mutex_Condition * m_writer_mc;
    std::queue<int> * m_read_que;
    std::queue<int> * m_writer_que;
    volatile bool m_stop;
    std::shared_ptr<std::thread> m_trd;
};


class Consume_1 {
public:
    Consume_1(std::queue<int> * que_1, std::queue<int> * que_2, Mutex_Condition * mc_1,Mutex_Condition * mc_2,Mutex_Condition * switch_mc) {
        m_read_que    = que_1;
        m_writer_que  = que_2;
        m_read_mc     = mc_1;
        m_writer_mc   = mc_2;
        m_stop        = false;
        m_switch_mc = switch_mc;
    }

    void runConsume() {
        while (!m_stop) {
            while (true) {
                std::unique_lock<std::mutex> ulg(m_read_mc->mt);
                while (m_read_que->empty()) {
                    m_read_mc->cv.wait(ulg);
                }
                //deal data
                //std::lock_guard<std::mutex> ulg(m_read_mc->mt);
                while (!m_read_que->empty()) {
                    m_read_que->pop();
                    std::cout << "m_read_queue pop" << std::endl;
                }
                m_switch_mc->cv.notify_one();
            }
        }
    }
    void join() {
        m_trd->join();
        m_trd.reset();
    }
    void start() {
        m_trd.reset(new std::thread(std::bind(std::mem_fun(&Consume_1::runConsume), this)));
    }
    void stop() {
        m_stop = true;
    }
private:
    Mutex_Condition * m_read_mc;
    Mutex_Condition * m_writer_mc;
    Mutex_Condition * m_switch_mc;
    std::queue<int> * m_read_que;
    std::queue<int> * m_writer_que;
    volatile bool m_stop;
    std::shared_ptr<std::thread> m_trd;
};
void que_switch_trd(std::queue<int> * read_que, std::queue<int> * writer_que, Mutex_Condition * read_mc, Mutex_Condition * writer_mc,Mutex_Condition * switch_mc) {
    while (true) {
        {
            std::unique_lock<std::mutex> ulg(writer_mc->mt);
            while (writer_que->empty()) {
                writer_mc->cv.wait(ulg);
            }
            std::lock_guard<std::mutex> ulg_2(read_mc->mt);
            std::swap(*read_que, *writer_que);
            std::cout << "switch queue" << std::endl;
            if (!read_que->empty()) {
                read_mc->cv.notify_one();
            }
        }
        std::unique_lock<std::mutex> ulg_2(switch_mc->mt);
        while (!read_que->empty()) {
            switch_mc->cv.wait(ulg_2);
        }
    }
}
int main(){

    Mutex_Condition mc_1;
    Mutex_Condition mc_2;
    Mutex_Condition mc_3;
    std::queue<int> que_1;
    std::queue<int> que_2;

    Produce_1 produce_1(&que_1, &que_2, &mc_1, &mc_2);
    Consume_1 consume_1(&que_1, &que_2, &mc_1, &mc_2,&mc_3);

    std::thread trd(std::bind(&que_switch_trd, &que_1, &que_2, &mc_1, &mc_2,&mc_3));
    produce_1.start();
    consume_1.start();
    
    produce_1.join();
    consume_1.join();
    trd.join();

    return 0;
}

 

 

 

使用互斥鎖的實現

#include<mutex>
#include<thread>
#include<queue>
#include<iostream>
#include<chrono>

class DBQueue{
public:
    void push(int i_) {
        std::lock_guard<std::mutex> lock(m_mt);
        std::cout << "write_que push " << i_ << std::endl;
        m_write_que.push(i_);
    }
    void swap(std::queue<int> & read_que) {
        std::lock_guard<std::mutex> lock(m_mt);
        std::swap(m_write_que,read_que);
        std::cout << "switch swap" << std::endl;
    }
private:
    std::queue<int> m_write_que;
    std::mutex m_mt;
};
void produce(DBQueue * que) {
    while (true) {
        std::this_thread::sleep_for(std::chrono::microseconds(20*1000));
        que->push(1);
    }
}
void consume(DBQueue * que) {
    std::queue<int> read_que;
    while (true) {
        std::this_thread::sleep_for(std::chrono::microseconds(20*1000));
        if (read_que.empty()) {
            que->swap(read_que);
            //xxoo
            while (!read_que.empty()) {
                std::cout << "read_que pop" << std::endl;
                read_que.pop();
            }
        }
    }
}
int main()
{
    DBQueue que;
    std::thread trd_1(std::bind(&produce, &que));
    std::thread trd_2(std::bind(&consume, &que));
    trd_1.join();
    trd_2.join();
    return 0;
}

 兩個版本的區別 sleep的區別，sleep處理的時效性較差，不加sleep，cpu占用率又比較高，所以條件變量是比較好的選擇。