Rust 异步并发引擎设计复盘：从 tokio 默认配置到定制化调度器的 5 倍吞吐提升

Rust 异步并发引擎设计复盘：从 tokio 默认配置到定制化调度器的 5 倍吞吐提升

一、Tokio 默认配置的隐蔽天花板：为什么 6 核 CPU 跑不满

团队内部的消息分发引擎基于 Rust + Tokio 实现，承担着每秒 20 万条消息的扇出分发任务。在 8 核 16G 的物理机上部署后，最大稳定吞吐量仅停留在 4.2 万 QPS，CPU 利用率在 55%~60% 之间徘徊——6 个核有 2.5 个处于闲置状态。

常规思维会认为 Tokio 的 work-stealing 调度器已经足够高效，但火焰图和 tokio-console 的分析显示，问题在于任务粒度与调度策略的错配。系统中有两类任务：I/O 密集的消息接收任务（微秒级的 epoll 唤醒）和计算密集的序列化/压缩任务（毫秒级）。Tokio 默认的调度器对这两种任务一视同仁，导致计算任务长时间占据 worker 线程，阻塞了 I/O 任务的快速响应。

二、自定义 I/O 调度器：分离 I/O 线程与计算线程

改造的第一步是将 I/O 处理与计算处理在运行时层面彻底分离。Rust 的 tokio::runtime::Builder 支持多运行时实例，可以在同一个进程中创建两个独立的 Tokio runtime，各自拥有独立的 worker 线程池：

use tokio::runtime::{Builder, Runtime};
use std::sync::Arc;

/// 异构调度器 —— 将 I/O 任务和计算任务分离到不同的 Tokio 运行时
pub struct HeteroScheduler {
// I/O 专用运行时：少量线程，高优先级，只处理 epoll 事件和轻量逻辑
io_runtime: Runtime,
// 计算专用运行时：大量线程，处理序列化/压缩/加密等耗时操作
cpu_runtime: Runtime,
// 两个运行时之间的任务传递通道
compute_tx: flume::Sender<ComputeTask>,
}

impl HeteroScheduler {
pub fn new(io_threads: usize, cpu_threads: usize) -> Self {
// I/O 运行时：2 个 worker 线程，固定绑定到低编号 CPU 核
let io_runtime = Builder::new_multi_thread()
.worker_threads(io_threads)
.thread_name("io-worker")
// 关键：启用 I/O 驱动和定时器，但不承担计算压力
.enable_io()
.enable_time()
.build()
.expect("Failed to build I/O runtime");

// 计算运行时：CPU 核数-2 个 worker 线程，用于处理耗时任务
let cpu_runtime = Builder::new_multi_thread()
.worker_threads(cpu_threads)
.thread_name("cpu-worker")
// 计算运行时不需要 I/O 驱动，避免不必要的 epoll 开销
.build()
.expect("Failed to build CPU runtime");

// 使用 flume 的无界通道传递计算任务
// flume 的 SPSC 模式比 tokio::mpsc 在高并发场景下延迟低约 40%
let (compute_tx, compute_rx) = flume::unbounded::<ComputeTask>();

// 在计算运行时中启动消费者
cpu_runtime.spawn(async move {
while let Ok(task) = compute_rx.recv_async().await {
// 每个计算任务在独立的 spawn_blocking 中执行
// 避免阻塞计算运行时中的其他异步任务
tokio::task::spawn_blocking(move || {
(task.handler)(task.payload);
})
.await
.ok();
}
});

Self { io_runtime, cpu_runtime, compute_tx }
}

/// 将计算任务从 I/O 运行时投递到计算运行时
pub fn offload_compute(&self, task: ComputeTask) {
// 非阻塞投递，I/O 线程立即返回处理下一个事件
if self.compute_tx.send(task).is_err() {
tracing::error!("Compute channel closed, task dropped");
}
}
}

三、CPU 亲和性与无锁数据结构的深度优化

在多 NUMA 节点的机器上，将 I/O worker 线程绑定到与网卡相同的 NUMA 节点，可以消除跨 NUMA 访问的内存延迟（约 40~80ns 额外开销）：

/// CPU 亲和性设置 —— 将线程绑定到指定 NUMA 节点的 CPU 核
fn set_cpu_affinity(core_ids: &[usize]) {
let mut cpuset = nix::sched::CpuSet::new();
for &core_id in core_ids {
cpuset.set(core_id).expect("Failed to set CPU affinity");
}
// 将当前线程的 CPU 亲和性设置为指定核
nix::sched::sched_setaffinity(
nix::unistd::Pid::from_raw(0), // 0 代表当前线程
&cpuset,
)
.expect("Failed to set CPU affinity");
}

消息分发中的关键路径——目标路由表查询——从 RwLock<HashMap> 切换为无锁的 dashmap：

use dashmap::DashMap;
use std::sync::Arc;

/// 无锁路由表 —— 用 DashMap 替代 RwLock<HashMap>
/// DashMap 内部使用分片锁，写操作只锁定目标分片，
/// 读操作大部分情况下不需要获取锁
pub struct RouteTable {
// topic -> Vec<channel_sender>
subscribers: Arc<DashMap<String, Vec<flume::Sender<Message>>>>,
}

impl RouteTable {
/// 路由分发：根据消息 topic 将消息发送给所有订阅者
/// 这个函数在热路径上，每秒调用 20 万次，锁开销必须为零
pub fn dispatch(&self, topic: &str, msg: Message) -> usize {
// DashMap 的 get 是无锁读取（仅在扩容时有短暂阻塞）
let count = if let Some(subscribers) = self.subscribers.get(topic) {
let mut sent = 0;
// 并发发送给所有订阅者，不需要持有锁
for tx in subscribers.value() {
if tx.send(msg.clone()).is_ok() {
sent += 1;
}
}
sent
} else {
0
};
count
}
}

四、内存分配热点与 Arena Allocator

火焰图的另一个发现是 jemalloc 的 malloc 调用占用了约 18% 的 CPU 时间——每秒 20 万条消息的分配和释放产生了大量的堆操作。引入 Arena Allocator（使用 bumpalo crate）将短生命周期对象（消息体、路由中间结果）分配在 Bump Allocator 中，批量释放：

use bumpalo::Bump;

/// 批量消息处理 —— 使用 Arena 减少 malloc/free 频次
pub fn process_message_batch(
batch: &[RawMessage],
route_table: &RouteTable,
) {
// 创建 4KB 初始容量的 Arena
// 每批消息处理完后统一释放，无需逐个 free
let arena = Bump::with_capacity(4096);

for raw_msg in batch {
// 在 Arena 中分配消息对象，生命周期与 arena 绑定
let msg = arena.alloc(Message::from_raw(raw_msg));
route_table.dispatch(&msg.topic, msg.clone());
}
// arena 离开作用域时自动释放所有分配，O(1) 复杂度
}

最终优化后的压测结果：

五、总结

Rust 异步引擎的深度优化经验：

适用边界：异构调度方案适用于 QPS > 5 万且计算占比 > 30% 的场景。对于纯 I/O 或纯计算场景，默认 Tokio 调度器已足够。