先别急着抄代码得先把一个事情想明白asio 的io_context本身只是个事件循环单线程跑它代码确实好写但 CPU 多核时代里一个核满转、其他核看戏这就不太像话了。这篇要讲的IOServicePool就是解决这个问题的常见多线程模型它的思路不是简单开一堆线程去抢同一个io_context而是把“一堆 io_context 一堆线程”做成池新建连接时轮询分发给某一个io_context让同一条连接的所有异步回调始终落在同一个线程里。这样既吃满多核又保住了单连接内部几乎不加锁的清爽体验。写给谁看已经能用 asio 写出最简单的异步 accept / async_read 的人或者正在把单线程网络服务改成多线程、却被并发问题折腾得头疼的人。看完你能得到一个可以直接抄作业的IOServicePool类、一个配套的 echo server以及我在实战里踩过的一些坑。1. 先搞清楚为什么单线程IO跑不满CPU才需要这个池子1.1 单io_context 单线程代码最简单扩展性也最差用单线程跑io_context::run()是最容易理解的模型。所有异步事件排在一个队列里线程从队列里取任务一个个执行。好处是天然不会出现数据竞争因为你根本不会同时有两个执行者在碰同一份数据。坏处也很明显你的服务端再怎么能扛也只有一个执行者。假设机器有 8 核其他 7 个核基本闲着一旦业务里带点 CPU 密集的逻辑比如编解码、校验、消息拼包单线程吞吐很快就触顶。很多人觉得“异步 高性能”但异步只解决 IO 等待问题CPU 利用率还是要靠多线程把任务分散到不同核上去跑。早几年我刚用 asio 时也犯过这个误区写了个 echo server压测一看CPU 使用率稳定在 100% 左右但那是单个核的 100%系统总 CPU 占用可能只有 12.5%。到了这一步就明白网络层的人手再多带宽也擀不出第二张饼。1.2 单io_context 多线程看似能并行锁又回来了既然单线程不够很多人第一反应是让多个线程同时调用同一个io_context的run()。这确实能让事件循环变成“一个队列多个消费者”不同连接的事件可以被不同线程并行处理。但这样做有几个很微妙的代价。首先是队列锁竞争。多个消费者同时从任务队列里取任务必然要对队列本身加锁连接数量大了之后这个锁会成为新的热点。其次同一个连接的前后两个异步回调可能被不同的线程执行。这一点最伤你在回调 A 里往某个连接对象上写了一个中间状态下一次回调 B 跑在另一个线程上访问同一个状态时数据竞争就来了。为了安全你又得给每个连接加 strand 或者加锁。结果就是代码从“无锁”变成“到处锁”复杂度上去了性能还不一定比单线程好多少。所以“多个线程跑一个 io_context”这个方案并不是银弹它更适合连接量不大、跨线程状态非常少的场景。真想撑大并发得换思路。1.3 IO Service Pool的核心思路连接与线程固定绑定IO Service Pool 的核心思想可以总结成一句话把并发从“连接内部”挪到“连接之间”。具体做法是池里准备 N 个io_context每个io_context配一个线程跑run()。当 acceptor 收到新连接时按轮询方式把这条连接分发给其中一个io_context之后这条连接的所有异步操作都挂在这个io_context上。因为每个io_context只有一个线程所以同一条连接的读写回调永远在同一个线程里串行执行不需要加锁。这就像你去银行办事每个窗口一个柜员客户取号后被叫到某个窗口钥匙、资料、问题全在同一个柜员手里处理完你不用在窗口之间跑来跑去柜员之间也不会因为同一份资料打架。把三种模型放一起看会更直观模型线程数同一连接回调线程锁需求多核利用率单 io_context 单线程1固定同一线程无低单 io_context 多线程N可能不同线程需要锁或 strand高但有竞争IO Service PoolN固定同一线程连接内无锁高且竞争小这个模型在 C 网络服务里很常见很多项目虽然不是完全同名但底层思路一模一样。接下来我们就把它真的写出来。2. 从零实现一个IOServicePool2.1 类成员的选型io_context、work_guard和线程怎么组合先想清楚类里要装什么。最核心的是三个东西std::vectorio_context池里的多个事件循环std::vectorWorkGuard防止run()在空闲时退出std::vectorstd::thread每个 io_context 对应的执行线程为什么要 work_guard 这是新手最容易疑惑的点。io_context::run()的规则是如果任务队列里没有可执行的任务它会立刻返回。如果你池子刚建好还没有任何连接进来这时线程跑run()可能一瞬间就退出来了后面连接真来了也没人处理。work_guard的作用就是给io_context挂一个“长期 work 项”让run()在等待任务时不返回直到你主动reset()掉它。可以理解成给事件循环上了一份“保底工资”没事干也得待命。另外还要注意 io_context 不可拷贝所以成员存储用 vector 时要通过emplace_back()构造千万别尝试拷贝。2.2 关键代码构造、启动、停止一个都不能少下面是一份完整可用的实现兼容 standalone 版 asio也兼容 boost.asio。如果你的项目用的是 boost.asio提前定义USE_BOOST_ASIO这个宏就行。#pragma once #include atomic #include cstddef #include thread #include vector #ifndef USE_BOOST_ASIO #include asio.hpp namespace net asio; #else #include boost/asio.hpp namespace net boost::asio; #endif class IOServicePool { public: using IOService net::io_context; using WorkGuard net::executor_work_guardIOService::executor_type; explicit IOServicePool(std::size_t size 0) : m_nextService(0) { if (size 0) { size std::thread::hardware_concurrency(); } if (size 0) { size 2; } m_ioServices.reserve(size); m_workGuards.reserve(size); for (std::size_t i 0; i size; i) { m_ioServices.emplace_back(); m_workGuards.emplace_back(net::make_work_guard(m_ioServices.back())); } } IOServicePool(const IOServicePool) delete; IOServicePool operator(const IOServicePool) delete; ~IOServicePool() { stop(); } // 轮询方式选择一个 io_context IOService getIOService() { std::size_t index m_nextService.fetch_add(1) % m_ioServices.size(); return m_ioServices[index]; } // 启动所有线程 void run() { if (m_started.exchange(true)) { return; } for (std::size_t i 0; i m_ioServices.size(); i) { m_threads.emplace_back([this, i]() { try { m_ioServices[i].run(); } catch (const std::exception e) { // 这里建议记日志不要吞异常 } }); } } // 停止并等待所有线程退出 void stop() { for (auto guard : m_workGuards) { guard.reset(); } for (auto io : m_ioServices) { io.stop(); } for (auto thread : m_threads) { if (thread.joinable()) { thread.join(); } } m_threads.clear(); m_started false; } private: std::vectorIOService m_ioServices; std::vectorWorkGuard m_workGuards; std::vectorstd::thread m_threads; std::atomicstd::size_t m_nextService; std::atomicbool m_started{false}; };几个细节说明一下。构造函数里size std::thread::hardware_concurrency()表示默认线程数等于 CPU 逻辑核数。这个选择后面会专门讨论不是越大越好。如果hardware_concurrency()返回 0保底给 2避免空指针或除零问题。run()启动时用m_started.exchange(true)判断是否已经启动避免重复创建线程。如果已经启动过直接返回。stop()里先reset()掉所有 work_guard再对每个 io_context 调用stop()最后join()所有线程。顺序上可以先 reset 再 stop也可以只 stop。不过建议都写上语义更明确reset 是让 run 在“无事可做”时可以正常返回stop 是强制打断所有阻塞中的事件循环join 是确保线程资源被回收。2.3 轮询分配索引为什么必须用原子变量getIOService()里的m_nextService.fetch_add(1) % m_ioServices.size()这行代码看着简单但很值得说一说。fetch_add是原子操作它返回的是加之前的值。所以第一次调用返回 0第二次返回 1第三次返回 2以此类推再取模就实现了轮询。为什么不能用普通的std::size_t自增因为accept回调可能发生在 acceptor 所在线程而多个连接几乎同时到达时可能会有多个线程同时进入getIOService()。如果普通整数自增两个线程同时读到同一个值就会把两条连接分到同一个 io_context破坏轮询的均衡性甚至出现数据竞争。用std::atomicstd::size_t则能保证每一次取号都是唯一的。有些实现会一次性接受多个连接才分发一次但这里单次 accept 一个就足够教学和大多数业务使用了。3. 用IOServicePool写一台可运行的Echo服务器光有池子还看不出效果得把它接进一个真正的服务器里。下面我用一个最小 echo 服务器演示完整的接入流程。3.1 Session对象连接级无锁的关键连接处理对象我习惯叫Session。在 IO Service Pool 模型里Session 最重要的一个设计点就是它从创建到销毁始终只跟某一个 io_context 交互。#include array #include memory #include asio.hpp class Session : public std::enable_shared_from_thisSession { public: explicit Session(net::ip::tcp::socket socket) : m_socket(std::move(socket)) { } void start() { doRead(); } private: void doRead() { auto self shared_from_this(); m_socket.async_read_some( net::buffer(m_data), [this, self](const std::error_code ec, std::size_t length) { if (ec) { return; } doWrite(length); }); } void doWrite(std::size_t length) { auto self shared_from_this(); net::async_write( m_socket, net::buffer(m_data, length), [this, self](const std::error_code ec, std::size_t) { if (ec) { return; } doRead(); }); } net::ip::tcp::socket m_socket; std::arraychar, 4096 m_data{}; };关键点有两个。第一个是std::enable_shared_from_this。每个异步回调里都捕获self保证这条 Session 在挂起一个异步操作期间不会被析构。如果你忘记捕获 self一个操作完成后对象被释放下一次回调就会访问到野指针这是 asio 编程里非常经典的崩溃源。第二个是async_write而不是async_write_some。async_write会把 buffer 里的数据全部写完才回调而async_write_some可能只写了一部分。echo 场景里数据量不大但养成用async_write的习惯能少踩很多“写一半”的坑。3.2 acceptor分发连接千万别在acceptor线程里做业务接下来是服务器主体。acceptor 我用一个独立的 io_context 来跑不混进业务池里。这样职责清晰accept 只负责接受连接和分发不碰业务数据。#include iostream #include memory #include asio.hpp class EchoServer { public: EchoServer(net::io_context acceptorIo, IOServicePool pool, std::uint16_t port) : m_acceptor(acceptorIo, net::ip::tcp::endpoint(net::ip::tcp::v4(), port)), m_pool(pool) { doAccept(); } private: void doAccept() { auto io m_pool.getIOService(); m_acceptor.async_accept( io, [this](const std::error_code ec, net::ip::tcp::socket socket) { if (!ec) { std::make_sharedSession(std::move(socket))-start(); } doAccept(); }); } net::ip::tcp::acceptor m_acceptor; IOServicePool m_pool; };这里最关键的代码是m_acceptor.async_accept(io, handler)。注意我把m_pool.getIOService()作为第一个参数传给了async_accept。这个重载的作用是接受新连接时让新 socket 直接关联到目标 io_context 的 executor 上而不是 acceptor 自己的 io_context。很多人在初次实现 IO Service Pool 时会犯一个错他们先正常async_accept拿到 socket然后在 accept handler 里把 socket 丢给另一个线程去处理。表面上看连接确实换了一个线程处理但 socket 内部关联的执行器仍然指向 acceptor 的 io_context所以后续异步回调还是会派发回 acceptor 线程池子形同虚设。所以分发连接这件事必须在 accept 阶段完成不是后续转手完成的。这也是这个模型最核心、最容易被忽略的技术点。3.3 main函数和验证手段打印线程ID看效果main 函数按下面的方式写就行。int main() { try { IOServicePool pool; net::io_context acceptorIo; EchoServer server(acceptorIo, pool, 12345); pool.run(); acceptorIo.run(); } catch (const std::exception e) { std::cerr exception: e.what() std::endl; return 1; } return 0; }先启动业务线程池再让主线程跑 acceptor 的 io_context。这样 accept 回调一旦触发就能立刻分发到已经在运行的业务线程上。验证方法也很简单在 Session 的doRead里加一行日志打印“当前会话 ID 线程 ID”再在 accept handler 里打印接受连接时的线程 ID。你观察同一条连接的读写回调线程 ID 应该始终一致不同连接的线程 ID 则可能不同。如果发现同一连接在不同线程上跳动优先怀疑是不是async_accept没有传入目标 io_context或者代码里用了不属于这个连接绑定线程的 strand/post。4. 实战中一定会踩的坑与排查记录这部分内容我整理成表格每一行都是实际能遇到的问题最后再展开说几个重要的。现象根本原因排查与解决程序启动后很快退出没有任何报错io_context 没有可执行任务run 直接返回检查是否持有 work_guard或者是否有挂起的异步 acceptaccept 后回调全部跑在 acceptor 线程业务线程没有负载socket 没有在 accept 阶段绑定目标 io_context使用async_accept(io, handler)的重载让新 socket 直接关联目标 executor多个线程同时访问共享的全局连接表偶发崩溃跨连接的共享状态没有加锁连接内可以无锁跨连接共享变量必须加锁或把具体任务 post 到目标 io_contextstop() 后进程还卡着不退出有线程还阻塞在 run或 join 顺序不对先 reset work_guard再 stop io_context最后 join 所有线程连接数一多某个线程 CPU 明显偏高其他线程空闲轮询分发不均匀长连接和短连接混在一起可改为记录每个 io_context 的活动连接数分发给最空闲的 io_context4.1 程序启动就退出work_guard怎么救场我自己第一次写 io_context 池时程序启动后一秒就退出了完全没有任何异常。排查过程也很折磨人因为代码既没崩也没报错。后来才意识到是work_guard的问题。io_context::run()的工作方式是“有活干就干活没活干就回家”。如果你没有挂一个 pending 的异步操作也没有 work_guardrun 会立刻返回线程自然就结束了。IO Service Pool 里线程是要长期待命的所以每个 io_context 都必须配一个 work_guard。还要注意一个细节work_guard 是“持有”才会生效。如果你创建了一个临时 work_guard用完就丢那它也不会起任何作用。我见过有人在构造函数里写了auto guard net::make_work_guard(io);但 guard 是个局部变量一离开作用域就被析构了池子还是照样退出。work_guard 必须存在类成员里生命周期跟 io_context 一样长。4.2 连接回调全跑到acceptor线程associated executor问题如果你把 accept 到的新 socketstd::move到 Session 里然后在 Session 的构造函数中直接async_read_some却发现回调最终还是在 acceptor 线程里执行这不是错觉而是 asio 的 associated executor 机制在起作用。每个异步对象都有一个关联的执行器默认情况下 socket 在执行它创建的 io_context 上。如果这个 socket 不是通过async_accept(io, handler)关联到新 io_context那么它后续的异步回调都会投递回原来的执行器。所以正确姿势是要让连接绑定到哪一个 io_context必须从 accept 那一步就指定。再强调一次这不是能不能 move socket 的问题而是 executor 绑定的问题。4.3 回调里真的不用加锁吗要看你共享了什么IO Service Pool 的优点确实是“单连接内无锁”但这句话有严格边界。它只保证同一条连接的回调串行执行不保证不同连接之间不共享数据。举个例子你有一个全局的std::unordered_mapsessionId, State用来记录所有连接的状态。一个连接的回调里往 map 里写数据另一个连接的回调里读数据这两个回调可能被分配到两个不同的 io_context 线程上于是数据竞争还是会出现。这种跨连接共享数据必须加锁或者把涉及共享数据的操作 post 到同一个 io_context / strand 上执行。边界理清楚之后代码能写得很清爽连接内部的 Session 成员变量随意读写不用锁连接之间的公共资源老老实实加锁或串行化。4.4 如何在CtrlC时优雅关闭服务器简单教学 demo 可以不处理信号但如果你把它接进真实服务会发现直接 CtrlC 导致进程退出时很多日志和连接状态都来不及处理。比较常见的做法是注册一个信号处理器收到 SIGINT 或 SIGTERM 后调用pool.stop()并让 acceptor 的 io_context 也停下来。你可以用一个全局或静态的 io_context 指针在信号处理函数中调用 stop。注意 stop 要做成可重入的用m_started标志防重复。实际项目中我会在stop()里先停止接受新连接再等待已有连接完成当前请求最后回收连接资源。但这套“优雅停机”逻辑比 IO Service Pool 本身复杂得多这里点到为止。5. 线程数怎么定、均衡性怎么调5.1 池大小默认值到底怎么选我在代码里默认用std::thread::hardware_concurrency()也就是 CPU 逻辑核数。这个选择在纯 echo 场景下问题不大因为业务逻辑短IO 线程大多在 epoll_wait 上等待并不会把每个核都占满。但如果你的业务里带了不少 CPU 计算建议从核数开始做压测观察各核占用。两套实际经验可以参考如果 acceptor 单独占用一个线程可以把业务池线程数设为hardware_concurrency() - 1避免线程切换太频繁。如果业务中还有独立的逻辑线程池比如数据库操作线程、编解码线程池IO 池可以再小一点比如核数的一半。IO 池只负责网络收发和轻量业务真正重的计算不应该占着网络线程。压测时不要只看最大连接数要看“在目标延迟内能达到多少 QPS”。我用 echo 测试时四核机器上单线程模型吞吐到一定水平就顶死了换成四线程池之后吞吐明显往上走但连接量很小的时候两者差距不大。线程数不是越大越好开太多线程光是上下文切换就会把性能吃回去。5.2 轮询不均衡怎么办从固定轮询到动态调度固定轮询实现简单但会有一个实际问题如果一条连接是长连接另一条是短连接轮询分配会让长连接长时间占着某个 io_context而其他连接不断新来导致某些线程忙死、某些线程闲死。要解决也很直接在getIOService()里不再简单轮询而是选择当前活动连接数最少的 io_context。每个 io_context 维护一个std::atomicsize_t的“活动连接数”计数连接建立时加一连接销毁时减一。然后取 count 最小的 io_context。每次 accept 才需要选一次连接频率远低于业务回调频率所以即使对计数加个简单锁性能影响也可忽略。这是我实测下来比较有效的改进方案。5.3 IO线程里的阻塞操作一个最隐蔽的性能杀手最后一个坑必须单独拎出来说不要在 io_context 的回调里做阻塞操作。无论是sleep、磁盘同步读、还是调用一个会等锁的第三方库只要阻塞了其中一个 io_context 线程这整个 io_context 上的所有连接就都卡住了。这比单线程模型更可怕因为你不是卡一个线程而是卡了一批连接。解决方向有两个。第一把涉及阻塞的调用放到独立的线程池完成后通过net::post把结果抛回连接所在的 io_context让连接生命周期的处理仍然保持串行。第二如果阻塞时间很短可以考虑异步版本比如文件用异步打开读数据库用异步客户端。实在没有异步版本再用前一种方案。我早期踩过一次很惨的某个队列中间件 SDK 的网络回调里做了磁盘同步写压测一段时间后整个服务所有连接集体超时。查了半天发现不是系统资源不够而是一个 io_context 线程被磁盘 IO 卡住了池子里其他线程也跟着受牵连。从那以后我定了一条规矩io_context 线程里只准做内存操作和异步调用任何可能阻塞的活都扔给别的线程。几句实在话IO Service Pool 是我用 asio 写多线程网络服务时最顺手的模型之一。它没有玄学就是把“让连接绑定线程”这个原则严格落到代码里。我的体会是模型的代码本身不难写难的是你要时刻记住两个边界一是连接必须在 accept 阶段绑定目标 io_context二是“连接内无锁”不等于“全局无锁”。把这两条边界守住这套东西在几千连接的场景下能跑得很稳。最后再分享一个小经验写完这个池子别急着压测先在日志里把 session 和线程 ID 打出来多发几个连接观察一下。模型对不对看线程 ID 的分布比看计数器直观得多。等确认同一条连接固定在一个线程上之后再去做性能优化也不迟。