一、引言本文目标从零实现一个轻量、可用、可扩展的 C 线程池。关键技术点std::threadstd::mutex / std::unique_lockstd::condition_variablestd::function / future / packaged_task / bind二、什么是线程池线程池是一种用于管理和复用线程的并发编程模型。它的核心思想是预先创建一组工作线程并将它们放入一个“池”中进行管理。当有新任务需要处理时不再创建和销毁线程而是直接将任务提交给线程池由池中空闲的线程来执行。三、为什么需要线程池频繁地创建和销毁线程会带来显著的系统开销包括内存分配、切换到内核态等。线程池通过复用线程有效解决了这个问题并带来了以下优势降低资源开销避免了频繁创建和销毁线程的开销提升了系统性能。提高响应速度任务到达时无需等待线程创建可以立即被分配给空闲线程执行。提高线程的可管理性线程是稀缺资源无限制地创建线程会消耗大量系统内存甚至导致内存溢出。线程池可以对线程数量进行统一分配、调优和监控。四、线程池的核心组成工作线程集合 (Worker Threads)池中预先创建好的一组线程它们会持续运行不断从任务队列中获取并执行任务。任务队列 (Task Queue)一个线程安全的队列用于存放所有待执行的任务。它作为任务提交者和工作线程之间的缓冲区。同步机制 (Synchronization)互斥锁 (Mutex)用于保护任务队列确保在多线程环境下对队列的访问是安全的防止竞态条件。条件变量 (Condition Variable)用于工作线程的等待和唤醒。当任务队列为空时工作线程会进入等待状态当有新任务加入时会通知唤醒一个或所有等待的线程。任务接口 (Task Interface)一个用于提交任务的方法允许外部将各种类型的任务函数、Lambda表达式等提交到线程池中。五、C线程池的实现因为在代码中使用到了一些异步编程技术所以先做个简单的介绍。std::condition_variable——条件变量它是一种线程间的同步机制当没有任务时它会阻塞工作线程。生产者线程将任务加入队列后会通过同一个条件变量唤醒在该条件变量下等待的线程。这么做的好处是避免了工作线程循环检测队列中有没有任务带来的CPU开销。代码中用到的接口主要有3个void wait (unique_lockmutex lck, Predicate pred);第一个参数是互斥锁该函数内部会将锁释放避免线程休眠时持有锁导致其他活跃线程拿不到锁。第二个参数是一个可调用对象这个可调用对象必须能够返回true或false。而且这个可调用对象会循环的执行直到它的返回结果是true。pred可调用对象返回true后线程被唤醒重新获取锁向下执行。void notify_one() noexcept;唤醒一个在该条件变量下等待的线程。void notify_all() noexcept;唤醒所有在该条件变量下等待的线程。std::future用来获取异步执行的结果。如果没有std::future在C中想要获取其他线程的返回值的话我们需要将这个返回值写入全局变量这样其他的线程才可以看到。全局变量是共享资源多线程场景下是需要加锁保护的std::future封装了这些底层的细节提供了一种同步获取结果的方式。get()方法会阻塞调用线程直到拿到结果才继续往下执行所以说它是一种同步获取结果的方式。std::function是一种函数包装器统一函数的类型。为什么这么说函数指针、lambda表达式虽然都是可调用对象但是它们的类型是完全不同的所以就不能够放到同一个容器中统一管理。但是经过function的包装后它们就有了统一的类型。下面在实现线程池中就可以看到如何使用。std::packaged_task是一种任务包装器用来获取异步执行任务的结果的。它内部有一个关联的future对象函数的返回值会被写入到这个future对象中同时它还提供了一个获取这个future对象的接口get_future()其他线程拿到这个关联的future后就可以调用get()方法获取异步执行的结果。顺便说一下packaged_task内部重载了可以直接通过packaged_task对象执行它包装的任务。std::bind绑定函数参数并返回一个可调用对象。比如函数Add(int a, int b)原本是需要传入两个参数的是以这种形式 Add(10, 20) 调用的。但是经过std::bind绑定后例如 auto func std::bind(Add, 10, 20); 在调用的时候就不用传参了直接这样 func() 调用效果是一样的。为什么要绑定参数在下面的线程池实现中就很明白了。123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115#include iostream#include vector#include queue#include thread#include future#include mutex#include functional#include chrono#include type_traits#include condition_variableclassThreadPool {public:ThreadPool(size_tthread_num 4):_thread_num(thread_num),_start(false),_stop(false){}~ThreadPool(){if(_start !_stop) stop();}ThreadPool(constThreadPool) delete;ThreadPool operator(constThreadPool) delete;ThreadPool(ThreadPool) delete;ThreadPool operator(ThreadPool) delete;voidstart() {std::unique_lockstd::mutex lock(_mutex);if(_start)return;_workers.reserve(_thread_num);for(size_ti 0; i _thread_num; i) {_workers.emplace_back(std::thread([this](){work_loop();}));}_start true;}voidstop() {{std::unique_lockstd::mutex lock(_mutex);if(!_start || _stop)return;// 在join回收线程之前必须先将_stop置为true// 否则工作线程可能会一直阻塞在条件变量上导致无法正常退出甚至会导致程序崩溃_stop true;}_cond.notify_all();for(auto worker : _workers) {if(worker.joinable()) worker.join();}}templateclassF,class... Argsauto submit(F f, Args... args)-std::futurestd::invoke_result_tF, Args... {usingreturn_type std::invoke_result_tF, Args...;// 绑定函数参数并交给任务包装器auto task std::make_sharedstd::packaged_taskreturn_type()(std::bind(std::forwardF(f), std::forwardArgs(args)...));// 获取关联的futurestd::futurereturn_type res task-get_future();// 加锁入队列{std::unique_lockstd::mutex lock(_mutex);if(_stop || !_start)throwstd::runtime_error(线程池未启动);_tasks.emplace([task](){(*task)();});}_cond.notify_one();returnres;}private:voidwork_loop() {while(true) {std::functionvoid() task;{std::unique_lockstd::mutex lock(_mutex);_cond.wait(lock, [this](){return_stop || !_tasks.empty();});if(_stop _tasks.empty())return;task std::move(_tasks.front());_tasks.pop();}task();}}private:std::vectorstd::thread _workers;std::queuestd::functionvoid() _tasks;std::mutex _mutex;std::condition_variable _cond;size_t_thread_num;bool_start;bool_stop;};intadd(inta,intb) {returna b;}voidprint() {std::cout -------------------print------------------- std::endl;std::cout Hello World! std::endl;}intmain() {ThreadPool pool(4);pool.start();std::cout ThreadPoolTest std::endl;pool.submit([](){std::cout -------------------lambda------------------- std::endl;std::cout this is a lambda! std::endl;});std::this_thread::sleep_for(std::chrono::seconds(3));auto ret1 pool.submit(add, 10, 20);std::cout -------------------add------------------- std::endl;std::cout 10 20 ret1.get() std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));pool.submit(print);pool.stop();return0;}运行结果在上述代码实现中用到了C17的语法。这里我解释一下队列中的任务参数为空并且返回值为void但是带返回值的add函数为什么可以插入队列中。通过std::bind绑定函数的所有参数所有就做到了“无参”。通过std::packaged_task获取函数的返回值所以不担心返回值拿不到。通过lambda封装一层不管原本是否有参数是否有返回值加入队列中的任务都是满足无参和无返回值的。这三个操作组合在一起不管函数有无返回值都能适配到任务队列中。另外在绑定参数时不仅对函数参数进行了完美转发还对函数本身进行了完美转发。对参数进行完美转发是因为要保持它本身的左右值属性。对函数进行完美转发是因为我们在使用时可能会直接在submit函数传入lambda这时候lambda它是一个右值采用完美转发可以保持它的右值属性触发移动语义也就避免了std::bind内部对它进行拷贝。