Folly Executors 与线程池深度指南CPUThreadPoolExecutor 与 IOThreadPoolExecutor 的原理、选型与实践【免费下载链接】follyAn open-source C library developed and used at Facebook.项目地址: https://gitcode.com/GitHub_Trending/fol/follyFolly 在 folly/docs/Executors.md 中系统讲解了其线程池Thread Pool与 Executor 体系的设计动机、实现原理和使用方式。本文以该文档为骨架结合 folly/executors 目录下的源码实现深入剖析 Folly 为何要自研两套线程池CPU 与 IO 分离、它们各自的数据结构与调度细节以及如何借助全局 Executor、Observers 与 PoolStats 在生产环境中高效运行并发代码。读完本文你将掌握CPUThreadPoolExecutor、IOThreadPoolExecutor、ThreadPoolExecutor基类的核心机制并能在自己的代码中正确选型与配置。快速上手从全局 Executor 开始Folly 提供两个具体的线程池实现IOThreadPoolExecutor与CPUThreadPoolExecutor并将它们作为完整异步框架folly/futures的一部分内置。大多数场景下你并不需要手动构造线程池而是直接获取进程级的全局 Executor再与 Future 组合使用。最简单、最常见的用法是将 Future 延续continuation调度到 CPU 线程池上执行auto f someFutureFunction().via(getCPUExecutor()).then(...);如果需要在 Thrift / memcache 客户端上发起调用则需要先拿到一个事件循环EventBase此时从 IO 线程池获取getEventBase()再通过via(getCPUExecutor())回到 CPU 线程池处理结果auto f getClient(getIOExecutor()-getEventBase())-callSomeFunction(args...) .via(getCPUExecutor()) .then([](Result r) { /* do something with result */ });getCPUExecutor()返回全局 CPU 线程池folly/executors/GlobalExecutor.hgetIOExecutor()返回全局 IO 线程池可用-getEventBase()取出一个按 round-robin 方式选中的EventBase直接在上面调度 IO 工作。全局 Executor 的线程数与新旧 API全局 Executor 的线程数量可以通过 gflags 显式配置folly_global_cpu_executor_threads全局 CPU 线程池创建的线程数folly_global_io_executor_threads全局 IO 线程池创建的线程数。两者默认值均为 0。在 folly/executors/GlobalExecutor.cpp 中可以看到当 flag 为 0 时线程数会回退为folly::available_concurrency()即机器可用的核数size_t nthreads FLAGS_folly_global_cpu_executor_threads; nthreads nthreads ? nthreads : folly::available_concurrency(); return new std::shared_ptrImmutableGlobalCPUExecutor( new ImmutableGlobalCPUExecutor( nthreads, std::make_sharedNamedThreadFactory(GlobalCPUThreadPool)));需要说明的是文档中示例使用的getCPUExecutor()/getIOExecutor()在当前版本中已被标记为 deprecated。从 folly/executors/GlobalExecutor.h 的注释可以看出官方推荐使用getGlobalCPUExecutor()/getGlobalIOExecutor()获取不可变的全局 Executor返回KeepAlive可安全配合 Future / coroutine 使用保证前向进度若确有替换全局 Executor 的需求使用getUnsafeMutableGlobalCPUExecutor()/setUnsafeMutableGlobalCPUExecutor()等 UnsafeMutable 系列接口getGlobalCPUExecutorWeakRef()返回弱 KeepAlive不阻止全局 Executor 在关闭时析构适合尽力而为的后台任务但不应用于 Future / coroutine 延续因为延续依赖前向进度保证弱引用可能导致死锁。为什么不用 C11 的 std::launchC11 的std::launch只有两种模式async与deferred在生产系统中两者都不是理想选择async每次 launch 都会无限制地新起一个线程线程数量完全不受控deferred任务被延迟到真正需要结果时才执行且届时在当前线程里同步地运行阻塞调用方。Folly 的线程池则不同任务总是尽可能早地被调度执行同时对最大任务数 / 线程数设有上限因此永远不会使用超出需要的线程数。这种按需生长、有界并发的模型正是生产级服务所需要的。为什么需要自研线程池文档给出的理由是当时C11 时代现成的线程池实现都不完整——基于 pipe 的线程池太慢而多个较早的实现不支持std::function无法与 Future/回调风格的代码无缝协作。因此 Folly 需要自己提供一套面向高并发、低延迟、且与std::function和异步框架深度集成的线程池实现。为什么需要两种不同类型的线程池这是理解 Folly 线程池架构最关键的问题核心在于IO 事件循环与公平队列在操作系统原语层面互相排斥epoll 需要 fdevent_fd是最新的通知机制但它有个副作用——一个活跃的 fd 会触发所有正在等待它的 epoll 循环惊群效应thundering herd。因此如果你想要一个公平队列一个总队列 vs. 每工作线程一个队列就需要借助信号量semaphore来实现公平唤醒信号量进不了 epoll 循环semaphore 无法被放入 epoll 等待集合中所以基于信号量的公平队列与 IO 事件循环不兼容IO 与 CPU 本就该分离即便技术上可行通常也应当把 IO 密集与 CPU 密集的工作分开以便在 IO 路径上获得更强的尾延迟tail latency保证。正是基于上述原因Folly 提供了两套定位不同的线程池IOThreadPoolExecutor面向事件循环与 IO和CPUThreadPoolExecutor面向计算密集任务。IOThreadPoolExecutorevent_fd 每线程 NotificationQueueIOThreadPoolExecutor是一个面向 IO 密集型任务的线程池folly/executors/IOThreadPoolExecutor.h 的类注释完整概括了它的设计要点使用event_fd进行通知并唤醒 epoll 循环每个线程每个 epoll对应一个队列具体是NotificationQueueIOThread结构中持有自己的EventBase*任务通过ioThread-eventBase-runInEventBaseThread(...)投递见 folly/executors/IOThreadPoolExecutor.cpp无谓系统调用被消除如果目标线程已经在运行、并没有阻塞在 epoll 上等待那么只需要把新任务放进它的队列即可无需额外的 syscall 去唤醒事件循环空闲线程回收内存如果某个线程等待超过数秒它的栈会被madvise掉。实现上由MemoryIdlerTimeout完成——它是AsyncTimeoutEventBase::LoopCallback的合体事件循环空闲一段时间后会调用MemoryIdler::flushLocalMallocCaches()与MemoryIdler::unmapUnusedStack(...)见 folly/executors/IOThreadPoolExecutor.cpp。不过文档也指出当前任务在队列间是 round-robin 调度的所以除非系统完全没有工作否则该优化效果有限getEventBase()按 round-robin 选择调用方可以直接拿到一个EventBase在上面调度 IO 工作队列几乎无竞争由于每线程一个队列队列上的竞争极小因此用自旋锁 std::deque这种轻量结构即可承载任务且没有最大队列长度限制默认每核一线程默认情况下 IO 线程数与 CPU 核数一致——只要这些线程不阻塞配置比核数更多的 IO 线程通常没有意义。构造与选项explicit IOThreadPoolExecutor( size_t numThreads, std::shared_ptrThreadFactory threadFactory std::make_sharedNamedThreadFactory(IOThreadPool), folly::EventBaseManager* ebm folly::EventBaseManager::get(), Options options Options());Options支持三个配置项folly/executors/IOThreadPoolExecutor.h配置项默认值说明waitForAllfalse析构/stop()时是否等待事件循环完全退出enableThreadIdCollectionfalse是否显式开启线程 ID 收集返回WorkerProvidermaxReadAtOnce取 flagfolly_iothreadpoolexecutor_max_read_at_once默认-1即不设限事件循环中每次循环最多读取的事件数其中folly_iothreadpoolexecutor_max_read_at_once与dynamic_iothreadpoolexecutor默认true即 IO 线程池动态创建线程最小线程数从 0 开始两个 gflags 定义在 folly/executors/IOThreadPoolExecutor.cpp。stop() 的行为差异文档与源码都特别强调了一个容易踩坑的点对于 IOThreadPoolExecutorstop()表现得像join()。因为未完成的任务属于事件循环EventBase它们会在 EventBase 析构时被继续执行所以stop()会等待这些任务完成。这与 CPUThreadPoolExecutor 的尽力而为停止语义不同详见 folly/executors/IOThreadPoolExecutor.h。CPUThreadPoolExecutorLifoSem MPMC 单队列CPUThreadPoolExecutor是面向 CPU 密集任务的线程池folly/executors/CPUThreadPoolExecutor.h 的类注释完整说明了其设计单一队列后端是folly::LifoSemfolly::MPMCQueue。由于全局只有一个队列所有工作线程与所有生产者线程都会命中同一条队列竞争可能相当高——而 MPMC 队列多生产者多消费者无锁队列恰好在这种场景下表现出色MPMC 队列决定了存在最大队列长度有界队列在满时的行为由QueueBehaviorIfFull决定。以 folly/executors/task_queue/LifoSemMPMCQueue.h 为例THROW模式会在队列满时抛出QueueFullExceptionBLOCK模式则阻塞写端LifoSem 按 LIFO 顺序唤醒线程即始终只保持恰好够用的少数线程在运行并尽量复用同一批线程以获得更好的 cache locality其余线程被挂起直到出现工作尖峰时才被唤醒空闲线程栈回收所有 Folly 的BlockingQueue实现都基于LifoSem或ThrottledLifoSem长期不活跃的线程其栈会被madvise掉stop()会在退出时完成所有未完成任务支持优先级优先级通过多条队列实现——每个工作线程总是先检查最高优先级的队列。但线程本身并不设置 OS 优先级pthreads 线程优先级在实践中的表现不佳因此一连串长时间运行的低优先级任务仍可能占满所有线程。队列工厂默认、LIFO 与节流 LIFOCPUThreadPoolExecutor提供了一组静态工厂方法用于构造任务队列folly/executors/CPUThreadPoolExecutor.cpp工厂方法后端特点makeDefaultQueue()由 flagfolly_cputhreadpoolexecutor_use_throttled_lifo_sem决定默认走ThrottledLifoSem若 flag 开启否则走LifoSemmakeLifoSemQueue()UnboundedBlockingQueueCPUTask, LifoSem无界、LIFO 唤醒makeThrottledLifoSemQueue(wakeUpInterval)UnboundedBlockingQueueCPUTask, ThrottledLifoSem无界、节流唤醒可配置唤醒间隔makeDefaultPriorityQueue(numPriorities)/makeLifoSemPriorityQueue/makeThrottledLifoSemPriorityQueuePriorityUnboundedBlockingQueue...对应上述语义的多优先级队列版本使用有界优先队列时可用PriorityLifoSemMPMCQueueCPUTask(numPriorities, maxQueueSize)在maxQueueSize构造重载中使用。死锁警告文档对应的源码注释folly/executors/CPUThreadPoolExecutor.h给出了一个重要的实践警告如果使用有界队列QueueBehaviorIfFull::BLOCK且线程池中的任务还会继续向该线程池提交任务一旦队列变满就可能死锁多个使用阻塞队列的线程池之间存在环形依赖时同样可能死锁。规避方式有二只用无界队列默认且推荐的做法或者只从不属于该线程池的线程中提交任务。构造与动态线程数CPUThreadPoolExecutor提供多种构造函数重载包括CPUThreadPoolExecutor(size_t numThreads, Options opt {})最简单形态CPUThreadPoolExecutor(size_t numThreads, int8_t numPriorities, ...)多优先级版本CPUThreadPoolExecutor(size_t numThreads, int8_t numPriorities, size_t maxQueueSize, ...)多优先级 有界队列版本CPUThreadPoolExecutor(std::pairsize_t, size_t numThreads, ...)显式指定(maxThreads, minThreads)。关于动态线程数folly/executors/CPUThreadPoolExecutor.cpp 显示当 flagdynamic_cputhreadpoolexecutor为真时numThreads会被展开为(numThreads, 0)即最小线程数为 0、由线程池按负载动态创建/回收线程为假时则固定线程数。构造时setNumThreads(numThreads.first)会把maxThreads_设为指定值minThreads_会在动态模式下被置为 1保证至少有一个可运行的线程。线程池可以在minThreads_与maxThreads_之间动态变化实际运行线程数由activeThreads_追踪。任务入队后如果无法保证已有活跃线程会处理它就会调用ensureActiveThreads()启动新线程直至达到maxThreads_空闲线程的回收则交由各子类的空闲超时机制完成。相关机制见 folly/executors/ThreadPoolExecutor.h 的类注释与 folly/executors/ThreadPoolExecutor.cpp 的setNumThreads实现。任务过期回调基类ThreadPoolExecutor::add(func, expiration, expireCallback)提供了任务过期语义如果func在入队后expiration时间内尚未开始执行则执行expireCallback见 folly/executors/ThreadPoolExecutor.h。CPUThreadPoolExecutor还额外提供带优先级的add(func, priority, expiration, expireCallback)重载。ThreadPoolExecutor共享的基类逻辑ThreadPoolExecutor是所有具体线程池的基类包含线程的启动 / 停止 / 统计逻辑——这些逻辑与任务具体如何被运行是解耦的因此被提取为公共基类folly/executors/ThreadPoolExecutor.h。它提供的核心能力包括线程生命周期管理addThreads/removeThreads/joinStoppedThreads/stopAndJoinAllThreads动态调线程数setNumThreads、numThreads()、numActiveThreads()、setThreadDeathTimeout()停止与等待stop()尽力而为未执行任务不保证执行完与join()批量遍历withAll(FunctionRefvoid(ThreadPoolExecutor))用于对所有已注册线程池执行操作主要服务于统计导出线程工厂setThreadFactory/getThreadFactory默认实现是NamedThreadFactory如CPUThreadPool、IOThreadPool、GlobalCPUThreadPool等命名前缀见 folly/executors/thread_factoryCPU 时间统计getUsedCpuTime()返回线程池所有线程含已退出线程累计的 CPU 时间需要系统支持 per-thread CPU clock否则返回 0且该操作可能较昂贵。Observers监听线程的创建与销毁ThreadPoolExecutor::Observer是一个观察者接口用于监听线程的 start/stop 事件folly/executors/ThreadPoolExecutor.hclass Observer { public: virtual ~Observer() default; virtual void threadStarted(ThreadHandle*) noexcept {} virtual void threadStopped(ThreadHandle*) noexcept {} virtual void threadPreviouslyStarted(ThreadHandle* h) noexcept { threadStarted(h); } virtual void threadNotYetStopped(ThreadHandle* h) noexcept { threadStopped(h); } };它的典型用途是创建每线程一份的对象如 thread-local 资源、连接池、性能计数器同时保证在线程被动态加入或移出线程池时这些对象也能被正确地创建与清理。通过addObserver/removeObserver注册。IOThreadPoolExecutor还扩展了IOObserver接口额外提供registerEventBase(EventBase)/unregisterEventBase(EventBase)钩子见 folly/executors/IOThreadPoolExecutor.h便于在 IO 线程的 EventBase 创建/销毁时执行对应初始化与清理。Stats线程池运行指标ThreadPoolExecutor::PoolStats结构体提供了线程池级别的统计信息folly/executors/ThreadPoolExecutor.h字段含义threadCount线程池中的线程总数idleThreadCount空闲线程数activeThreadCount活跃线程数pendingTaskCount待执行任务数totalTaskCount累计任务总数processedTaskCount已处理任务数maxIdleTime最长空闲时间通过getPoolStats()获取getPendingTaskCount()则单独返回待执行任务数。如果需要任务级的观测入队、出队、处理完成三个阶段的耗时可以注册TaskObserver接口见 folly/executors/ThreadPoolExecutor.hclass TaskObserver { public: virtual ~TaskObserver() default; virtual void taskEnqueued(const TaskInfo) noexcept {} virtual void taskDequeued(const DequeuedTaskInfo) noexcept {} virtual void taskProcessed(const ProcessedTaskInfo) noexcept {} };TaskInfo/DequeuedTaskInfo/ProcessedTaskInfo层层继承分别携带优先级、requestId、入队时间、taskId以及waitTime出队时间 − 入队时间和runTime处理耗时。任务处理完成后runTask会回调所有已注册的TaskObserver::taskProcessed见 folly/executors/ThreadPoolExecutor.cpp。注意TaskObserver出于性能考虑只能添加、不能移除会在线程池析构时统一销毁旧的subscribeToTaskStats(TaskStatsCallback)接口已被标记为 deprecated建议迁移到addTaskObserver。选型建议与总结回到文档的核心结论选择线程池时可以遵循以下准则IO 密集 / 需要事件循环如 Thrift、memcache 客户端、异步 socket使用IOThreadPoolExecutor通过getEventBase()直接调度 IO 工作默认每核一线程即可除非线程会阻塞CPU 密集计算如 JSON 解析、加密、计算型回调使用CPUThreadPoolExecutor借助 LifoSem 的 LIFO 唤醒获得 cache locality并可选用多优先级队列区分任务重要程度通用异步代码直接使用全局 ExecutorgetGlobalCPUExecutor()/getGlobalIOExecutor()配合via()/then()组成 Future 流水线避免每次手工创建线程池需要监控通过getPoolStats()观察池级指标通过addTaskObserver获取任务级 wait/run 耗时通过 Observer 管理每线程资源注意停止语义差异IOThreadPoolExecutor::stop()类似join()会等待事件循环中的任务完成CPUThreadPoolExecutor::stop()则尽力完成未完成任务后返回避免有界阻塞队列引发死锁优先使用无界队列或只从线程池外部提交任务。这套IO 与 CPU 分离、事件循环与任务队列各司其职的设计是 Folly 得以在 Facebook 大规模生产环境中提供稳定尾延迟的基石之一。理解其背后的系统原语约束event_fd、epoll、信号量、惊群比记住 API 本身更能帮助你做出正确的并发架构决策。相关源码与测试如 folly/executors/test/IOThreadPoolExecutorTest.cpp、folly/executors/test/ThreadPoolExecutorTest.cpp、folly/executors/test/GlobalExecutorTest.cpp可作为进一步研究的入口。【免费下载链接】follyAn open-source C library developed and used at Facebook.项目地址: https://gitcode.com/GitHub_Trending/fol/folly创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Folly Executors 与线程池深度指南：CPUThreadPoolExecutor 与 IOThreadPoolExecutor 的原理、选型与实践
Folly Executors 与线程池深度指南CPUThreadPoolExecutor 与 IOThreadPoolExecutor 的原理、选型与实践【免费下载链接】follyAn open-source C library developed and used at Facebook.项目地址: https://gitcode.com/GitHub_Trending/fol/follyFolly 在 folly/docs/Executors.md 中系统讲解了其线程池Thread Pool与 Executor 体系的设计动机、实现原理和使用方式。本文以该文档为骨架结合 folly/executors 目录下的源码实现深入剖析 Folly 为何要自研两套线程池CPU 与 IO 分离、它们各自的数据结构与调度细节以及如何借助全局 Executor、Observers 与 PoolStats 在生产环境中高效运行并发代码。读完本文你将掌握CPUThreadPoolExecutor、IOThreadPoolExecutor、ThreadPoolExecutor基类的核心机制并能在自己的代码中正确选型与配置。快速上手从全局 Executor 开始Folly 提供两个具体的线程池实现IOThreadPoolExecutor与CPUThreadPoolExecutor并将它们作为完整异步框架folly/futures的一部分内置。大多数场景下你并不需要手动构造线程池而是直接获取进程级的全局 Executor再与 Future 组合使用。最简单、最常见的用法是将 Future 延续continuation调度到 CPU 线程池上执行auto f someFutureFunction().via(getCPUExecutor()).then(...);如果需要在 Thrift / memcache 客户端上发起调用则需要先拿到一个事件循环EventBase此时从 IO 线程池获取getEventBase()再通过via(getCPUExecutor())回到 CPU 线程池处理结果auto f getClient(getIOExecutor()-getEventBase())-callSomeFunction(args...) .via(getCPUExecutor()) .then([](Result r) { /* do something with result */ });getCPUExecutor()返回全局 CPU 线程池folly/executors/GlobalExecutor.hgetIOExecutor()返回全局 IO 线程池可用-getEventBase()取出一个按 round-robin 方式选中的EventBase直接在上面调度 IO 工作。全局 Executor 的线程数与新旧 API全局 Executor 的线程数量可以通过 gflags 显式配置folly_global_cpu_executor_threads全局 CPU 线程池创建的线程数folly_global_io_executor_threads全局 IO 线程池创建的线程数。两者默认值均为 0。在 folly/executors/GlobalExecutor.cpp 中可以看到当 flag 为 0 时线程数会回退为folly::available_concurrency()即机器可用的核数size_t nthreads FLAGS_folly_global_cpu_executor_threads; nthreads nthreads ? nthreads : folly::available_concurrency(); return new std::shared_ptrImmutableGlobalCPUExecutor( new ImmutableGlobalCPUExecutor( nthreads, std::make_sharedNamedThreadFactory(GlobalCPUThreadPool)));需要说明的是文档中示例使用的getCPUExecutor()/getIOExecutor()在当前版本中已被标记为 deprecated。从 folly/executors/GlobalExecutor.h 的注释可以看出官方推荐使用getGlobalCPUExecutor()/getGlobalIOExecutor()获取不可变的全局 Executor返回KeepAlive可安全配合 Future / coroutine 使用保证前向进度若确有替换全局 Executor 的需求使用getUnsafeMutableGlobalCPUExecutor()/setUnsafeMutableGlobalCPUExecutor()等 UnsafeMutable 系列接口getGlobalCPUExecutorWeakRef()返回弱 KeepAlive不阻止全局 Executor 在关闭时析构适合尽力而为的后台任务但不应用于 Future / coroutine 延续因为延续依赖前向进度保证弱引用可能导致死锁。为什么不用 C11 的 std::launchC11 的std::launch只有两种模式async与deferred在生产系统中两者都不是理想选择async每次 launch 都会无限制地新起一个线程线程数量完全不受控deferred任务被延迟到真正需要结果时才执行且届时在当前线程里同步地运行阻塞调用方。Folly 的线程池则不同任务总是尽可能早地被调度执行同时对最大任务数 / 线程数设有上限因此永远不会使用超出需要的线程数。这种按需生长、有界并发的模型正是生产级服务所需要的。为什么需要自研线程池文档给出的理由是当时C11 时代现成的线程池实现都不完整——基于 pipe 的线程池太慢而多个较早的实现不支持std::function无法与 Future/回调风格的代码无缝协作。因此 Folly 需要自己提供一套面向高并发、低延迟、且与std::function和异步框架深度集成的线程池实现。为什么需要两种不同类型的线程池这是理解 Folly 线程池架构最关键的问题核心在于IO 事件循环与公平队列在操作系统原语层面互相排斥epoll 需要 fdevent_fd是最新的通知机制但它有个副作用——一个活跃的 fd 会触发所有正在等待它的 epoll 循环惊群效应thundering herd。因此如果你想要一个公平队列一个总队列 vs. 每工作线程一个队列就需要借助信号量semaphore来实现公平唤醒信号量进不了 epoll 循环semaphore 无法被放入 epoll 等待集合中所以基于信号量的公平队列与 IO 事件循环不兼容IO 与 CPU 本就该分离即便技术上可行通常也应当把 IO 密集与 CPU 密集的工作分开以便在 IO 路径上获得更强的尾延迟tail latency保证。正是基于上述原因Folly 提供了两套定位不同的线程池IOThreadPoolExecutor面向事件循环与 IO和CPUThreadPoolExecutor面向计算密集任务。IOThreadPoolExecutorevent_fd 每线程 NotificationQueueIOThreadPoolExecutor是一个面向 IO 密集型任务的线程池folly/executors/IOThreadPoolExecutor.h 的类注释完整概括了它的设计要点使用event_fd进行通知并唤醒 epoll 循环每个线程每个 epoll对应一个队列具体是NotificationQueueIOThread结构中持有自己的EventBase*任务通过ioThread-eventBase-runInEventBaseThread(...)投递见 folly/executors/IOThreadPoolExecutor.cpp无谓系统调用被消除如果目标线程已经在运行、并没有阻塞在 epoll 上等待那么只需要把新任务放进它的队列即可无需额外的 syscall 去唤醒事件循环空闲线程回收内存如果某个线程等待超过数秒它的栈会被madvise掉。实现上由MemoryIdlerTimeout完成——它是AsyncTimeoutEventBase::LoopCallback的合体事件循环空闲一段时间后会调用MemoryIdler::flushLocalMallocCaches()与MemoryIdler::unmapUnusedStack(...)见 folly/executors/IOThreadPoolExecutor.cpp。不过文档也指出当前任务在队列间是 round-robin 调度的所以除非系统完全没有工作否则该优化效果有限getEventBase()按 round-robin 选择调用方可以直接拿到一个EventBase在上面调度 IO 工作队列几乎无竞争由于每线程一个队列队列上的竞争极小因此用自旋锁 std::deque这种轻量结构即可承载任务且没有最大队列长度限制默认每核一线程默认情况下 IO 线程数与 CPU 核数一致——只要这些线程不阻塞配置比核数更多的 IO 线程通常没有意义。构造与选项explicit IOThreadPoolExecutor( size_t numThreads, std::shared_ptrThreadFactory threadFactory std::make_sharedNamedThreadFactory(IOThreadPool), folly::EventBaseManager* ebm folly::EventBaseManager::get(), Options options Options());Options支持三个配置项folly/executors/IOThreadPoolExecutor.h配置项默认值说明waitForAllfalse析构/stop()时是否等待事件循环完全退出enableThreadIdCollectionfalse是否显式开启线程 ID 收集返回WorkerProvidermaxReadAtOnce取 flagfolly_iothreadpoolexecutor_max_read_at_once默认-1即不设限事件循环中每次循环最多读取的事件数其中folly_iothreadpoolexecutor_max_read_at_once与dynamic_iothreadpoolexecutor默认true即 IO 线程池动态创建线程最小线程数从 0 开始两个 gflags 定义在 folly/executors/IOThreadPoolExecutor.cpp。stop() 的行为差异文档与源码都特别强调了一个容易踩坑的点对于 IOThreadPoolExecutorstop()表现得像join()。因为未完成的任务属于事件循环EventBase它们会在 EventBase 析构时被继续执行所以stop()会等待这些任务完成。这与 CPUThreadPoolExecutor 的尽力而为停止语义不同详见 folly/executors/IOThreadPoolExecutor.h。CPUThreadPoolExecutorLifoSem MPMC 单队列CPUThreadPoolExecutor是面向 CPU 密集任务的线程池folly/executors/CPUThreadPoolExecutor.h 的类注释完整说明了其设计单一队列后端是folly::LifoSemfolly::MPMCQueue。由于全局只有一个队列所有工作线程与所有生产者线程都会命中同一条队列竞争可能相当高——而 MPMC 队列多生产者多消费者无锁队列恰好在这种场景下表现出色MPMC 队列决定了存在最大队列长度有界队列在满时的行为由QueueBehaviorIfFull决定。以 folly/executors/task_queue/LifoSemMPMCQueue.h 为例THROW模式会在队列满时抛出QueueFullExceptionBLOCK模式则阻塞写端LifoSem 按 LIFO 顺序唤醒线程即始终只保持恰好够用的少数线程在运行并尽量复用同一批线程以获得更好的 cache locality其余线程被挂起直到出现工作尖峰时才被唤醒空闲线程栈回收所有 Folly 的BlockingQueue实现都基于LifoSem或ThrottledLifoSem长期不活跃的线程其栈会被madvise掉stop()会在退出时完成所有未完成任务支持优先级优先级通过多条队列实现——每个工作线程总是先检查最高优先级的队列。但线程本身并不设置 OS 优先级pthreads 线程优先级在实践中的表现不佳因此一连串长时间运行的低优先级任务仍可能占满所有线程。队列工厂默认、LIFO 与节流 LIFOCPUThreadPoolExecutor提供了一组静态工厂方法用于构造任务队列folly/executors/CPUThreadPoolExecutor.cpp工厂方法后端特点makeDefaultQueue()由 flagfolly_cputhreadpoolexecutor_use_throttled_lifo_sem决定默认走ThrottledLifoSem若 flag 开启否则走LifoSemmakeLifoSemQueue()UnboundedBlockingQueueCPUTask, LifoSem无界、LIFO 唤醒makeThrottledLifoSemQueue(wakeUpInterval)UnboundedBlockingQueueCPUTask, ThrottledLifoSem无界、节流唤醒可配置唤醒间隔makeDefaultPriorityQueue(numPriorities)/makeLifoSemPriorityQueue/makeThrottledLifoSemPriorityQueuePriorityUnboundedBlockingQueue...对应上述语义的多优先级队列版本使用有界优先队列时可用PriorityLifoSemMPMCQueueCPUTask(numPriorities, maxQueueSize)在maxQueueSize构造重载中使用。死锁警告文档对应的源码注释folly/executors/CPUThreadPoolExecutor.h给出了一个重要的实践警告如果使用有界队列QueueBehaviorIfFull::BLOCK且线程池中的任务还会继续向该线程池提交任务一旦队列变满就可能死锁多个使用阻塞队列的线程池之间存在环形依赖时同样可能死锁。规避方式有二只用无界队列默认且推荐的做法或者只从不属于该线程池的线程中提交任务。构造与动态线程数CPUThreadPoolExecutor提供多种构造函数重载包括CPUThreadPoolExecutor(size_t numThreads, Options opt {})最简单形态CPUThreadPoolExecutor(size_t numThreads, int8_t numPriorities, ...)多优先级版本CPUThreadPoolExecutor(size_t numThreads, int8_t numPriorities, size_t maxQueueSize, ...)多优先级 有界队列版本CPUThreadPoolExecutor(std::pairsize_t, size_t numThreads, ...)显式指定(maxThreads, minThreads)。关于动态线程数folly/executors/CPUThreadPoolExecutor.cpp 显示当 flagdynamic_cputhreadpoolexecutor为真时numThreads会被展开为(numThreads, 0)即最小线程数为 0、由线程池按负载动态创建/回收线程为假时则固定线程数。构造时setNumThreads(numThreads.first)会把maxThreads_设为指定值minThreads_会在动态模式下被置为 1保证至少有一个可运行的线程。线程池可以在minThreads_与maxThreads_之间动态变化实际运行线程数由activeThreads_追踪。任务入队后如果无法保证已有活跃线程会处理它就会调用ensureActiveThreads()启动新线程直至达到maxThreads_空闲线程的回收则交由各子类的空闲超时机制完成。相关机制见 folly/executors/ThreadPoolExecutor.h 的类注释与 folly/executors/ThreadPoolExecutor.cpp 的setNumThreads实现。任务过期回调基类ThreadPoolExecutor::add(func, expiration, expireCallback)提供了任务过期语义如果func在入队后expiration时间内尚未开始执行则执行expireCallback见 folly/executors/ThreadPoolExecutor.h。CPUThreadPoolExecutor还额外提供带优先级的add(func, priority, expiration, expireCallback)重载。ThreadPoolExecutor共享的基类逻辑ThreadPoolExecutor是所有具体线程池的基类包含线程的启动 / 停止 / 统计逻辑——这些逻辑与任务具体如何被运行是解耦的因此被提取为公共基类folly/executors/ThreadPoolExecutor.h。它提供的核心能力包括线程生命周期管理addThreads/removeThreads/joinStoppedThreads/stopAndJoinAllThreads动态调线程数setNumThreads、numThreads()、numActiveThreads()、setThreadDeathTimeout()停止与等待stop()尽力而为未执行任务不保证执行完与join()批量遍历withAll(FunctionRefvoid(ThreadPoolExecutor))用于对所有已注册线程池执行操作主要服务于统计导出线程工厂setThreadFactory/getThreadFactory默认实现是NamedThreadFactory如CPUThreadPool、IOThreadPool、GlobalCPUThreadPool等命名前缀见 folly/executors/thread_factoryCPU 时间统计getUsedCpuTime()返回线程池所有线程含已退出线程累计的 CPU 时间需要系统支持 per-thread CPU clock否则返回 0且该操作可能较昂贵。Observers监听线程的创建与销毁ThreadPoolExecutor::Observer是一个观察者接口用于监听线程的 start/stop 事件folly/executors/ThreadPoolExecutor.hclass Observer { public: virtual ~Observer() default; virtual void threadStarted(ThreadHandle*) noexcept {} virtual void threadStopped(ThreadHandle*) noexcept {} virtual void threadPreviouslyStarted(ThreadHandle* h) noexcept { threadStarted(h); } virtual void threadNotYetStopped(ThreadHandle* h) noexcept { threadStopped(h); } };它的典型用途是创建每线程一份的对象如 thread-local 资源、连接池、性能计数器同时保证在线程被动态加入或移出线程池时这些对象也能被正确地创建与清理。通过addObserver/removeObserver注册。IOThreadPoolExecutor还扩展了IOObserver接口额外提供registerEventBase(EventBase)/unregisterEventBase(EventBase)钩子见 folly/executors/IOThreadPoolExecutor.h便于在 IO 线程的 EventBase 创建/销毁时执行对应初始化与清理。Stats线程池运行指标ThreadPoolExecutor::PoolStats结构体提供了线程池级别的统计信息folly/executors/ThreadPoolExecutor.h字段含义threadCount线程池中的线程总数idleThreadCount空闲线程数activeThreadCount活跃线程数pendingTaskCount待执行任务数totalTaskCount累计任务总数processedTaskCount已处理任务数maxIdleTime最长空闲时间通过getPoolStats()获取getPendingTaskCount()则单独返回待执行任务数。如果需要任务级的观测入队、出队、处理完成三个阶段的耗时可以注册TaskObserver接口见 folly/executors/ThreadPoolExecutor.hclass TaskObserver { public: virtual ~TaskObserver() default; virtual void taskEnqueued(const TaskInfo) noexcept {} virtual void taskDequeued(const DequeuedTaskInfo) noexcept {} virtual void taskProcessed(const ProcessedTaskInfo) noexcept {} };TaskInfo/DequeuedTaskInfo/ProcessedTaskInfo层层继承分别携带优先级、requestId、入队时间、taskId以及waitTime出队时间 − 入队时间和runTime处理耗时。任务处理完成后runTask会回调所有已注册的TaskObserver::taskProcessed见 folly/executors/ThreadPoolExecutor.cpp。注意TaskObserver出于性能考虑只能添加、不能移除会在线程池析构时统一销毁旧的subscribeToTaskStats(TaskStatsCallback)接口已被标记为 deprecated建议迁移到addTaskObserver。选型建议与总结回到文档的核心结论选择线程池时可以遵循以下准则IO 密集 / 需要事件循环如 Thrift、memcache 客户端、异步 socket使用IOThreadPoolExecutor通过getEventBase()直接调度 IO 工作默认每核一线程即可除非线程会阻塞CPU 密集计算如 JSON 解析、加密、计算型回调使用CPUThreadPoolExecutor借助 LifoSem 的 LIFO 唤醒获得 cache locality并可选用多优先级队列区分任务重要程度通用异步代码直接使用全局 ExecutorgetGlobalCPUExecutor()/getGlobalIOExecutor()配合via()/then()组成 Future 流水线避免每次手工创建线程池需要监控通过getPoolStats()观察池级指标通过addTaskObserver获取任务级 wait/run 耗时通过 Observer 管理每线程资源注意停止语义差异IOThreadPoolExecutor::stop()类似join()会等待事件循环中的任务完成CPUThreadPoolExecutor::stop()则尽力完成未完成任务后返回避免有界阻塞队列引发死锁优先使用无界队列或只从线程池外部提交任务。这套IO 与 CPU 分离、事件循环与任务队列各司其职的设计是 Folly 得以在 Facebook 大规模生产环境中提供稳定尾延迟的基石之一。理解其背后的系统原语约束event_fd、epoll、信号量、惊群比记住 API 本身更能帮助你做出正确的并发架构决策。相关源码与测试如 folly/executors/test/IOThreadPoolExecutorTest.cpp、folly/executors/test/ThreadPoolExecutorTest.cpp、folly/executors/test/GlobalExecutorTest.cpp可作为进一步研究的入口。【免费下载链接】follyAn open-source C library developed and used at Facebook.项目地址: https://gitcode.com/GitHub_Trending/fol/folly创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考