Async Rust với Tokio: async/await, task và channel

1. Tổng quan

Rust có sẵn cú pháp async/await nhưng không kèm runtime để chạy các Future. Tokio là runtime phổ biến nhất cho việc này. Nó cung cấp scheduler, timer, I/O bất đồng bộ và các công cụ đồng bộ hóa.

Bài viết đi từ hàm async đầu tiên tới các thao tác thường dùng: spawn task, chạy song song bằng join!, đặt timeout bằng select!, chia sẻ trạng thái giữa các task, gửi dữ liệu qua channel và xử lý code blocking. Phần cuối nói về cách viết test và những lỗi hay gặp.

2. Cài đặt

Bạn thêm Tokio vào Cargo.toml:

[dependencies]
tokio = { version = "1", features = ["full"] }

[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }

Feature full bật toàn bộ module: runtime đa luồng, time, sync, net, macros… Cách này tiện khi học. Ở dự án thật, bạn nên chỉ bật những feature thực sự dùng để giảm thời gian biên dịch, ví dụ ["rt-multi-thread", "macros", "time", "sync"]. Feature test-util cho phép điều khiển thời gian trong test. Ta sẽ dùng nó ở mục 8.

3. Hàm async đầu tiên

3.1. #[tokio::main] và .await

use std::time::Duration;
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100)).await; // giả lập gọi database
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let order = fetch_order(42).await;
    println!("{order}");
}

Giải thích:

• async fn fetch_order không trả về String ngay. Nó trả về một impl Future<Output = String>.
• tokio::time::sleep là phiên bản async của sleep. Trong lúc chờ, thread được trả lại cho runtime để chạy việc khác.
• #[tokio::main] là macro. Nó tạo runtime đa luồng rồi chạy main bên trong runtime đó bằng block_on.

Order #42

3.2. Future là lazy

Đây là điểm khác biệt lớn so với Promise trong JavaScript: future trong Rust không làm gì cả cho tới khi bị .await hoặc được poll.

async fn load_customer(id: u32) -> String {
    println!("Đang tải customer {id}");
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let fut = load_customer(7); // chưa chạy dòng nào trong hàm
    println!("Trước khi await");
    let customer = fut.await;
    println!("{customer}");
}

Trước khi await
Đang tải customer 7
Customer 7

Dòng “Đang tải customer 7” chỉ xuất hiện sau khi gọi .await. Nếu bạn quên .await, hàm sẽ không bao giờ chạy. Compiler sẽ cảnh báo unused implementer of Future that must be used, nên bạn đừng bỏ qua cảnh báo này.

4. Spawn task

.await chạy các future tuần tự. Muốn chúng chạy đồng thời, bạn dùng tokio::spawn để tạo task. Task là một “green thread” nhẹ do Tokio quản lý.

use std::time::{Duration, Instant};
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100 * id as u64)).await;
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut handles = Vec::new();

    for id in 1..=3 {
        handles.push(tokio::spawn(async move { fetch_order(id).await }));
    }

    for handle in handles {
        let order = handle.await.expect("task bị panic");
        println!("{order}");
    }
    println!("Tổng thời gian: ~{} ms", start.elapsed().as_millis());
}

Order #1
Order #2
Order #3
Tổng thời gian: ~300 ms

Nếu chạy tuần tự, tổng thời gian sẽ là 100 + 200 + 300 = 600 ms. Khi spawn, ba task chạy đồng thời nên tổng chỉ khoảng 300 ms.

Một số điểm cần nắm:

• tokio::spawn trả về JoinHandle<T>. Khi .await handle, bạn nhận về Result<T, JoinError>. Kết quả là Err nếu task bị panic hoặc bị hủy.
• Future truyền vào phải thỏa Send + 'static, vì task có thể chạy trên thread khác và sống lâu hơn scope hiện tại. Đó là lý do cần async move.
• Task bắt đầu chạy ngay khi được spawn, không cần chờ .await handle.

Lưu ý: Drop một JoinHandle không hủy task. Task vẫn tiếp tục chạy ở chế độ detached. Muốn hủy, bạn gọi handle.abort().

Khi số lượng task thay đổi, bạn có thể dùng tokio::task::JoinSet. Nó cho phép lấy kết quả theo thứ tự task hoàn thành bằng join_set.join_next().await.

5. Chạy đồng thời trong cùng một task: join!, select!, timeout

5.1. join! và try_join!

Không phải lúc nào cũng cần spawn. tokio::join! poll nhiều future đồng thời trong cùng task và chờ tất cả hoàn thành:

use std::time::{Duration, Instant};
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

async fn fetch_customer(id: u32) -> String {
    sleep(Duration::from_millis(150)).await;
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let (order, customer) = tokio::join!(fetch_order(1), fetch_customer(7));
    println!("{order} - {customer} ({} ms)", start.elapsed().as_millis());
}

