基於C++11實現的線程池

1.C++11中引入了lambada表達式，很好的支持異步編程

2.C++11中引入了std::thread，可以很方便的構建線程，更方便的可移植特性

3.C++11中引入了std::mutex，可以很方便的構建線程鎖互斥訪問，更方便的可移植特性

4.C++11中引入了std::condition_variable，可以不依賴於win32 api實現自己的消費者生產者模型

5.利用改進版本的shared_ptr，可以很好的解決多線程生命周期的棘手問題

/************************************************************************/
/*                                                                      */
/************************************************************************/

#ifndef __CARBON_THREAD_POOL_H
#define __CARBON_THREAD_POOL_H

#include <vector>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>
#include <string>
#include <sstream>
#include <deque>

namespace CARBON {

    //************************************
    // Method:    Create
    // Returns:   std::shared_ptr
    // Qualifier: 用於創建智能指針實例
    // Parameter: args, 可變參數，接受任意個數的參數，傳遞給T的構造函數
    //************************************
    template<typename T, typename... ARG>
    std::shared_ptr<T> Create(ARG&&... args)
    {
        struct TEnableShared : public T
        {
            TEnableShared(ARG&&... args)
                : T(std::forward<ARG>(args)...)
            {}
        };

        return std::make_shared<TEnableShared>(std::forward<ARG>(args)...);
    }

    class ThreadPool : public std::enable_shared_from_this<ThreadPool>
    {
    protected:
        ThreadPool()
            : _stop(false)
        {}

        virtual ~ThreadPool()
        {
            {
                std::unique_lock<std::mutex> lock(_lock);
                _stop = true;
            }
            _condition.notify_all();
            for (std::thread &worker : _workers)
                worker.join();
        }

    public:
        // initialize thread pool with number of threads
        bool InitializePool(size_t threads)
        {
            if (!_workers.empty()) return true;

            for (size_t i = 0; i < threads; ++i)
            {
                std::weak_ptr<ThreadPool> _wtp = this->shared_from_this();
                auto th = [](std::weak_ptr<ThreadPool> wtp) {
                    for (;;)
                    {
                        std::function<void()> task;

                        {
                            std::shared_ptr<ThreadPool> stp = wtp.lock();
                            if (!stp)
                                return;

                            std::unique_lock<std::mutex> lock(stp->_lock);
                            auto shipment = [&] ()->bool { return stp->_stop || !stp->_tasks.empty(); };
                            stp->_condition.wait(lock, shipment);
                            if (stp->_stop)
                                return;
                            if (stp->_tasks.empty())
                                continue;
                            task = std::move(stp->_tasks.front()).task;
                            stp->_tasks.pop_front();
                        }

                        task();
                    }
                };
                _workers.emplace_back(th, _wtp);
            }

            return !_workers.empty();
        }

        //************************************
        // Method:    EnqueueTask
        // Returns:   std::future, 值類型由functor f指定
        // Qualifier: 可以借由返回的std::future獲取結果，但是更建議在functor中做異步通知
        // Parameter: taskid 用於接受任務的id描述
        // Parameter: functor f, 函數對象，用於執行任務
        // Parameter: args, 可變參數，接受任意個數的參數，傳遞給functor f
        //************************************
        template<class F, class... Args>
        auto EnqueueTask(std::string& taskid, F&& f, Args&&... args)
            ->std::future<typename std::result_of<F(Args...)>::type>
        {
            if (_workers.empty())
                throw std::runtime_error("ThreadPool not initialized yet");

            using return_type = typename std::result_of<F(Args...)>::type;

            auto task = std::make_shared<std::packaged_task<return_type()>>(
                std::bind(std::forward<F>(f), std::forward<Args>(args)...)
                );

            std::future<return_type> res = task->get_future();
            {
                std::unique_lock<std::mutex> lock(_lock);

                // don't allow enqueueing after stopping the pool
                if (_stop)
                    throw std::runtime_error("enqueue on stopped ThreadPool");

                stThreadTask st;
                std::stringstream ss;
                ss << (void*)task.get();
                ss >> taskid;
                st.taskid = taskid;
                st.task = [task]() { (*task)(); };
                _tasks.push_back(st);
            }
            _condition.notify_one();
            return res;
        }

        //************************************
        // Method:    GetTasksSize
        // Returns:   size_t
        // Qualifier: 獲取等待任務隊列的任務數，正在執行的任務已經彈出隊列，所以不參與計算
        //************************************
        size_t GetTasksSize()
        {
            std::unique_lock<std::mutex> lock(_lock);
            return _tasks.size();
        }

        //************************************
        // Method:    RemoveTask
        // Returns:   bool, 找到任務並移除則返回true，否則返回false
        // Qualifier: 正在執行的任務已經彈出任務隊列，應該在其它地方通知任務退出
        // Qualifier: 執行完成的任務已經彈出任務隊列，無法移除不存在的任務
        // Qualifier: 該接口只能移除處在等待中的任務
        // Parameter: taskid是任務的唯一標示，由EnqueueTask返回
        //************************************
        bool RemoveTask(const std::string& taskid)
        {
            std::unique_lock<std::mutex> lock(_lock);
            for (auto& t = _tasks.begin(); t != _tasks.end(); ++t)
            {
                if (taskid == t->taskid)
                {
                    _tasks.erase(t);
                    return true;
                }
            }

            return false;
        }

    private:
        typedef struct stThreadTask
        {
            std::function<void()> task;
            std::string taskid;
        }stThreadTask;

        // need to keep track of threads so we can join them
        std::vector< std::thread > _workers;
        // the task queue
        std::deque< stThreadTask > _tasks;

        // synchronization
        std::mutex _lock;
        std::condition_variable _condition;
        bool _stop;
    };
}

#endif

使用enable_shared_from_this來確保內部線程訪問指針時，不會因為指針失效造成的非法訪問

weak_ptr很好的保證了ThreadPool的生命周期安全性和實效性

由於使用了share_from_this，將初始化代碼整體拿出來放到InitializePool中實現

ThreadPool的構造函數和析構函數聲明為protected，用於保證外部不要直接生成ThreadPool實例

應該使用Create函數來生成ThreadPool實例

測試代碼如下：

namespace {
    std::condition_variable _exit_cv;
}

void func(int n)
{
    std::cout << "func with n " << n << std::endl;
}

using CARBON::ThreadPool;

std::string taskid;
std::shared_ptr<ThreadPool> stp = CARBON::Create<ThreadPool>();
std::weak_ptr<ThreadPool> _wtp = stp;
stp->InitializePool(2);


stp->EnqueueTask(taskid, [](std::function<void(int)> cbf, std::weak_ptr<ThreadPool> wtp) ->int {
        std::cout << "task1\n";

        for (int i = 0; i < 5; ++i) {
            std::mutex mtx;
            std::unique_lock<std::mutex> lck(mtx);
            if(_exit_cv.wait_for(lck, std::chrono::milliseconds(400)) == std::cv_status::no_timeout)
                break;

            if (cbf) cbf(i);
            if (wtp.expired())
                break;
        }

        return 5;
    }, func, _wtp);

當需要中斷線程執行時，應該在外部通知線程中的任務自行退出

例子中可以在主線程中這么做

    _exit_cv.notify_all();

_exit_cv用於模擬sleep操作
func用於模擬任務結果的異步通知，這里為了省事使用了函數指針，實際工作中應該使用functor來傳遞，以保證生命周期的有效性
比如std::bind和shared_ptr一起構造的functor對象