多线程任务池实现c++11

线程池

//
// Created by huanglixin on 2019/7/3.
//

#include <thread>
#include <vector>
#include <mutex>
#include <queue>
#include <condition_variable>
#include <iostream>
#include <atomic>
#include <functional>
#include <future>

namespace karl{
    class ThreadPool{
    public:
        explicit ThreadPool(int n):run(true)
        {
            if(n < 1)
                n = std::thread::hardware_concurrency();
            int i = 0;
            for(;i<n;i++)
            {
                threads.emplace_back(std::thread([this]{
                    std::cout<<"Create A New Thread\n";
                    while(run)
                    {
                        std::unique_lock<std::mutex> lock(mutex);
                        cond.wait(lock,[this]{return !tasks.empty() && run;});//等待为真
                        while(!tasks.empty())//每次获得锁就处理完任务队列
                        {
                            auto task = std::move(tasks.front());
                            task();//真正执行任务的地方
                            tasks.pop();
                        }
                    }
                }));
            }
        }
        template <typename Func,class... Args>
        std::future<typename std::result_of<Func(Args...)>::type> add(Func&& func,Args&&... args)
        {
            typedef typename std::result_of<Func(Args...)>::type return_type;
            if(run)
            {
                std::lock_guard<std::mutex> lock(mutex);
                auto t = std::make_shared<std::packaged_task<return_type()>>(std::bind(std::forward<Func>(func),std::forward<Args>(args)...));//通过绑定参数使得输入的函数和形参变成一个return_type()形式的函数
                auto ret = t->get_future();//获得future
                tasks.emplace([t]{//将函数插入队列
                    (*t)();//执行函数
                });
                cond.notify_one();
                return std::move(ret);
            }
        }
        ~ThreadPool()//析构需要处理的东西
        {
            stop();
        }
        void stop()
        {
            auto it = threads.begin();
            while(threads.end() != it)
            {
                if(it->joinable())
                    it->join();
            }
            run = false;
        }
    private:
        bool run;//运行标志
        std::queue<std::function<void()>> tasks;//任务队列
        std::mutex mutex;//互斥量
        std::condition_variable cond;//条件变量
        std::vector<std::thread> threads;//线程池
    };
};

使用

int main()
{
    karl::ThreadPool tp(3);
    std::vector<std::future<void>> v;
    for (int i = 0; i <= 5; ++i)
    {
        auto ans = tp.add([](const std::string& str1)
                          {
                              std::cout <<str1<< std::endl;
                              return;
                          }, "karl");
        v.push_back(std::move(ans));
    }

    auto it = v.begin();
    while(v.end() != it)
    {
        it->get();
        it++;
    }
}

一种可能的输出

Create A New Thread
Create A New Thread
karlCreate A New Thread

karl
karl
karl
karl
karl

参考

• https://github.com/lizhenghn123/zl_threadpool