rust-tokio-style

name	rust-tokio-style
description	Tokio 异步运行时编码风格 Skill。蒸馏自 tokio-rs/tokio 源码、Carl Lerche 和 Alice Ryhl 的设计文档、Tokio Tutorial、官方 PR review 习惯以及 mini-redis 参考实现。 触发词：「Tokio 风格」「Rust 异步风格」「tokio style」「async Rust 惯用法」「Rust 运行时开发」。 适用：Tokio-based 服务开发、异步库封装、自定义 Future 实现、运行时集成、网络服务框架开发。

Tokio · 编码 DNA

"Don't block the async executor. Don't block the async executor. Don't block the async executor." — Tokio team motto "async fn is not magic. It's a state machine." — Alice Ryhl

角色定义

此 Skill 激活后，你写出的代码应该让 Tokio 核心 contributor 在 PR review 时感觉 「这代码对 async Rust 有真正的理解」，而不是「这是把同步代码机械翻译成 async 的」。

这意味着：不阻塞执行器、Pin 用法精确、错误类型显式设计、feature flags 细粒度。

---

命名 DNA

6 条直觉规则：

1. async fn 命名和同步版本对称，不加 async_ 前缀

   // ✅ tokio 的 API 镜像标准库，用法直觉
   use tokio::fs::File;
   let f = File::open("foo.txt").await?;
   
   use tokio::net::TcpListener;
   let listener = TcpListener::bind("127.0.0.1:8080").await?;
   
   // ❌ 画蛇添足
   async fn async_read_file() { ... }
   async fn fetch_data_async() { ... }
   

2. 错误类型：库用 thiserror，应用用 anyhow，不混用

   // ✅ 库代码 — 显式错误类型，调用方可以匹配
   #[derive(Debug, thiserror::Error)]
   pub enum AcceptError {
       #[error("io error: {0}")]
       Io(#[from] std::io::Error),
       #[error("connection limit reached: {0}")]
       LimitReached(usize),
   }
   
   // ✅ 应用代码 — anyhow::Result 省心
   async fn run() -> anyhow::Result<()> {
       let listener = TcpListener::bind("0.0.0.0:8080").await?;
       Ok(())
   }
   
   // ❌ 库代码用 Box<dyn Error>（调用方无法 match 错误类型）
   pub async fn accept(listener: &TcpListener) -> Result<TcpStream, Box<dyn std::error::Error>> { ... }
   

3. handle/sender/receiver 语义命名，不用泛型名

   // ✅ 清晰表达所有权和数据流向
   let (tx, rx) = mpsc::channel::<Message>(32);
   let handle = tokio::spawn(worker_task());
   let permit = semaphore.acquire().await?;
   
   // ❌ 语义模糊
   let (s, r) = mpsc::channel::<Message>(32);
   let h = tokio::spawn(worker_task());
   

4. spawn_blocking 的结果变量加 _result 后缀，强调需要 .await 两层

   // ✅ 两个 ? 提醒开发者：spawn 可能 panic，blocking 操作本身也可能出错
   let content = tokio::task::spawn_blocking(|| {
       std::fs::read_to_string("large_file.txt")
   })
   .await          // JoinError（task panic）
   .expect("task panicked")?;  // io::Error
   
   // ❌ 忘记处理 JoinError
   let content = tokio::task::spawn_blocking(|| {
       std::fs::read_to_string("large_file.txt").unwrap()  // panic 会变成 JoinError
   }).await.unwrap();
   

5. select! 分支变量用语义名，不用 _ 或泛型字母

   // ✅
   tokio::select! {
       result = socket.read(&mut buf) => {
           handle_read(result?)?;
       }
       _ = shutdown_rx.recv() => {
           info!("received shutdown signal");
           break;
       }
       _ = tokio::time::sleep(timeout) => {
           warn!("connection timed out");
           break;
       }
   }
   
   // ❌ 语义丢失
   tokio::select! {
       r = s.read(&mut b) => { ... }
       x = rx.recv() => { ... }
   }
   

6. Pin<Box<dyn Future>> 类型别名化，不内联

   // ✅ 清晰且可复用
   type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
   
   fn make_future() -> BoxFuture<'static, Result<(), Error>> {
       Box::pin(async { Ok(()) })
   }
   
   // ❌ 内联类型噪音
   fn make_future() -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>> {
       Box::pin(async { Ok(()) })
   }
   

---

结构偏好

async fn vs 手写 Future：明确边界

// ✅ 99% 场景用 async fn，让编译器生成状态机
pub async fn handle_connection(mut socket: TcpStream) -> Result<()> {
    let mut buf = BytesMut::with_capacity(4096);
    loop {
        let n = socket.read_buf(&mut buf).await?;
        if n == 0 { break; }
        socket.write_all(&buf).await?;
        buf.clear();
    }
    Ok(())
}

// ✅ 需要精确控制 Poll 语义时才手写 Future（如实现 Stream、Sink、自定义 waker）
use std::task::{Context, Poll};
use std::pin::Pin;
use std::future::Future;

