Rust 2026 经验谈 - 异步常见陷阱与调试

1
use std::cell::RefCell;
2

3
async fn captured_non_send() {
4
    let local: RefCell<i32> = RefCell::new(0); // RefCell 不是 Send
5
    // 即使不跨 await，local 被 Future 持有
6
    *local.borrow_mut() = 42;
7
    some_async_work().await;
8
}
9

10
// 修复：限制非 Send 变量的生命周期，使其不跨 await
11
async fn fixed_scope() {
12
    {
13
        let local: RefCell<i32> = RefCell::new(0);
14
        *local.borrow_mut() = 42;
15
        // local 在这里 drop，不持有到 await
16
    }
17
    some_async_work().await;
18
}

1
use std::sync::Mutex;
2

3
async fn mutex_guard_across_await(data: &Mutex<Vec<i32>>) {
4
    let mut guard = data.lock().unwrap();  // MutexGuard 不是 Send
5
    guard.push(1);
6
    some_async_work().await;               // guard 还活着！
7
    guard.push(2);
8
}
9

10
// 修复：缩小锁的生命周期
11
async fn mutex_guard_no_await(data: &Mutex<Vec<i32>>) {
12
    {
13
        let mut guard = data.lock().unwrap();
14
        guard.push(1);
15
    }  // guard 在 await 前 drop
16
    some_async_work().await;
17
    {
18
        let mut guard = data.lock().unwrap();
19
        guard.push(2);
20
    }
21
}

1
use std::sync::Arc;
2
use tokio::sync::RwLock;
3

4
struct Processor {
5
    buffer: Arc<RwLock<Vec<u8>>>,
6
}
7

8
impl Processor {
9
    async fn process(&self) {
10
        {
11
            let mut buf = self.buffer.write().await;
12
            buf.push(1);
13
        }
14
        some_async_work().await;
15
        {
16
            let mut buf = self.buffer.write().await;
17
            buf.push(2);
18
        }
19
    }
20
}

1
impl Processor {
2
    fn prepare(&mut self) -> u8 {
3
        self.buffer.push(1);
4
        self.buffer[0]
5
    }
6

7
    fn finalize(&mut self, val: u8) {
8
        self.buffer.push(val);
9
    }
10

11
    async fn process(&mut self) {
12
        let val = self.prepare();
13
        some_async_work().await;
14
        self.finalize(val);
15
    }
16
}

1
use tokio_util::task::TaskTracker;
2

3
async fn scoped_processing() {
4
    let tracker = TaskTracker::new();
5
    let data = vec![1, 2, 3];
6

7
    for item in &data {
8
        let item = *item;
9
        tracker.spawn(async move {
10
            process_item(item).await;
11
        });
12
    }
13

14
    tracker.close();
15
    tracker.wait().await;
16
}
17

18
async fn process_item(item: i32) {
19
    some_async_work().await;
20
    println!("处理: {item}");
21
}

1
use tokio::net::TcpStream;
2
use tokio::time::{self, Duration};
3

4
async fn dangerous_select(mut socket: TcpStream) {
5
    let data = b"hello world";
6

7
    loop {
8
        tokio::select! {
9
            // 危险！如果 timeout 分支先完成，write 被取消
10
            // 但可能已经写了部分数据，不知道写了多少
11
            result = socket.write(data) => {
12
                if let Ok(n) = result {
13
                    println!("写了 {n} 字节");
14
                }
15
            }
16
            _ = time::sleep(Duration::from_secs(1)) => {
17
                println!("超时");
18
            }
19
        }
20
    }
21
}

1
use tokio::sync::mpsc;
2

3
async fn cancel_safe_select() {
4
    let (tx, mut rx) = mpsc::channel::<i32>(100);
5

6
    tokio::spawn(async move {
7
        tx.send(1).await.unwrap();
8
    });
9

10
    loop {
11
        tokio::select! {
12
            val = rx.recv() => {
13
                // rx.recv() 是 cancel-safe 的！
14
                // 取消时不会丢失消息
15
                match val {
16
                    Some(v) => println!("收到: {v}"),
17
                    None => break,
18
                }
19
            }
20
            _ = time::sleep(Duration::from_secs(1)) => {
21
                println!("超时");
22
            }
23
        }
24
    }
25
}

1
async fn isolated_write(mut socket: TcpStream, data: &[u8]) {
2
    let (tx, rx) = tokio::sync::oneshot::channel();
3

4
    tokio::spawn(async move {
5
        let result = socket.write(data).await;
6
        let _ = tx.send(result);
7
    });
8

9
    tokio::select! {
10
        result = rx => {
11
            // 写入完成
12
        }
13
        _ = time::sleep(Duration::from_secs(5)) => {
14
            // 超时，但写入任务仍在运行（不被取消）
15
        }
16
    }
17
}

1
use std::sync::Mutex;
2
use std::sync::Arc;
3

4
// 推荐：std::sync::Mutex + 短临界区
5
async fn with_std_mutex() {
6
    let data = Arc::new(Mutex::new(vec![]));
7

8
    let data_clone = data.clone();
9
    tokio::spawn(async move {
10
        // 锁住 -> 做事 -> 解锁，不跨 await
11
        let mut guard = data_clone.lock().unwrap();
12
        guard.push(1);
13
        // guard 自动 drop，不持有到 await
14
        drop(guard);
15

16
        some_async_work().await;
17
    });
18
}

1
use tokio::sync::Mutex;
2

3
// 必须用 tokio::sync::Mutex 的场景：锁内含 await
4
async fn with_tokio_mutex() {
5
    let data = Arc::new(Mutex::new(vec![]));
6

7
    let data_clone = data.clone();
8
    tokio::spawn(async move {
9
        let mut guard = data_clone.lock().await;
10
        guard.push(1);
11
        // 必须在锁内 await（如数据库操作）
12
        some_async_work_with_guard(&mut guard).await;
13
        guard.push(2);
14
    });
15
}

1
use console_subscriber::ConsoleLayer;
2
use tokio::runtime::Builder;
3

4
fn main() {
5
    // 启用 tokio-console 支持
6
    console_subscriber::init();
7

8
    let rt = Builder::new_multi_thread()
9
        .enable_all()
10
        .build()
11
        .unwrap();
12

13
    rt.block_on(async {
14
        // 你的异步代码
15
        my_app().await;
16
    });
17
}