c++ 11 std::thread std::mutex等实现线程队列演示例子！

Dear 丶 2023-06-28 13:28 13阅读 0赞

#include <iostream>           // std::cout
    #include <thread>             // std::thread
    #include <mutex>              // std::mutex, std::unique_lock
    #include <condition_variable> // std::condition_variable
    #include <vector>
    #include <functional>
    #include <queue>
    #include <atomic>
    #include <wtypes.h>
    #include <stdio.h>
    
    
    class CInterFace 
    {
    public:
    	virtual void OnCallBack(int nData)= 0 ;
    };
    
    
    class CTaskQuery
    {
    public:
    	CTaskQuery();
    	~CTaskQuery();
    	void init(CInterFace *pCallback);
    
    	void AddTask(int nTaskId);
    	static bool doWork(void *param);
    	bool OnTaskWork();
    	bool Start();
    	bool Stop();
    private:
    	std::thread				 m_threadTask;
    	std::mutex				 m_mNotifyLock; //等待锁[不可重入锁 ]
    	std::recursive_mutex	 m_dataLock;    //数据锁[可重入 递归锁] 
    	std::condition_variable  m_cvNotify;    //通知事件
    	std::queue<int>			 m_TaskQue;
    	CInterFace *			 m_pCallback = nullptr;
    	std::atomic_bool		 m_bIsQuit = false;  //原子操作 退出任务
    };
    
    CTaskQuery::CTaskQuery()
    {
    	m_pCallback = nullptr;
    	m_threadTask = std::thread(doWork,this);
    
    }
    
    CTaskQuery::~CTaskQuery()
    {
    	Stop();
    }
    
    void CTaskQuery::init(CInterFace *pCallback)
    {
    	m_pCallback = pCallback;
    }
    
    void CTaskQuery::AddTask(int nTaskId)
    {
    	{
    		std::lock_guard<std::recursive_mutex> ltmp(m_dataLock);
    		m_TaskQue.push(nTaskId);
    	}
    	
    	Start();
    }
    
    bool CTaskQuery::doWork(void *param)
    {
    	CTaskQuery  *p = static_cast<CTaskQuery*>(param);
    	p->OnTaskWork();
    
    	return true;
    }
    
    bool CTaskQuery::OnTaskWork()
    {
    	while (true)
    	{
    		{
    			std::unique_lock<std::mutex> lock(m_mNotifyLock);
    			m_cvNotify.wait(lock);
    		}
    		
    		if (m_bIsQuit)
    		{
    			std::cout<<"exit thread id: " <<m_threadTask.get_id()<<std::endl;
    			break;
    		}
    
    		std::lock_guard<std::recursive_mutex> lcokTmp(m_dataLock);
    		while (!m_TaskQue.empty())
    		{
    			int nData = m_TaskQue.front();
    			if (m_pCallback != nullptr)
    			{
    				m_pCallback->OnCallBack(nData);
    			}
    			m_TaskQue.pop();
    		}
    
    	}
    	return true;
    }
    
    bool CTaskQuery::Start()
    {
    	HANDLE hId = m_threadTask.native_handle();
    	if (hId == nullptr)
    	{
    		m_threadTask = std::thread(doWork, this);
    	}
    
    	m_cvNotify.notify_one();
    
    	return true;
    }
    
    bool CTaskQuery::Stop()
    {
    	m_bIsQuit = true;
    	m_cvNotify.notify_one();
    	m_threadTask.join();
    
    	return true;
    }
    
    class CTest : public CInterFace
    {
    public:
    	CTest();
    	~CTest();
    
    	virtual void OnCallBack(int nData)
    	{
    		std::cout << "recv data: " << nData <<std::endl;
    	}
    
    private:
    	CTaskQuery m_TaskQ;
    };
    
    CTest::CTest()
    {
    	m_TaskQ.init(this);
    	for (int i= 0; i< 100; i++)
    	{
    		m_TaskQ.AddTask(i);
    	}
    	
    }
    
    CTest::~CTest()
    {
    
    }
    
    
    //测试
    
    int main()
    {
    	{
    		CTest testCallbk;
    
    		getchar();
    	}
    return 0;
    }