C++11线程池

左手的ㄟ右手 2022-05-28 06:52 167阅读 0赞

纯C++11线程池，简单功能，无优先级设置等高级功能  
参考代码：

https://github.com/progschj/ThreadPool

--------------------

// ThreadPool.h

#pragma once
    #ifndef THREAD_POOL_H
    #define THREAD_POOL_H
    
    #include <vector>
    #include <queue>
    #include <memory>
    #include <thread>
    #include <mutex>
    #include <condition_variable>
    #include <future>
    #include <functional>
    #include <stdexcept>
    
    class ThreadPool
    {
        using Task = std::function<void()>;
    public:
        // add task func
        template<class Func, class... Args>
        auto addTask(Func&& func, Args&&... args)-> std::future<typename std::result_of<Func(Args...)>::type>
        {
            using funcReturnType = typename std::result_of<Func(Args...)>::type;
    
            auto task = std::make_shared< std::packaged_task<funcReturnType()> >(std::bind(std::forward<Func>(func), std::forward<Args>(args)...));
    
            std::future<funcReturnType> res = task->get_future();
            {
                std::unique_lock<std::mutex> lock(m_mutex);
                if (m_isStop.load())
                {
                    throw std::runtime_error("addTask on stopped ThreadPool");
                }
    
                m_taskQueue.emplace([task](){ (*task)(); });
            }
            m_condition.notify_one();
            return res;
        }
    
    public:
        explicit ThreadPool();
        ~ThreadPool();
    
        // start threadpool
        void start(int numThreads);
        // tasks size
        size_t queueSize();
    
    private:
        std::atomic<bool> m_isStop;
        std::mutex m_mutex;
        std::condition_variable m_condition;
        std::vector<std::thread> m_workThreads;
        std::queue<Task> m_taskQueue;
    };
    #endif  //THREAD_POOL_H

--------------------

// ThreadPool.cpp
    #include "ThreadPool.h"
    
    ThreadPool::ThreadPool()
        :m_isStop(false)
    {
    }
    
    ThreadPool::~ThreadPool()
    {
        m_isStop.store(true);
        m_condition.notify_all();
    
        for (std::thread& workThread : m_workThreads)
        {
            workThread.detach();
        }
    }
    
    /** 
     * @fn     ThreadPool::start
     * @access public 
     * @brief  start threadpool
     * @param  int numThreads : thread number
     * @return void
     */
    void ThreadPool::start(int numThreads)
    {
        m_workThreads.reserve(numThreads);
        for (size_t i = 0; i < numThreads; ++i)
        {
            m_workThreads.emplace_back(
                [this]
            {
                //std::thread::id thread_id = std::this_thread::get_id(); // get thread id, no use here.
                while (!m_isStop)
                {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock{ m_mutex };
                        m_condition.wait(lock, [this]{ return m_isStop.load() || !m_taskQueue.empty(); });
    
                        if (m_isStop.load() && m_taskQueue.empty())
                        {
                            return;
                        }
    
                        task = std::move(m_taskQueue.front());
                        m_taskQueue.pop();
                    }
                    task();
                }
            }
            );
        }
    }
    
    /** 
     * @fn     ThreadPool::queueSize
     * @access public 
     * @brief  get current tasks size
     * @return size_t
     */
    size_t ThreadPool::queueSize()
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        return m_taskQueue.size();
    }

--------------------

使用方法参考：

ThreadPool pool;
    pool.start(4);
    pool.addTask([=]()
    { 
        //do sth. here. 
    });