异构线程池的c++实现方案

概要

通常线程池是同质的，每个线程都可以执行任意的task（每个线程中的task顺序执行），如下图所示：

 但本文所介绍的线程和task之间有绑定关系，如A task只能跑在A thread上（因此称为异构线程池，每个线程的功能是有所区别的），如下图所示：

接口设计

TThreadPool接口设计

// 线程池
class TThreadPool
{
public:TThreadPool() {}TThreadPool(const TThreadPool&) = delete;TThreadPool& operator=(const TThreadPool& other) = delete;~TThreadPool();bool add_thread(std::string name); // add one thread into poolbool delete_thread(std::string name); // remove one thread from poolTThread* get_thread_by_name(std::string threadname); // get the thread by name, don't delete return object by yourselfbool append_task_by_thread(const std::string threadname, const std::function<void()>& task); // add task to pointed threadprivate:std::mutex m_mutex;std::map<std::string, TThread*> m_threads;
};

 TThreadPool类的主要功能是管理创建的线程（TThread，它是线程的具体实现），它提供了增加/删除线程的接口，同时给每个线程打上了标签（name）。

TThread接口设计

// 对std::thread的封装类
class TThread
{
public:TThread(std::string name);TThread(const TThread& other) = delete;TThread& operator=(const TThread& other) = delete;~TThread();bool push_task(const std::function<void()>& task); // one add taskstd::thread::id get_thread_id(); // for log purposevoid set_max_task_size(int s); // avoid thread too busyint get_task_size(); // get current task numberprivate:void work(); // real work threadvoid notify(); // notify work thread to quit
private:std::string s_name;std::atomic_bool b_running;std::thread* p_thread;std::mutex m_mutex;std::queue<std::function<void()> > m_tasks;std::condition_variable m_cond;std::atomic<int> i_maxTaskSize;
};

TThread类的主要功能是分配任务（push_task函数）和处理任务（work函数）。

代码实现

TThreadPool类

TThreadPool::~TThreadPool() {std::unique_lock<std::mutex> lk(m_mutex);for(auto iter=m_threads.begin(); iter!=m_threads.end(); iter++) {if(iter->second != nullptr) {delete iter->second;}}m_threads.clear();
}bool TThreadPool::add_thread(std::string name) {std::unique_lock<std::mutex> lk(m_mutex);if(m_threads.count(name)) {return false;}auto tt = new TThread(name);if(tt == nullptr) {return false;}m_threads[name] = tt;return true;
}bool TThreadPool::delete_thread(std::string name) {std::unique_lock<std::mutex> lk(m_mutex);if(m_threads.count(name) == 0) {return false;}delete m_threads[name];m_threads.erase(name);return true;
}TThread* TThreadPool::get_thread_by_name(std::string threadname) {std::unique_lock<std::mutex> lk(m_mutex);if(m_threads.count(threadname) == 0) {return nullptr;}return m_threads[threadname];
}bool TThreadPool::append_task_by_thread(const std::string threadname, const std::function<void()>& task)
{std::unique_lock<std::mutex> lk(m_mutex);if(m_threads.count(threadname) == 0) {return false;}auto tt = m_threads[threadname];return tt->push_task(task);
}

TThread类

const int MAX_TASK_SIZE = 100; // max task size in one threadTThread::TThread(std::string name) : s_name(name), b_running(true), i_maxTaskSize(MAX_TASK_SIZE) {p_thread = new std::thread(&TThread::work, this);
}TThread::~TThread() {notify(); // notify work thread quitif(p_thread->joinable()) {p_thread->join();}delete p_thread;
}bool TThread::push_task(const std::function<void()>& task) {std::unique_lock<std::mutex> lk(m_mutex);if(!b_running) {return false;}if(m_tasks.size() > i_maxTaskSize.load()) {return false;}m_tasks.push(task);m_cond.notify_one();return true;
}std::thread::id TThread::get_thread_id() {return p_thread->get_id();
}void TThread::set_max_task_size(int s) {if(s <= 0) {return;}i_maxTaskSize.store(s);
}void TThread::work() {std::cout << std::this_thread::get_id() << " begin work thread" << std::endl;while (b_running) { // quit even tasks remainingstd::function<void()> task;{std::unique_lock<std::mutex> lk(m_mutex);if (!m_tasks.empty()) {task = m_tasks.front();m_tasks.pop();} else if (b_running && m_tasks.empty()) {m_cond.wait(lk);}}if (task)task(); // do the task}std::cout << std::this_thread::get_id() << " end work thread"  << std::endl;
}void TThread::notify() {std::unique_lock<std::mutex> lk(m_mutex);b_running = false;m_cond.notify_one(); // mutex will be released here, therefore another thread would lock it afterward
}int TThread::get_task_size() {std::unique_lock<std::mutex> lk(m_mutex);return m_tasks.size();
}

使用方式

有两种方式可以调用对应的线程

公共代码

void func1(int i) {std::cout << "into func1: " << i << std::endl;sleep(2); // simulate real work
}TThreadPool thread_pool;thread_pool.add_thread("vdr"); // 启动vdr线程thread_pool.add_thread("xgb"); // 启动xgb线程

方式一、(先获取线程对象，然后对该线程对象添加任务)

auto tt = thread_pool.getThreadByName("vdr");
tt->push_task(std::bind(func1, 2));
tt->push_task(std::bind(func1, 5));

方式二、（直接通过线程池给对应线程添加任务）

thread_pool.append_task_by_thread("vdr", std::bind(func1, 2));
thread_pool.append_task_by_thread("vdr", std::bind(func1, 5));

注：

task是std::function<void()>类型，上面的demo是普通函数实现的，真实场景应该是类函数，实现如下：

class A {
public:void func(std::string str) {std::cout << "into A func: " << str << std::endl;}
};A a;thread_pool.append_task_by_thread("vdr", std::bind(&A::func, &a, "2"));thread_pool.append_task_by_thread("vdr", std::bind(&A::func, &a, "5"));