Order #1 - Customer 7 (200 ms)

Tổng thời gian bằng future lâu nhất, khoảng 200 ms. Các future không cần 'static, nên chúng được phép mượn biến cục bộ.

Khi các hàm trả về Result, bạn dùng tokio::try_join!. Macro này dừng ngay ở lỗi đầu tiên và drop các future còn lại:

let (order, customer) = tokio::try_join!(fetch_order_checked(1), fetch_customer_checked(7))?;

Ở đây fetch_order_checked và fetch_customer_checked là các hàm async trả về Result<String, AppError>.

5.2. timeout

use tokio::time::{timeout, Duration};

match timeout(Duration::from_millis(50), fetch_order(1)).await {
    Ok(order) => println!("{order}"),
    Err(_) => println!("Quá thời gian chờ"),
}

Quá thời gian chờ

fetch_order cần 200 ms, vượt giới hạn 50 ms. Khi đó timeout trả về Err(Elapsed), và future bên trong bị drop, tức là bị hủy.

5.3. select!

select! chờ nhánh đầu tiên hoàn thành rồi hủy các nhánh còn lại:

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(());
    });

    tokio::select! {
        order = fetch_order(1) => println!("Xử lý xong {order}"),
        _ = shutdown_rx => println!("Nhận tín hiệu shutdown, dừng xử lý"),
    }
}

Nhận tín hiệu shutdown, dừng xử lý

Tín hiệu shutdown đến sau 100 ms, sớm hơn fetch_order (200 ms), nên nhánh thứ hai thắng.

Lưu ý: Hủy trong Rust nghĩa là future bị drop tại điểm .await gần nhất. Nếu nhánh bị hủy đang làm dở một thao tác không nguyên tử, chẳng hạn đã đọc một nửa message từ socket, dữ liệu có thể bị mất. Bạn nên kiểm tra mục “cancel safety” trong tài liệu của từng hàm trước khi đưa nó vào select! trong vòng lặp.

6. Chia sẻ dữ liệu giữa các task

6.1. Arc<Mutex<T>>

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

#[tokio::main]
async fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("laptop", 10u32)])));
    let mut handles = Vec::new();

    for _ in 0..5 {
        let inventory = Arc::clone(&inventory);
        handles.push(tokio::spawn(async move {
            let mut stock = inventory.lock().unwrap();
            *stock.get_mut("laptop").unwrap() -= 1;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!("Còn lại: {}", inventory.lock().unwrap()["laptop"]);
}

Còn lại: 5

Ở đây ta dùng std::sync::Mutex, không phải tokio::sync::Mutex. Đây là lựa chọn được khuyến nghị khi critical section ngắn và không có .await bên trong, vì mutex chuẩn nhanh hơn.

Bạn chỉ cần tokio::sync::Mutex khi phải giữ lock qua một điểm .await, ví dụ khi giữ một kết nối dùng chung trong lúc chờ I/O.

6.2. Channel mpsc

Thay vì chia sẻ bộ nhớ, bạn có thể truyền message giữa các task qua channel:

use tokio::sync::mpsc;

#[derive(Debug)]
struct Order {
    id: u32,
    amount: u64,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Order>(32);

    for id in 1..=3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(Order { id, amount: id as u64 * 100 }).await.unwrap();
        });
    }
    drop(tx); // đóng sender gốc để vòng lặp bên dưới kết thúc được

    let mut total = 0;
    while let Some(order) = rx.recv().await {
        println!("Nhận {order:?}");
        total += order.amount;
    }
    println!("Tổng: {total}");
}

Nhận Order { id: 1, amount: 100 }
Nhận Order { id: 3, amount: 300 }
Nhận Order { id: 2, amount: 200 }
Tổng: 600

Thứ tự nhận có thể khác nhau giữa các lần chạy. Tham số 32 là dung lượng buffer. Khi buffer đầy, send sẽ chờ, tạo ra cơ chế back-pressure tự nhiên. recv() trả về None khi mọi sender đã bị drop. Nếu quên drop(tx), chương trình sẽ treo mãi ở vòng while.

Ngoài mpsc, Tokio còn có:

• oneshot: gửi một giá trị duy nhất.
• broadcast: nhiều consumer, mỗi consumer nhận mọi message.
• watch: consumer chỉ quan tâm giá trị mới nhất.

7. Xử lý code blocking

Tokio dùng cooperative scheduling. Task chỉ nhường thread tại các điểm .await. Một thao tác CPU nặng hoặc I/O đồng bộ sẽ chiếm luôn worker thread và làm các task khác bị đói.