struct Delay {
    when: std::time::Instant,
}

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if std::time::Instant::now() >= self.when {
            Poll::Ready(())
        } else {
            // 注册 waker，到时间后会被唤醒
            let waker = cx.waker().clone();
            let when = self.when;
            std::thread::spawn(move || {
                let now = std::time::Instant::now();
                if now < when { std::thread::sleep(when - now); }
                waker.wake();
            });
            Poll::Pending
        }
    }
}

// ❌ 滥用手写 Future 实现简单的异步逻辑（async fn 更清晰）

不阻塞执行器：CPU 密集和阻塞 IO 必须 spawn_blocking

// ✅ CPU 密集型任务放进 spawn_blocking
async fn process_image(path: PathBuf) -> Result<Vec<u8>> {
    let result = tokio::task::spawn_blocking(move || {
        // 图像处理是 CPU 密集，不能在 async 上下文直接运行
        image::open(&path)?.to_rgb8().to_vec()
    }).await??;
    Ok(result)
}

// ❌ 在 async fn 里直接调用同步阻塞 API
async fn bad_read(path: &str) -> Result<String> {
    // 阻塞整个 Tokio 线程，可能导致其他任务饥饿！
    Ok(std::fs::read_to_string(path)?)
}

Graceful Shutdown 标准模式：

// ✅ 用 CancellationToken 或 watch channel 广播 shutdown 信号
use tokio_util::sync::CancellationToken;

async fn run_server(token: CancellationToken) -> Result<()> {
    let listener = TcpListener::bind("0.0.0.0:8080").await?;

    loop {
        tokio::select! {
            result = listener.accept() => {
                let (socket, addr) = result?;
                let token = token.clone();
                tokio::spawn(handle_connection(socket, addr, token));
            }
            _ = token.cancelled() => {
                info!("server shutting down");
                break;
            }
        }
    }
    Ok(())
}

// 顶层：监听 Ctrl+C
#[tokio::main]
async fn main() -> Result<()> {
    let token = CancellationToken::new();
    let server_token = token.clone();

    let server = tokio::spawn(run_server(server_token));

    tokio::signal::ctrl_c().await?;
    token.cancel();
    server.await??;
    Ok(())
}

feature flags 细粒度设计（tokio 自身的标准）：

# ✅ Cargo.toml — 按功能模块拆分，用户按需引入
[features]
default = []
full = ["io-util", "io-std", "net", "sync", "time", "rt", "rt-multi-thread", "macros"]
io-util   = ["bytes"]
net       = ["io-util", "socket2", "mio/net"]
rt        = []
rt-multi-thread = ["rt", "num_cpus"]
macros    = ["tokio-macros"]
time      = []
sync      = []

# ❌ 一个 feature 包含所有功能（强制用户引入不需要的依赖）
[features]
default = ["full"]
full = ["dep:tokio", "dep:hyper", "dep:serde", "dep:sqlx"]

文档测试（doc test）是一等公民：

/// 从 channel 接收一个值，超时返回 None。
///
/// # Examples
///
/// ```
/// use tokio::sync::mpsc;
/// use tokio::time::{timeout, Duration};
///
/// #[tokio::test]
/// async fn example() {
///     let (tx, mut rx) = mpsc::channel(1);
///     tx.send(42).await.unwrap();
///
///     let result = timeout(Duration::from_millis(100), rx.recv()).await;
///     assert_eq!(result.unwrap(), Some(42));
/// }
/// ```
pub async fn recv_timeout<T>(
    rx: &mut mpsc::Receiver<T>,
    duration: Duration,
) -> Option<T> {
    tokio::time::timeout(duration, rx.recv()).await.ok().flatten()
}

---

注释哲学

公开 API 注释三要素：是什么、注意什么（Cancel Safety）、示例

/// 等待 TCP 连接到来。
///
/// # Cancel Safety
///
/// 此方法是 cancel safe 的。如果 `accept` 在返回前被 drop，
/// 不会有连接丢失 — 下次调用会重新等待。
///
/// # Errors
///
/// 当底层 OS 调用失败时返回 [`io::Error`]。
///
/// # Examples
///
/// ```no_run
/// use tokio::net::TcpListener;
///
/// #[tokio::main]
/// async fn main() -> std::io::Result<()> {
///     let listener = TcpListener::bind("127.0.0.1:8080").await?;
///
///     loop {
///         let (socket, addr) = listener.accept().await?;
///         println!("new connection from {}", addr);
///         tokio::spawn(async move {
///             // 处理连接...
///             drop(socket);
///         });
///     }
/// }
/// ```
pub async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> { ... }

Cancel Safety 必须在文档里明确声明：

// ✅ 明确说明 cancel safety
/// # Cancel Safety
/// This method is **not** cancel safe. If cancelled before completion,
/// partial data may have been read into `buf`. Use [`read_exact`] instead
/// if you need atomic reads.

/// # Cancel Safety
/// This method is cancel safe. The future can be dropped at any time and
/// re-polled without losing progress.

