安全线程队列 — C++

一、安全队列

#pragma once
#include <queue>
#include <memory>
#include <mutex>
#include <condition_variable>

template <typename T>
class SafeQueue
{
private:
    mutable std::mutex mMutex;
    std::queue<T> mQueue;
    std::condition_variable mCond;

public:
    SafeQueue() {}
    SafeQueue(SafeQueue const &other)
    {
        std::lock_guard<std::mutex> lk(other.mMutex);
        mQueue = other.mQueue;
    }

    void push(T new_value)
    {
        std::lock_guard<std::mutex> lk(mMutex);
        mQueue.push(new_value);
        mCond.notify_one();
    }

    void waitAndPop(T &value)
    {
        std::unique_lock<std::mutex> lk(mMutex);
        mCond.wait(lk, [this]() { return !mQueue.empty(); });
        value = mQueue.front();
        mQueue.pop();
    }

    std::shared_ptr<T> waitAndPop()
    {
        std::unique_lock<std::mutex> lk(mMutex);
        mCond.wait(lk, [this]()  { return !mQueue.empty(); });
        std::shared_ptr<T> res(std::make_shared<T>(mQueue.front()));
        mQueue.pop();
        return res;
    }

    bool tryPop(T &value)
    {
        std::lock_guard<std::mutex> lk(mMutex);
        if (mQueue.empty())
            return false;
        value = mQueue.front();
        mQueue.pop();
        return true;
    }

    std::shared_ptr<T> tryPop()
    {
        std::lock_guard<std::mutex> lk(mMutex);
        if (mQueue.empty())
            return std::shared_ptr<T>();
        std::shared_ptr<T> res(std::make_shared<T>(mQueue.front()));
        mQueue.pop();
        return res;
    }

    bool empty() const
    {
        std::lock_guard<std::mutex> lk(mMutex);
        return mQueue.empty();
    }
};

二、安全队列一般使用在多线程调用，容易出现调用错乱，为了能同步调用，把每一个回调都根据先后顺序放在一个安全队列中执行

三、举例使用安全队列
在一个类中有多个回调通知。

enum class ManagerEventType
{
    NOTIFY_Base_1_RESULT,
    NOTIFY_Base_2_Begin,
    NOTIFY_Base_3_END,
};

struct Slot
{
    std::string area;
    std::string key;
    std::string value;
};

struct ManagerEvent
{
    ManagerEventType eventType;
    Slot slot;
    bool status;
};

class Manager : public RESULTCallback, 
                public BeginCallback, 
                public ENDCallback
{
public：
 // 初始化mInitThreadExit标志位线程队列是否退出
Manager::Manager()
    : mInitThreadExit(false),
      mEventQueue(),
      mWorkThread(nullptr) {

}

// 初始化的时候执行这个同步线程。
void AISpeechManager::init()
{
    VLOGD("AISpeechManager::init() -> start");
    mWorkThread = std::make_shared<std::thread>(&AISpeechManager::workLoop, this);
}
 
AISpeechManager::~AISpeechManager()
{
    if (mInitThreadExit == false) {
        mInitThreadExit = true;
          // 等待线程执行结束之后再退出，joinable检测是否用过join，没有用过在使用join
        if ((mWorkThread != nullptr) && (mWorkThread->joinable())) {
            mWorkThread->join();
        }
    }
}

private:
   Slot mslot；
   std::atomic<bool> mInitThreadExit{false};
   SafeQueue<ManagerEvent> mEventQueue;
   std::shared_ptr<std::thread> mWorkThread;

private:

   // 异步通知过来先统一扔进安全队列中等待上一个执行结束之后会通知先一个调用执行
   void resulte() {
         mslot.area = "学号"
          mslot.key = "123"
          mslot.value = "liming"
          ManagerEvent event;
          event.eventType = ManagerEventType::NOTIFY_Base_1_RESULT;
          event.slot= mslot;
          sendEvent(event);
    }
    
    void beginSts(bool beginSts){
          ManagerEvent event;
          event.eventType = ManagerEventType::NOTIFY_Base_2_Begin;
          event.status = begin;
          sendEvent(event);
    }
    void end(){
          ManagerEvent event;
          event.eventType = ManagerEventType::NOTIFY_Base_3_end;
          event.status = begin;
          sendEvent(event);
   }
   
    void sendEvent(ManagerEvent event) {
       mEventQueue.push(event);
    }
    
    void workLoop() {
       while (!mInitThreadExit) {
        ManagerEvent event;
        VLOGD("Manager Begin wait for event");
        mEventQueue.waitAndPop(event);
        VLOGD("Manager Receive event:%d", event.eventType);
        switch (event.eventType) {
            // 在下面case中执行扔进线程队列中的方法要做的事情
            case ManagerEventType::NOTIFY_Base_2_Begin: { 
                if (event.status) {
                    // status is ture do something
                } else {
                    // status is false do something
                }
            }
            case ManagerEventType::NOTIFY_Base_1_RESULT: { 
                 Slot slot = event.slot;
                 // do something
            }
            case ManagerEventType::NOTIFY_Base_3_end:{ 
                 // do something
            }
    }
}