Tokio 任务泄露检测用 tokio-console 和 metrics 找到僵尸任务这就是我今天要聊的Tokio 任务泄露。它不是传统意义上的内存泄露但后果一样严重——资源耗尽、响应变慢、最终 OOM。更坑的是Rust 编译器帮不了你因为从编译器的视角这些代码完全正确。今天我会分享我是怎么用 tokio-console 和 metrics 来检测和解决这类问题的。一、任务泄露是怎么发生的先写一个最常见的泄露场景一个 WebSocket 服务每个连接 spawn 一个读写任务。但因为一个隐蔽的逻辑 bug连接断开后任务没有退出。use tokio::net::{TcpListener, TcpStream}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::sync::mpsc; /// 最典型泄露场景WebSocket 连接处理 /// 每个连接 spawn 两个任务但退出条件有 bug async fn handle_connection(mut socket: TcpStream) { // mpsc channel读任务把收到的消息发到写任务 let (tx, mut rx) mpsc::channel::String(32); // ← 注意容量 32 let (mut read_half, mut write_half) socket.split(); // 读任务从 socket 读数据 let read_task tokio::spawn(async move { let mut buf vec![0u8; 1024]; loop { match read_half.read(mut buf).await { Ok(0) { // 连接关闭了对端发了 FIN println!(连接关闭读任务退出); break; } Ok(n) { let msg String::from_utf8_lossy(buf[..n]).to_string(); // BUG在这里如果 channel 满了send 会阻塞等待 // 但发送端永远不会被唤醒任务永远卡在这里 if tx.send(msg).await.is_err() { break; // ← 只有接收端关闭才会走这里 } } Err(e) { eprintln!(读错误: {}, e); break; } } } }); // 写任务把收到的消息写回 socket let write_task tokio::spawn(async move { while let Some(msg) rx.recv().await { if write_half.write_all(msg.as_bytes()).await.is_err() { break; } } println!(写任务退出); }); // 等待任一任务完成正常情况读任务退出后写任务也会退出 tokio::select! { _ read_task {} _ write_task {} } }表面看这段代码没问题但实际上有一个经典的死锁泄露读任务往 channel 发消息channel 容量只有 32写任务消费消息但如果写任务来不及消费比如网络慢channel 满了读任务卡在tx.send(msg).await上永远等不到 slot但写任务也在等读任务发消息——互相等待谁也没退出这就是一个僵尸任务——它没有崩runtime 认为它还在正常运行但实际上它什么有用的事也不做就是占着内存。二、用 tokio-console 可视化任务状态tokio-console 是 Tokio 官方出的一个实时监控工具能让你看到 runtime 里有多少任务在跑、每个任务在做什么、在哪个.await点卡住了。使用起来很简单三步走第一步在Cargo.toml加依赖[dependencies] tokio { version 1, features [full, tracing] } console-subscriber 0.4第二步初始化 console subscriber/// 在 main 函数最开头初始化 /// console_subscriber 会启动一个 gRPC 服务供 tokio-console 连接 #[tokio::main] async fn main() { // init() 必须在任何 tokio 任务 spawn 之前调用 console_subscriber::init(); // 正常启动你的应用... let listener TcpListener::bind(0.0.0.0:8080).await.unwrap(); loop { let (socket, _) listener.accept().await.unwrap(); tokio::spawn(handle_connection(socket)); } }第三步在另一个终端运行tokio-console# 安装 tokio-console只需要装一次 cargo install tokio-console # 运行 tokio-console启动后你会看到一个类似htop的 TUI 界面。左侧是任务列表每一项显示了任务 ID、名称、状态RUNNING / IDLE / BLOCKING、已运行时间、以及当前的 poll 次数。右侧是任务详情显示任务卡在哪个文件的哪一行。stateDiagram-v2 [*] -- spawned: tokio::spawn() state Tokio Runtime as Runtime { spawned -- poll_start: runtime 调度 state poll_start { [*] -- executing: poll() 开始 executing -- awaiting: .await 遇到未就绪的 Future awaiting -- waker_registered: 返回 Poll::Pending waker_registered -- [*]: 进等待队列 } waker_registered -- poll_start: IO 就绪, waker 唤醒 executing -- completed: 返回 Poll::Ready } completed -- [*]: 任务结束 note right of awaiting 如果 waker 永远不会被触发 → 任务永远停在 awaiting → 这就是【僵尸任务】 end notetokio-console 的威力在于你可以直观地看到某个任务运行了 30 秒了还在 IDLE 状态总 poll 次数只有 5 次说明它大部分时间在等可能是等着错误的 waker。正常任务应该是频繁地在 RUNNING 和 IDLE 之间切换。三、用 metrics 建立任务泄露的告警机制tokio-console 适合调试和手动排查但要防止线上出问题需要在代码里埋点用 metrics 做持续监控。use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use tokio::sync::Semaphore; /// 全局任务计数器 /// 用 AtomicU64 保证无锁的增删操作 #[derive(Clone)] struct TaskMetrics { /// 当前活跃任务数spawn 时 1任务退出时 -1 active_tasks: ArcAtomicU64, /// 历史累计创建的任务数 total_spawned: ArcAtomicU64, /// 历史累计退出的任务数 total_completed: ArcAtomicU64, } impl TaskMetrics { fn new() - Self { TaskMetrics { active_tasks: Arc::new(AtomicU64::new(0)), total_spawned: Arc::new(AtomicU64::new(0)), total_completed: Arc::new(AtomicU64::new(0)), } } fn spawn_guard(self) - TaskGuard { // 创建新任务时活跃数1累计1 self.active_tasks.fetch_add(1, Ordering::Relaxed); self.total_spawned.fetch_add(1, Ordering::Relaxed); TaskGuard { metrics: self.clone() } } fn snapshot(self) - TaskSnapshot { TaskSnapshot { active: self.active_tasks.load(Ordering::Relaxed), spawned: self.total_spawned.load(Ordering::Relaxed), completed: self.total_completed.load(Ordering::Relaxed), } } } /// RAII 守卫任务退出时自动减活跃数 struct TaskGuard { metrics: TaskMetrics, } impl Drop for TaskGuard { fn drop(mut self) { // 任务结束活跃数-1完成数1 self.metrics.active_tasks.fetch_sub(1, Ordering::Relaxed); self.metrics.total_completed.fetch_add(1, Ordering::Relaxed); } } /// 封装 spawn自动记录 metrics fn spawn_trackedF(metrics: TaskMetrics, name: str, future: F) where F: std::future::FutureOutput () Send static, { let guard metrics.spawn_guard(); let name name.to_string(); tokio::spawn(async move { // ⚠️ 关键把 guard move 进 async 块 // 这样任务退出时async 块结束guard 的 Drop 会被调用 let _guard guard; tracing::info!(task_name %name, 任务开始); future.await; tracing::info!(task_name %name, 任务退出); }); } /// 定期打印 metrics用于发现泄露趋势 async fn metrics_reporter(metrics: TaskMetrics) { let mut interval tokio::time::interval( std::time::Duration::from_secs(10) ); loop { interval.tick().await; let snap metrics.snapshot(); tracing::info!( active snap.active, spawned snap.spawned, completed snap.completed, delta snap.spawned.saturating_sub(snap.completed), 任务 metrics ); // 告警规则如果 spawned - completed 持续上涨 // 且 active_tasks 一直在增长 → 大概率有任务泄露 if snap.active 1000 { tracing::error!( active_tasks snap.active, ⚠️ 活跃任务数超过 1000可能存在任务泄露 ); } } }核心思路就是用 RAIITaskGuard在创建时1在Drop时-1。只要把 guard move 进tokio::spawn的闭包里任务退出时自动减。如果spawned和completed的差距越来越大就说明有任务没有正常退出。四、根因修复模式从发现到修复的三步法找到泄露后怎么修复我总结了三步法第一步确保取消安全Cancellation Safety。很多泄露是因为select!中一个分支被 drop 了但另一个分支没有收到取消信号。修复方式是给 channel 的发送端设置超时/// 修复版用 try_send 超时 替代 send async fn safe_sendT(tx: mpsc::SenderT, msg: T) - Result(), T { // try_send 不会阻塞满了直接返回 match tx.try_send(msg) { Ok(()) Ok(()), Err(mpsc::error::TrySendError::Full(msg)) { // channel 满了等 100ms 再试 tokio::time::sleep(std::time::Duration::from_millis(100)).await; // 再试一次还不行就丢弃 if tx.try_send(msg).is_err() { tracing::warn!(消息丢弃channel 持续满); } Ok(()) } Err(mpsc::error::TrySendError::Closed(_)) { // 接收端已关闭正常返回 Ok(()) } } }第二步join 相关联的任务。如果一个任务衍生出子任务应该在退出前 join 所有子任务而不是用完就忘/// 用 JoinSet 管理一组相关任务 /// 确保父任务退出前所有子任务都被等待 async fn managed_connection(socket: TcpStream) { let mut tasks tokio::task::JoinSet::new(); // spawn 进 JoinSet而不是裸 tokio::spawn tasks.spawn(read_loop(socket)); tasks.spawn(write_loop(socket)); tasks.spawn(heartbeat_timer()); // 等待任一任务出错 while let Some(result) tasks.join_next().await { match result { Ok(()) { tracing::info!(一个子任务正常退出); } Err(e) { tracing::error!(子任务 panic: {}, e); // 一个子任务出错 → 停止所有子任务 tasks.abort_all(); break; } } } // 确保所有子任务都已退出 while (tasks.join_next().await).is_some() {} tracing::info!(所有子任务已清理完毕); }第三步加上全局超时。不确定的死循环最危险给每个任务加一个保底超时/// 给任务加上兜底超时 async fn spawn_with_timeoutF(future: F, timeout_duration: std::time::Duration) where F: std::future::FutureOutput () Send static, { tokio::spawn(async move { match tokio::time::timeout(timeout_duration, future).await { Ok(()) { tracing::info!(任务正常完成); } Err(_elapsed) { tracing::error!( duration_secs timeout_duration.as_secs(), ⏰ 任务超时强制退出 ); // 超时后这里的 async 块结束任务被清理 } } }); }flowchart TD A[发现可疑活跃任务数持续上升] -- B[打开 tokio-console] B -- C{任务在哪个 .await 卡住了} C --|channel send 卡住| D[修复try_send 超时] C --|lock 没释放| E[修复缩小临界区] C --|无限循环| F[修复加退出条件] C --|子任务没 join| G[修复用 JoinSet] D -- H[加上 metrics 监控] E -- H F -- H G -- H H -- I[验证活跃数曲线平稳] I -- J[长期建立告警阈值]五、总结Tokio 任务泄露是 Rust 异步编程中最容易被忽视但又后果严重的问题。编译器和 borrow checker 帮不上忙因为没有借用错误——代码从所有权角度看完全合规只是逻辑上永不休眠。我的经验是预防优于排查。从一开始就养成三个习惯spawn的任务一定要有明确的退出条件——channel 关闭、超时、或显式的 cancellation token用 metrics 持续追踪活跃任务数——哪怕只是打日志也能看出趋势定期跑 tokio-console 看一眼——有时候数值上看不出问题但可视化会让你瞬间发现异常作为自学者经历过一次内存炸掉之后才重视起来的。希望这篇文章能帮你少踩一次这个坑。如果你也遇到过任务泄露欢迎在评论区分享你的排查经验我们下篇见