Article From:https://www.cnblogs.com/xudong-bupt/p/9211087.html

1. background”

c++11It provides better support for threads and conditional variables, and is much more convenient for writing multithreaded programs.
Look at c++’s concurrent programming and write down the learning notes.

3.1 wait

waitIt is used for unconditional wait, where Predicate indicates the check condition and can avoid false wake-up.

unconditional (1)   
void wait (unique_lock<mutex>& lck);
predicate (2)   
template <class Predicate>
  void wait (unique_lock<mutex>& lck, Predicate pred);

3.2 wait for

wait_forYou can specify timeout time, where Predicate indicates the check condition and can avoid false wake-up.

unconditional (1)   
template <class Rep, class Period>
  cv_status wait_for (unique_lock<mutex>& lck,
                      const chrono::duration<Rep,Period>& rel_time);
predicate (2)   
template <class Rep, class Period, class Predicate>
       bool wait_for (unique_lock<mutex>& lck,
                      const chrono::duration<Rep,Period>& rel_time, Predicate pred);

3. thread safety queue sample (producer and consumer model)”

A producer adds data to the queue; multiple consumers read the task from the queue.

#include <mutex>
#include <condition_variable>
#include <queue>
#include <vector>
#include <thread>
#include <iostream>

template<typename T>
class threadsafe_queue
{
private:
    std::mutex mut;
    std::queue<T> data_queue;
    std::condition_variable data_cond;
public:
    threadsafe_queue(){ }

    void push(T new_value) {
        std::lock_guard<std::mutex> lk(mut);
        data_queue.push(new_value);
        data_cond.notify_one();
    } 

    //Limitless waiting
    int pop(T& value) {
        std::unique_lock<std::mutex> lk(mut);
    
        // (1) Wait with judgement condition continues to wait if the condition is not satisfied.
        data_cond.wait(lk,[this]{return !data_queue.empty();}); 

        // (2)waitAfter waking up, you need to judge again to avoid false Awakening
        //while(true){
        //  data_cond.wait(lk);
        //  if (data_queue.empty())
        //      continue;
        //  else
        //      break;
        //} 
        value=data_queue.front();
        data_queue.pop();
        return 0;
    }

    //Limited waiting
    int pop_with_timeout(T& value, int timeout) {
        if (timeout < 0){
            return this->pop(value);
        }

        std::unique_lock<std::mutex> lk(mut);
        //Wait with a timeout time, with a judgment condition
        data_cond.wait_for(lk, std::chrono::milliseconds(timeout), [this] {
                std::cout << "thread id: " << std::this_thread::get_id() << " wait..." << std::endl;
                return !data_queue.empty();}
        );
        if (!data_queue.empty()){
            value=data_queue.front();
            data_queue.pop();
            return 0;
        }
        return -1;
    }

    bool is_empty(){
        std::lock_guard<std::mutex> lk(mut);
        return data_queue.empty();
    }
};

template<typename T>
void consume(threadsafe_queue<T> &queue, bool &stop){
    while(true){
        if (stop && queue.is_empty()){  //Terminate conditions
            break;
        }

        int job_id = 0;
        if (0 == queue.pop_with_timeout(job_id, 3)){
            std::cout << "thread id: " << std::this_thread::get_id() << ", job:" << job_id << std::endl;
        }
        std::this_thread::sleep_for (std::chrono::milliseconds(5));
    }
}

template<typename T>
void product(threadsafe_queue<T> &queue, bool &stop){
    for (auto i = 0; i < 100; ++i){
        queue.push(i);
        std::this_thread::sleep_for (std::chrono::milliseconds(1));
    }
    stop = true;    //Setting the end condition
}
int main(){
    threadsafe_queue<int> my_queue;
    bool stop_flag = false;

    //Producer
    std::thread prod(product<int>, std::ref(my_queue), std::ref(stop_flag));
    //Consumer
    std::vector<std::thread> cons;
    for(auto i = 0; i < 5; ++i){
        std::thread tmp = std::thread(consume<int>, std::ref(my_queue), std::ref(stop_flag));
        cons.emplace_back(std::move(tmp));
    }

    prod.join();
    for(auto &t : cons){
        t.join();
    }
    return 0;
}

Leave a Reply

Your email address will not be published. Required fields are marked *