fn hash_password(password: &str) -> String {
    // giả lập thao tác tốn CPU, ví dụ bcrypt/argon2
    std::thread::sleep(std::time::Duration::from_millis(500));
    format!("hashed:{password}")
}

#[tokio::main]
async fn main() {
    let password = String::from("s3cret");
    let hashed = tokio::task::spawn_blocking(move || hash_password(&password))
        .await
        .expect("task blocking bị panic");
    println!("{hashed}");
}

hashed:s3cret

spawn_blocking đẩy closure sang một thread pool riêng dành cho việc blocking. Nhờ vậy các worker thread của runtime vẫn rảnh để xử lý request khác.

Lưu ý: Không gọi std::thread::sleep, std::fs::read, hoặc client HTTP đồng bộ trực tiếp trong hàm async. Hãy dùng phiên bản async (tokio::time::sleep, tokio::fs) hoặc bọc chúng trong spawn_blocking. Với tính toán song song khối lượng lớn, Rayon thường phù hợp hơn.

8. Test code async

Macro #[tokio::test] tạo một runtime cho mỗi test. Mặc định đây là runtime đơn luồng.

use std::time::Duration;
use tokio::time::{sleep, timeout};

pub async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn fetch_order_returns_formatted_id() {
        assert_eq!(fetch_order(42).await, "Order #42");
    }

    #[tokio::test(start_paused = true)]
    async fn timeout_fires_when_order_service_is_slow() {
        let result = timeout(Duration::from_millis(50), fetch_order(1)).await;
        assert!(result.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn order_arrives_within_generous_timeout() {
        let result = timeout(Duration::from_secs(1), fetch_order(1)).await;
        assert_eq!(result.unwrap(), "Order #1");
    }
}

cargo test

running 3 tests
test tests::fetch_order_returns_formatted_id ... ok
test tests::timeout_fires_when_order_service_is_slow ... ok
test tests::order_arrives_within_generous_timeout ... ok

start_paused = true (cần feature test-util) đóng băng đồng hồ của Tokio. Khi mọi task đều đang chờ timer, runtime tự nhảy thời gian tới mốc kế tiếp. Nhờ vậy test về timeout chạy gần như tức thì và cho kết quả ổn định, không phụ thuộc tốc độ máy CI.

Nếu cần kiểm tra hành vi trên nhiều thread, bạn dùng #[tokio::test(flavor = "multi_thread", worker_threads = 2)].

9. Lỗi hay gặp

Giữ std::sync::MutexGuard qua .await. Guard này không phải Send, nên future cũng không còn Send. Khi đó tokio::spawn báo lỗi biên dịch future cannot be sent between threads safely. Cách sửa là thu hẹp scope của guard bằng một block { ... } trước khi .await, hoặc chuyển sang tokio::sync::Mutex.

Tạo runtime lồng nhau. Gọi Runtime::new().block_on(...) bên trong code đã chạy trên Tokio sẽ gây panic với thông báo Cannot start a runtime from within a runtime. Trong ngữ cảnh async, bạn chỉ cần .await.

Dùng async khi không cần. Một CLI tool đọc vài file, hay một job thuần tính toán, không được lợi gì từ Tokio. Bạn chỉ phải trả thêm thời gian biên dịch và độ phức tạp. Async phát huy tác dụng khi có nhiều thao tác I/O đồng thời: web server, crawler, proxy, client gọi nhiều service.

Spawn không giới hạn. Task rẻ nhưng không miễn phí. Spawn một task cho mỗi item trong danh sách một triệu phần tử có thể làm cạn tài nguyên phía downstream, chẳng hạn connection pool của database. Hãy giới hạn mức đồng thời bằng tokio::sync::Semaphore hoặc dùng buffer_unordered trong crate futures.

Bỏ qua JoinError. Nếu .await handle mà dùng let _ =, panic trong task sẽ bị nuốt mất. Bạn nên log lỗi hoặc propagate nó lên.

10. Kết luận

Bài viết đã đi qua các phần cốt lõi của async Rust với Tokio:

• Future là lazy và chỉ chạy khi được .await.
• tokio::spawn tạo task chạy đồng thời. Future truyền vào cần Send + 'static.
• join! và try_join! chạy nhiều future trong cùng task. select! và timeout xử lý việc chạy đua và hủy.
• Dữ liệu dùng chung có thể chia sẻ qua Arc<Mutex<T>>, hoặc truyền qua channel như mpsc và oneshot.
• Code blocking cần đưa vào spawn_blocking.
• #[tokio::test(start_paused = true)] giúp test về thời gian chạy nhanh và ổn định.

Khi đã nắm vững các phần này, bạn có thể tiếp tục với I/O mạng qua tokio::net, hoặc xây dựng web service trên nền Tokio bằng các framework như Axum.