C++ 簡單的任務隊列

任務隊列是指能夠實現任務在多線程間安全傳遞的先入先出的隊列。

任務是指組合了數據和操作的對象，這裡面定義為CTask類的對象。

任務的實現：

Task.cpp

 1 #include "stdafx.h"
 2 #include "Task.h"
 3 #include <iostream>
 4 using namespace std;
 5 
 6 CTask::CTask(int* nCount)
 7 {
 8     m_nCount = nCount;
 9 }
10 
11 CTask::~CTask()
12 {
13 }
14 
15 void CTask::DoWork()
16 {
17     (*m_nCount)++;
18 
19     cout << "Count = " << *m_nCount << endl;
20 }

Task.h

 1 #pragma once
 2 class CTask
 3 {
 4     int* m_nCount;
 5 public:
 6     CTask(int* nCount);
 7     ~CTask();
 8 
 9     void DoWork();
10 };

隊列的實現：

TaskQueue.cpp

 1 #include "stdafx.h"
 2 #include "TaskQueue.h"
 3 
 4 
 5 CTaskQueue::CTaskQueue()
 6 {
 7 }
 8 
 9 
10 CTaskQueue::~CTaskQueue()
11 {
12 }
13 
14 
15 //工作線程
16 void CTaskQueue::WorkThread()
17 {
18     while (m_bIsStart)
19     {
20         if (!m_taskQueue.empty())
21         {
22             CTask* it = m_taskQueue.front();
23             it->DoWork();
24             m_taskQueue.pop();
25             delete it;
26         }
27     }
28 }
29 
30 //向任務隊列添加任務
31 bool CTaskQueue::Push(CTask* task)
32 {
33     if (task == nullptr)
34     {
35         return false;
36     }
37 
38     m_mutex.lock();
39     m_taskQueue.push(task);
40     m_mutex.unlock();
41 
42     return true;
43 }
44 //從任務隊列獲取任務
45 CTask* CTaskQueue::Pop()
46 {
47     CTask* it = nullptr;
48 
49     m_mutex.lock();
50     if (!m_taskQueue.empty())
51     {
52         it = m_taskQueue.front();
53         m_taskQueue.pop();
54     }
55     m_mutex.unlock();
56     return it;
57 }
58 
59 bool CTaskQueue::Start()
60 {
61     if (m_bIsStart)
62     {
63         return false;
64     }
65     m_bIsStart = true;
66     m_thread = std::thread(&CTaskQueue::WorkThread, this);
67 
68     return true;
69 }
70 
71 void CTaskQueue::Stop()
72 {
73     m_bIsStart = false;
74     m_thread.join();
75 }

TaskQueue.h

 1 #pragma once
 2 #include "Task.h"
 3 #include <queue>
 4 #include <mutex>
 5 #include <thread>
 6 
 7 class CTaskQueue
 8 {
 9 public:
10     CTaskQueue();
11     ~CTaskQueue();
12 
13 private:
14     std::queue<CTask*> m_taskQueue;  //任務隊列
15     std::thread m_thread;
16     std::mutex m_mutex;
17     bool m_bIsStart;   //線程是否開啟
18 
19 public:
20     //工作線程
21     void WorkThread();  
22 
23     //向任務隊列添加任務
24     bool Push(CTask* task);
25     //從任務隊列獲取任務
26     CTask* Pop();
27 
28     //開啟線程
29     bool Start();
30     //終止線程
31     void Stop();
32 };

測試demo：

 1 // TaskQueue.cpp : 定義控制台應用程式的入口點。
 2 //
 3 
 4 #include "stdafx.h"
 5 #include "TaskQueue.h"
 6 #include "Task.h"
 7 
 8 void MyWorkTask1(CTaskQueue* pTaskQueue, int* nCount)
 9 {
10     for (size_t i = 0; i < 20; i++)
11     {    
12         CTask* task = new CTask(nCount);
13         pTaskQueue->Push(task);
14     }
15 }
16 
17 void MyWorkTask2(CTaskQueue* pTaskQueue, int* nCount)
18 {
19     for (size_t i = 0; i < 20; i++)
20     {
21         CTask* task = new CTask(nCount);
22         pTaskQueue->Push(task);
23     }
24 }
25 
26 int _tmain(int argc, _TCHAR* argv[])
27 {
28     CTaskQueue* pTaskQueue = new CTaskQueue();
29     pTaskQueue->Start();
30 
31     int* nCount = new  int(0);
32 
33     std::thread thread1(&MyWorkTask1, pTaskQueue, nCount);
34     std::thread thread2(&MyWorkTask2, pTaskQueue, nCount);
35 
36     //等待線程結束
37     if (thread1.joinable())
38     {
39         thread1.join();
40     }
41     if (thread2.joinable())
42     {
43         thread2.join();
44     }
45 
46     system("pause");
47     return 0;
48 }