Lập trình bất đồng bộ với Async và Tokio trong Rust

Chào mừng bạn đến với Phần 3: Advanced Systems & Production Rust. Khi xây dựng các hệ thống mạng, microservices hay web server phục vụ hàng triệu kết nối đồng thời, việc tạo mỗi luồng hệ điều hành (OS thread) cho mỗi kết nối sẽ sớm làm cạn kiệt tài nguyên bộ nhớ và CPU do chi phí context-switching quá lớn.

Rust giải quyết bài toán này thông qua mô hình Async/Await không tốn chi phí thời gian chạy (zero-cost async), kết hợp với runtime mạnh mẽ bậc nhất hệ sinh thái: Tokio.

graph TD
    A["Mô hình Async trong Rust"] --> B["Future Trait<br/>Lazy - chỉ chạy khi được poll"]
    A --> C["async / .await<br/>Cú pháp máy trạng thái (State Machine)"]
    A --> D["Tokio Runtime<br/>Reactor + Work-stealing Executor"]
    A --> E["tokio::spawn<br/>Green Tasks siêu nhẹ (~ vài trăm bytes)"]

Giải thích đơn giản (ELI5): Quán cà phê đông khách#

• Đa luồng truyền thống (OS Threads): Bạn thuê 100 nhân viên phục vụ. Mỗi nhân viên đứng đợi 1 khách duy nhất từ lúc gọi món, đợi pha chế đến khi khách uống xong. Rất tốn lương nhân viên (bộ nhớ RAM) và chật chội quán.
• Bất đồng bộ với Tokio (Async Tasks): Chỉ cần 4 nhân viên pha chế siêu tốc. Khi khách gọi món, nhân viên đưa cho khách một chiếc máy rung báo hiệu (Future). Trong lúc máy pha cà phê hoạt động, nhân viên phục vụ khách tiếp theo. Khi cà phê xong (Poll::Ready), máy rung lên và nhân viên giao hàng ngay lập tức!

1. Bản chất của Future trong Rust: Lazy by default#

Khác với JavaScript hay Python, các Future trong Rust có tính chất lười biếng (Lazy). Nếu bạn gọi một hàm async mà không .await nó, sẽ không có dòng mã nào thực sự chạy:

// Trait cốt lõi định nghĩa một Future
pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

pub enum Poll<T> {
    Ready(T),
    Pending,
}

sequenceDiagram
    autonumber
    actor Executor as Bộ điều phối Tokio Executor
    participant Task as Tác vụ Async Future
    participant Reactor as Bộ phản ứng IO Reactor
    
    Executor->>Task: poll(cx)
    Task->>Reactor: Đăng ký đợi I/O (ví dụ: Đọc Socket)
    Task-->>Executor: Trả về Poll::Pending (Nhường luồng)
    Note over Executor: Luồng Worker rảnh tay chạy tác vụ khác
    Reactor-->>Task: Dữ liệu đã về! Đánh thức tác vụ qua cx.waker()
    Executor->>Task: poll(cx) lần 2
    Task-->>Executor: Trả về Poll::Ready(data)

Khi bạn dùng .await, compiler sẽ chuyển đổi hàm async thành một Máy trạng thái (State Machine) lưu vết tiến độ thực thi mà không cấp phát heap nếu không cần thiết.

2. Tokio: Trái tim của hệ sinh thái Async Rust#

Rust cố tình không tích hợp runtime bất đồng bộ vào thư viện chuẩn để giữ ngôn ngữ nhỏ gọn. Tokio là runtime tiêu chuẩn công nghiệp được sử dụng rộng rãi nhất.

graph LR
    subgraph TokioRuntime ["Tokio Multi-Thread Runtime"]
        Reactor["I/O & Timer Reactor<br/>(epoll / kqueue / IOCP)"]
        W1["Worker Thread 1<br/>(Hàng đợi tác vụ)"]
        W2["Worker Thread 2<br/>(Hàng đợi tác vụ)"]
        W3["Worker Thread 3<br/>(Hàng đợi tác vụ)"]
    end
    
    W1 -.->|Work-stealing trộm việc| W2
    W2 -.->|Work-stealing trộm việc| W3
    Reactor -->|Đánh thức Task| W1

Để sử dụng Tokio, thêm vào Cargo.toml:

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

Chương trình Async đầu tiên với #[tokio::main]#

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

async fn fetch_user_data(user_id: u64) -> String {
    println!("Bắt đầu lấy dữ liệu người dùng {user_id}...");
    sleep(Duration::from_millis(500)).await; // Tạm dừng không chặn luồng OS
    format!("Dữ liệu người dùng #{user_id}")
}

#[tokio::main]
async fn main() {
    let result = fetch_user_data(42).await;
    println!("Kết quả: {result}");
}

3. Tạo tác vụ siêu nhẹ với tokio::spawn#

tokio::spawn tạo ra một task bất đồng bộ (green thread) được quản lý bởi bộ điều phối (executor) của Tokio. Mỗi task chỉ tốn vài trăm bytes bộ nhớ:

use tokio::task;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let mut handles = vec![];

    for i in 1..=5 {
        let handle = tokio::spawn(async move {
            sleep(Duration::from_millis(100 * i)).await;
            println!("Hoàn thành tác vụ #{i}");
            i * 10
        });
        handles.push(handle);
    }

    for handle in handles {
        let res = handle.await.unwrap();
        println!("Nhận giá trị: {res}");
    }
}

4. Chạy đồng thời với tokio::join! và tokio::select!#

tokio::join!: Đợi tất cả cùng hoàn thành song song#

async fn get_temperature() -> i32 { 25 }
async fn get_humidity() -> i32 { 60 }

#[tokio::main]
async fn main() {
    // Chạy đồng thời 2 futures trên cùng luồng hoặc worker pool
    let (temp, hum) = tokio::join!(get_temperature(), get_humidity());
    println!("Nhiệt độ: {temp}°C, Độ ẩm: {hum}%");
}

tokio::select!: Nhận kết quả từ future nào hoàn thành trước#

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

async fn slow_computation() -> &'static str {
    sleep(Duration::from_secs(2)).await;
    "Tính toán hoàn tất"
}

#[tokio::main]
async fn main() {
    tokio::select! {
        res = slow_computation() => {
            println!("Kết quả: {res}");
        }
        _ = sleep(Duration::from_secs(1)) => {
            println!("Hết giờ (Timeout)!");
        }
    }
}

Bảng so sánh: OS Threads vs Async Tasks (Tokio)#

Tóm tắt#

• Future trong Rust là các máy trạng thái lười biếng, chỉ tiến triển khi được poll thông qua .await.
• Tokio cung cấp executor đa luồng kiểu work-stealing, I/O driver và bộ định thời gian mạnh mẽ.
• Sử dụng tokio::spawn cho hàng vạn tác vụ đồng thời và tokio::select! / join! để phối hợp các luồng bất đồng bộ.

Tài liệu tham khảo#

• Asynchronous Programming in Rust (Async Book) ↗
• Tokio Official Documentation and Tutorial ↗
• The Rust Programming Language - Concurrency ↗