学习内容来自以下地址
http://www.cnblogs.com/qicosmos/p/4772486.html
github https://github.com/qicosmos/cosmos
有了任务队列 在程序运行之初 就预开启多个执行任务的线程 等待任务队列中传来的元素 对元素进行处理 执行完毕后 也不销毁线程 而是等待下个元素进行处理
使用std::list<std::shared_ptr<std::thread>> m_threadgroup; 变量记录线程指针 方便管理
测试例子中传递的元素类型定义为 td::function<void()> 即一个返回类型为void 无输入参数的函数
对应代码为using Task = std::function<void()>;
每当队列中添加该元素TASK 则有线程从队列中取出执行
代码如下
#pragma once #include <list>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <iostream> using namespace std; template<typename T>
class SyncQueue {
public:
SyncQueue(int maxSize):m_maxSize(maxSize),m_needStop(false){}
void Put(const T& x) {
Add(x);
}
void Put(T&& x) {
Add(std::forward<T>(x));
}
void Take(std::list<T>& list) {
std::unique_lock<std::mutex> locker(m_mutex);
m_notEmpty.wait(locker, [this] {return m_needStop || NotEmpty(); });
if (m_needStop)
return;
list = std::move(m_queue);
m_notFull.notify_one();
} void Take(T& t) {
std::unique_lock<std::mutex> locker(m_mutex);
m_notEmpty.wait(locker, [this] {return m_needStop || NotEmpty(); });
if (m_needStop)
return;
t = m_queue.front();
m_queue.pop_front();
m_notFull.notify_one();
} void Stop() {
{
std::lock_guard<std::mutex> locker(m_mutex);
m_needStop = true;
}
m_notFull.notify_all();
m_notEmpty.notify_all();
} bool Empty() {
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.empty();
} bool Full() {
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.size() == m_maxSize;
} size_t Size() {
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.size();
} int Count() {
return m_queue.size();
}
private:
bool NotFull()const {
bool full = m_queue.size() >= m_maxSize;
if (full)
cout << "buffer area is full,wait..." << endl;
return !full;
}
bool NotEmpty()const {
bool empty = m_queue.empty();
if (empty)
cout << "buffer area is empty,wait... " <<
" threadID: " << this_thread::get_id() <<endl;
return !empty;
} template<typename F>
void Add(F&& x) {
std::unique_lock<std::mutex> locker(m_mutex);
m_notFull.wait(locker, [this] {return m_needStop || NotFull(); });
if (m_needStop)
return;
m_queue.push_back(std::forward<F>(x));
m_notEmpty.notify_one();
}
private:
std::list<T> m_queue;
std::mutex m_mutex;
std::condition_variable m_notEmpty;
std::condition_variable m_notFull;
int m_maxSize;
bool m_needStop;
};
SyncQueue.h
// ThreadPool.cpp: 定义控制台应用程序的入口点。
// #include "stdafx.h"
#include <list>
#include <thread>
#include <functional>
#include <memory>
#include <atomic>
#include "SyncQueue.h" const int MaxTaskCount = ;
class ThreadPool {
using Task = std::function<void()>;
public:
ThreadPool(int numThreads = std::thread::hardware_concurrency()) :m_queue(MaxTaskCount) {
Start(numThreads);
}
~ThreadPool(void) {
Stop();
}
void Stop() {
std::call_once(m_flag, [this] {StopThreadGroup(); });
}
void AddTask(Task&& task) {
m_queue.Put(std::forward<Task>(task));
}
void AddTask(const Task& task) {
m_queue.Put(task);
} void Start(int numThreads) {
m_running = true;
for (int i = ; i < numThreads; i++) {
m_threadgroup.push_back(std::make_shared<std::thread>(&ThreadPool::RunInThread, this));
}
}
private:
void RunInThread() {
while(m_running) {
std::list<Task> list;
m_queue.Take(list); for (auto& task : list) {
if (!m_running)
return;
task();
}
}
} void StopThreadGroup() {
m_queue.Stop();
m_running = false;
for (auto thread : m_threadgroup) {
if (thread)
thread->join();
}
m_threadgroup.clear();
}
std::list<std::shared_ptr<std::thread>> m_threadgroup;
SyncQueue<Task> m_queue;
atomic_bool m_running;
std::once_flag m_flag;
}; //=========================================================
void TestThdPool()
{
ThreadPool pool;
pool.Start(); std::thread thd1([&pool] {
for (int i = ; i < ; i++) {
auto thdId = this_thread::get_id();
pool.AddTask([thdId] {
std::cout << "thread1 id : " << thdId << std::endl;
});
}
}); std::thread thd2([&pool] {
for (int i = ; i < ; i++) {
auto thdId = this_thread::get_id();
pool.AddTask([thdId] {
std::cout << "thread2 id : " << thdId << std::endl;
});
}
}); this_thread::sleep_for(std::chrono::seconds());
getchar();
pool.Stop();
thd1.join();
thd2.join();
} int main()
{
TestThdPool();
return ;
}
ThreadPool.cpp
运行情况如下：
buffer area is empty,wait... threadID: 10244
buffer area is empty,wait... threadID: 8168
buffer area is empty,wait... threadID: 10136
buffer area is empty,wait... threadID: 5812
buffer area is empty,wait... threadID: 4064
buffer area is empty,wait... threadID: 7872
thread1 id : 9712buffer area is empty,wait... threadID:
8168
thread1 id : 9712
buffer area is empty,wait... threadID: 10244thread1 id : 9712
thread2 id : buffer area is empty,wait... threadID: thread2 id : 65206520
5812
thread2 id : thread1 id : thread2 id : 9712buffer area is empty,wait... threadID: thread2 id : 65206520
6520
thread2 id :
6520
thread2 id :
6520thread1 id :
40649712
thread2 id :
6520buffer area is empty,wait... threadID: thread2 id : 6520
thread1 id : 97125812
thread1 id : 9712buffer area is empty,wait... threadID: thread1 id : 10136
9712
buffer area is empty,wait... threadID: 7872thread1 id : 9712
thread2 id : 6520buffer area is empty,wait... threadID:
10244
thread1 id : 9712
buffer area is empty,wait... threadID: 8168