谁能简单说明下c++线程池怎么用的

Posted 2023-02-15

参考技术A

先上实现！

实现：

#ifndef ILOVERS_THREAD_POOL_H
#define ILOVERS_THREAD_POOL_H
 
#include <iostream>
#include <functional>
#include <thread>
#include <condition_variable>
#include <future>
#include <atomic>
#include <vector>
#include <queue>
 
// 命名空间
namespace ilovers 
    class TaskExecutor;

 
class ilovers::TaskExecutor
    using Task = std::function<void()>;
private:
    // 线程池
    std::vector<std::thread> pool;
    // 任务队列
    std::queue<Task> tasks;
    // 同步
    std::mutex m_task;
    std::condition_variable cv_task;
    // 是否关闭提交
    std::atomic<bool> stop;
    
public:
    // 构造
    TaskExecutor(size_t size = 4): stop false
        size = size < 1 ? 1 : size;
        for(size_t i = 0; i< size; ++i)
            pool.emplace_back(&TaskExecutor::schedual, this);    // push_back(std::thread...)
        
    
    
    // 析构
    ~TaskExecutor()
        for(std::thread& thread : pool)
            thread.detach();    // 让线程“自生自灭”
            //thread.join();        // 等待任务结束， 前提：线程一定会执行完
        
    
    
    // 停止任务提交
    void shutdown()
        this->stop.store(true);
    
    
    // 重启任务提交
    void restart()
        this->stop.store(false);
    
    
    // 提交一个任务
    template<class F, class... Args>
    auto commit(F&& f, Args&&... args) ->std::future<decltype(f(args...))> 
        if(stop.load())    // stop == true ??
            throw std::runtime_error("task executor have closed commit.");
        
        
        using ResType =  decltype(f(args...));    // typename std::result_of<F(Args...)>::type, 函数 f 的返回值类型
        auto task = std::make_shared<std::packaged_task<ResType()>>(
                        std::bind(std::forward<F>(f), std::forward<Args>(args)...)
                );    // wtf !
            // 添加任务到队列
            std::lock_guard<std::mutex> lock m_task;
            tasks.emplace([task]()   // push(Task...)
                (*task)();
            );
        
        cv_task.notify_all();    // 唤醒线程执行
        
        std::future<ResType> future = task->get_future();
        return future;
    
    
private:
    // 获取一个待执行的 task
    Task get_one_task()
        std::unique_lock<std::mutex> lock m_task;
        cv_task.wait(lock, [this]() return !tasks.empty(); );    // wait 直到有 task
        Task task std::move(tasks.front());    // 取一个 task
        tasks.pop();
        return task;
    
    
    // 任务调度
    void schedual()
        while(true)
            if(Task task = get_one_task())
                task();    //
            else
                // return;    // done
            
        
    
;
 
#endif
 
void f()

    std::cout << "hello, f !" << std::endl;

 
struct G
    int operator()()
        std::cout << "hello, g !" << std::endl;
        return 42;
    
;
 
 
int main()
try
    ilovers::TaskExecutor executor 10;
    
    std::future<void> ff = executor.commit(f);
    std::future<int> fg = executor.commit(G);
    std::future<std::string> fh = executor.commit([]()->std::string  std::cout << "hello, h !" << std::endl; return "hello,fh !";);
    
    executor.shutdown();
    
    ff.get();
    std::cout << fg.get() << " " << fh.get() << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(5));
    executor.restart();    // 重启任务
    executor.commit(f).get();    //
    
    std::cout << "end..." << std::endl;
    return 0;
catch(std::exception& e)
    std::cout << "some unhappy happened... " << e.what() << std::endl;

实现原理：

线程池一般要复用线程，所以如果是取一个 task 分配给某一个 thread，执行完之后再重新分配，在语言层面基本都是不支持的：一般语言的 thread 都是执行一个固定的 task 函数，执行完毕线程也就结束了(至少 c++ 是这样)。要如何实现 task 和 thread 的分配呢？很简单，让每一个 thread 都去执行调度函数：循环获取一个 task，然后执行。这样保证了 thread 函数的唯一性，而且复用线程执行 task 。

代码的详细解释：

一个线程 pool，一个任务队列 queue ，应该没有意见。

任务队列是典型的生产者-消费者模型，本模型至少需要两个工具：一个 mutex + 一个条件变量，或是一个 mutex + 一个信号量。mutex 实际上就是锁，保证任务的添加和移除(获取)的互斥性，一个条件变量是保证获取 task 的同步性：一个 empty 的队列，线程应该等待(阻塞)。

stop 控制任务提交，是受了 Java 的影响，还有实现类不叫 ThreadPool 而是叫 TaskExecutor。

atomic<bool> 本身是原子类型，从名字上就懂：它们的操作 load()/store() 是原子操作，所以不需要再加 mutex。

求采纳，谢谢！

本回答被提问者采纳