---

反模式（绝不这样写）

1. tokio::main 内嵌套 Runtime::new()（运行时内创建运行时）

   // ❌ 在 async 上下文创建新的 runtime 会 panic
   #[tokio::main]
   async fn main() {
       let rt = tokio::runtime::Runtime::new().unwrap();  // 嵌套运行时 panic！
       rt.block_on(async { do_work().await });
   }
   // ✅ 如需隔离，用 spawn_blocking 里创建独立 runtime
   tokio::task::spawn_blocking(|| {
       let rt = tokio::runtime::Runtime::new().unwrap();
       rt.block_on(async { do_work().await })
   }).await?;
   

2. std::sync::Mutex 跨 .await 持锁（死锁）

   // ❌ std Mutex 跨 await 点持锁，Future 不是 Send，且可能死锁
   async fn bad(data: Arc<std::sync::Mutex<Vec<i32>>>) {
       let mut lock = data.lock().unwrap();
       some_async_op().await;  // 持锁 await！
       lock.push(42);
   }
   // ✅ 用 tokio::sync::Mutex 或缩短锁范围
   async fn good(data: Arc<tokio::sync::Mutex<Vec<i32>>>) {
       let mut lock = data.lock().await;
       lock.push(42);
       // lock 在这里 drop，不跨 await
   }
   // ✅ 或者 std Mutex 但不跨 await
   async fn also_good(data: Arc<std::sync::Mutex<Vec<i32>>>) {
       {
           let mut lock = data.lock().unwrap();
           lock.push(42);
       }  // lock 在 await 前 drop
       some_async_op().await;
   }
   

3. 裸 unwrap()/expect() 在库代码里（应该传播错误）

   // ❌ 库代码 panic 让调用方无法恢复
   pub async fn read_config(path: &str) -> Config {
       let content = tokio::fs::read_to_string(path).await.unwrap();
       toml::from_str(&content).expect("invalid config")
   }
   // ✅
   pub async fn read_config(path: &str) -> Result<Config, ConfigError> {
       let content = tokio::fs::read_to_string(path).await
           .map_err(|e| ConfigError::Io(e))?;
       toml::from_str(&content).map_err(ConfigError::Parse)
   }
   

4. 不加 #[must_use] 的 Future 返回类型

   // ❌ 调用方可能忘记 .await
   pub fn send_message(msg: String) -> impl Future<Output = Result<()>> { ... }
   // ✅
   #[must_use = "futures do nothing unless you `.await` or poll them"]
   pub fn send_message(msg: String) -> impl Future<Output = Result<()>> { ... }
   

5. select! 里 await 不是 cancel-safe 的 Future

   // ❌ read_exact 不是 cancel safe — select 取消时可能丢数据
   tokio::select! {
       result = socket.read_exact(&mut buf) => { ... }
       _ = shutdown.recv() => { break; }
   }
   // ✅ 包装成可以 cancel 的形式或用 read（cancel safe）
   tokio::select! {
       result = socket.read(&mut buf) => { ... }  // read 是 cancel safe
       _ = shutdown.recv() => { break; }
   }
   

6. tokio::spawn 的任务不 .await handle（忽略 panic）

   // ❌ 任务 panic 了也不知道
   tokio::spawn(async { risky_operation().await });
   // ✅ 要么 await handle，要么用 JoinSet 管理生命周期
   let handle = tokio::spawn(async { risky_operation().await });
   match handle.await {
       Ok(Ok(())) => {},
       Ok(Err(e)) => error!("task error: {}", e),
       Err(join_err) => error!("task panicked: {}", join_err),
   }
   

7. 手动实现 Future 时忘记注册 Waker

   // ❌ 没有注册 waker，Poll::Pending 后永远不会被唤醒（死锁）
   fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
       if self.ready { Poll::Ready(()) } else { Poll::Pending }
   }
   // ✅ 必须在返回 Pending 前注册 waker
   fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
       if self.ready {
           Poll::Ready(())
       } else {
           self.waker.register(cx.waker());  // 注册 waker！
           Poll::Pending
       }
   }
   

---

校验测试

1. 阻塞检测：代码里有没有 std::thread::sleep、std::fs::*、std::net::* 直接在 async fn 里调用？全部应该用 spawn_blocking 包裹或换成 tokio 的异步版本。
2. Cancel Safety 声明：所有公开的 async fn 文档里有没有 # Cancel Safety 段落？在 select! 里使用的 future 一定要是 cancel safe 的。
3. 错误类型选择：这是库代码（thiserror + 显式枚举）还是应用代码（anyhow::Result）？两者混用了吗？

---

来源

• tokio-rs/tokio 源码（tokio/src/net、tokio/src/sync、tokio/src/runtime）
• Tokio Tutorial 官方教程
• Alice Ryhl：Actors with Tokio、Async: What is blocking?
• mini-redis — Tokio 官方参考实现
• Carl Lerche：Tokio 设计哲学系列博文
• tokio-rs/tokio-util CancellationToken 设计