Async/Await trong Rust

Hãy tưởng tượng một nhà hàng. Ở trang Lập trình đồng thời, bạn đã học về luồng (thread). Điều đó giống như thuê một người phục vụ cho mỗi bàn. Cách đó chạy được, nhưng người phục vụ thì tốn kém, và phần lớn thời gian họ chỉ đứng chờ nhà bếp.

Bất đồng bộ (async) giống như có một người phục vụ nhận nhiều đơn. Người phục vụ nhận đơn ở bàn 1, đưa cho nhà bếp, và trong lúc món ăn đang nấu thì sang bàn 2. Không ai phải đứng không. Đó là ý tưởng đằng sau async và .await trong Rust: trong lúc một tác vụ đang chờ (mạng, một tệp, một bộ hẹn giờ), một tác vụ khác được chạy.

Async tỏa sáng khi chương trình của bạn mất nhiều thời gian để chờ: web server, client gọi API, ứng dụng chat, bot. Trang này dùng tokio, runtime async phổ biến nhất cho Rust.

Cài đặt tokio

Thư viện chuẩn của Rust cho bạn từ khóa async và .await, nhưng không có “động cơ” thật sự chạy code async. Động cơ đó gọi là runtime, và tokio là cái mà hầu hết các dự án dùng. Thêm nó vào một dự án mới:

cargo new async_demo
cd async_demo
cargo add tokio --features full

Lệnh đó thêm dòng này vào Cargo.toml của bạn:

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

Feature full bật mọi thứ (bộ hẹn giờ, task, mạng và nhiều hơn nữa). Đó là lựa chọn dễ nhất khi bạn đang học. Các nút “Tự chạy thử” trên trang này cũng dùng được: Rust Playground có sẵn tokio.

Chương trình async đầu tiên của bạn

use std::time::Duration;

async fn make_coffee() -> String {
    tokio::time::sleep(Duration::from_millis(20)).await;
    String::from("cà phê")
}

#[tokio::main]
async fn main() {
    println!("Đang gọi món...");
    let drink = make_coffee().await;
    println!("Đồ uống của bạn đây: {}", drink);
}

Kết quả:

Đang gọi món...
Đồ uống của bạn đây: cà phê

▶ Tự chạy thử

Có ba điều mới ở đây:

async fn tạo ra một hàm có thể tạm dừng giữa chừng rồi chạy tiếp sau.
.await nghĩa là “chờ cái này xong, và trong lúc chờ thì để các tác vụ khác chạy”. Bạn chỉ dùng được .await bên trong một hàm async.
#[tokio::main] khởi động runtime tokio và chạy async fn main của bạn trên đó. Không có nó, main không thể là async.

tokio::time::sleep là phiên bản async của “chờ một chút”. Nó chỉ tạm dừng tác vụ này, không phải cả luồng.

Future là lười

Đây là phần làm hầu hết mọi người bất ngờ. Gọi một async fn không chạy nó. Nó trả lại cho bạn một future, một giá trị mang nghĩa “công việc sẽ diễn ra sau”. Không có gì xảy ra cho tới khi bạn .await nó:

async fn say_hello() {
    println!("Xin chào từ bên trong future!");
}

#[tokio::main]
async fn main() {
    let future = say_hello(); // chưa có gì chạy cả
    println!("Đã tạo future, nhưng chưa bắt đầu");

    future.await; // giờ nó mới chạy
    println!("Xong");
}

Kết quả:

Đã tạo future, nhưng chưa bắt đầu
Xin chào từ bên trong future!
Xong

▶ Tự chạy thử

Hãy coi future như một tờ phiếu gọi món. Viết phiếu không làm món ăn chín. Đưa phiếu cho nhà bếp (.await) mới làm.

Chạy nhiều tác vụ cùng lúc với `join!`

Tới giờ, chúng ta chỉ chờ từng thứ một. Sức mạnh thật sự đến từ việc chờ nhiều thứ cùng lúc. tokio::join! chạy nhiều future cùng nhau và chờ tới khi tất cả xong. Nó trả kết quả của chúng cho bạn dưới dạng một tuple:

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

async fn fetch_user() -> String {
    sleep(Duration::from_millis(30)).await; // giả vờ đây là một lời gọi mạng
    String::from("Ana")
}

async fn fetch_order_count() -> u32 {
    sleep(Duration::from_millis(20)).await;
    3
}

#[tokio::main]
async fn main() {
    let (user, orders) = tokio::join!(fetch_user(), fetch_order_count());
    println!("{} có {} đơn hàng", user, orders);
}

Kết quả:

Ana có 3 đơn hàng

▶ Tự chạy thử

Nếu bạn .await lần lượt từng cái, tổng thời gian chờ sẽ khoảng 30 + 20 = 50ms. Với join!, cả hai lần chờ diễn ra cùng lúc, nên chỉ mất khoảng 30ms, tức thời gian của cái chậm nhất. Với lời gọi mạng thật, sự khác biệt đó cộng dồn rất nhanh.

Tạo tác vụ với `tokio::spawn`

join! rất tốt khi bạn có một số lượng future cố định. Khi bạn có một danh sách công việc, hãy dùng tokio::spawn. Nó giao một future cho runtime thành một task (tác vụ) riêng, bắt đầu chạy ngay, và trả lại cho bạn một JoinHandle. Sau đó, bạn .await handle để lấy kết quả:

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

async fn download(id: u64) -> String {
    sleep(Duration::from_millis(50 - id * 10)).await;
    format!("đã tải xong tệp {}", id)
}

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

    for id in 1..=3 {
        let handle = tokio::spawn(download(id)); // bắt đầu chạy ngay bây giờ
        handles.push(handle);
    }

    for handle in handles {
        let result = handle.await.unwrap();
        println!("{}", result);
    }
}

Kết quả:

đã tải xong tệp 1
đã tải xong tệp 2
đã tải xong tệp 3

▶ Tự chạy thử

Nhìn kỹ nhé: tệp 3 có thời gian ngủ ngắn nhất, nên thật ra nó xong trước. Nhưng chúng ta in kết quả theo thứ tự .await các handle, nên kết quả lúc nào cũng là 1, 2, 3. Cả ba lượt tải vẫn chạy cùng lúc.

Vì sao có .unwrap()? .await một JoinHandle trả về một Result, vì task có thể đã panic. Nếu vậy, bạn nhận được một Err thay vì làm sập cả chương trình. Điều này trông rất giống thread::spawn và .join() ở trang Lập trình đồng thời, và đó là có chủ ý.

Lỗi thường gặp

Quên .await. Vì future là lười, quên .await nghĩa là công việc không bao giờ diễn ra. Nếu bạn dùng giá trị đó, trình biên dịch sẽ chặn bạn lại:

async fn get_number() -> u32 {
    5
}

#[tokio::main]
async fn main() {
    let n = get_number(); // quên .await
    println!("{}", n + 1);
}

error[E0369]: cannot add `{integer}` to `impl Future<Output = u32>`
 --> src/main.rs:8:22
  |
8 |     println!("{}", n + 1);
  |                    - ^ - {integer}
  |                    |
  |                    impl Future<Output = u32>

Thông báo (bằng tiếng Anh) nói rằng n là một future của u32, không phải một u32. Cách sửa là let n = get_number().await;. Nếu hàm không trả về gì và bạn chỉ gọi nó, như save_file();, bạn chỉ nhận được một cảnh báo, nên hãy đọc kỹ các cảnh báo:

warning: unused implementer of `Future` that must be used
  |
7 |     save_file();
  |     ^^^^^^^^^^^
  |
  = note: futures do nothing unless you `.await` or poll them

Ghi chú cuối nghĩa là: future không làm gì cả trừ khi bạn .await hoặc poll nó.

Dùng std::thread::sleep trong code async. Lỗi này vẫn biên dịch bình thường, nên rất khó phát hiện:

async fn slow_task() {
    std::thread::sleep(std::time::Duration::from_secs(1)); // chặn cả luồng!
}

std::thread::sleep đóng băng toàn bộ luồng. Runtime không thể chuyển sang các tác vụ khác, nên mọi thứ trên luồng đó dừng lại, giống như một người phục vụ ngủ gật khi đang đứng ở một bàn. Trong code async, hãy luôn dùng tokio::time::sleep(...).await. Điều tương tự áp dụng cho các việc chậm, mang tính chặn khác như đọc tệp rất lớn bằng std::fs: tokio có phiên bản async (tokio::fs), hoặc bạn có thể dùng tokio::task::spawn_blocking.

Mong async làm việc tính toán nhanh hơn. Async giúp khi các tác vụ đang chờ. Nó không giúp gì khi các tác vụ đang bận, như thay đổi kích thước ảnh hay tính toán số liệu. Một người phục vụ không thể nấu nhanh hơn chỉ vì nhận nhiều đơn hơn. Với việc nặng về CPU, hãy dùng luồng (hoặc một crate như rayon).

Async hay luồng?

Dùng async khi…	Dùng luồng khi…
Phần lớn thời gian là chờ (mạng, cơ sở dữ liệu, bộ hẹn giờ)	Công việc là tính toán nặng
Bạn cần hàng nghìn tác vụ cùng lúc (như một web server)	Bạn chỉ cần vài việc chạy song song
Bạn đang dùng các crate async như reqwest hay axum	Bạn muốn tránh thêm một runtime vào dependency

Bạn không cần phải chọn chỉ một. Nhiều chương trình thực tế dùng async cho mạng và luồng cho việc nặng.

Cần các tác vụ gửi tin nhắn cho nhau? Tokio có channel riêng, tokio::sync::mpsc. Nó hoạt động giống bản chuẩn ở trang Channel, chỉ khác là bạn viết rx.recv().await, nên việc chờ tin nhắn không chặn luồng.

Người ta xây dựng gì với nó

Phần lớn hệ sinh thái async của Rust được xây trên tokio. Hai crate bạn sẽ gặp ở khắp nơi:

reqwest (tiếng Anh): một HTTP client để gọi web API. Bạn viết reqwest::get(url).await để tải một trang.
axum (tiếng Anh): một web framework để xây server và API, do nhóm tokio làm ra.

Cả hai đều đi rất hợp với serde_json để đọc và ghi JSON.

Luyện tập

1. Làm bữa sáng cùng một lúc

Viết ba hàm async: make_toast (ngủ 30ms), fry_eggs (ngủ 20ms) và brew_coffee (ngủ 10ms). Mỗi hàm trả về một String chứa tên món ăn. Chạy cả ba cùng lúc và in ra những món đã sẵn sàng. Gợi ý: tokio::join! nhận nhiều future và trả về một tuple chứa kết quả của chúng.

Xem lời giải

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

async fn make_toast() -> String {
    sleep(Duration::from_millis(30)).await;
    String::from("bánh mì nướng")
}

async fn fry_eggs() -> String {
    sleep(Duration::from_millis(20)).await;
    String::from("trứng")
}

async fn brew_coffee() -> String {
    sleep(Duration::from_millis(10)).await;
    String::from("cà phê")
}

#[tokio::main]
async fn main() {
    let (toast, eggs, coffee) = tokio::join!(make_toast(), fry_eggs(), brew_coffee());
    println!("Bữa sáng: {}, {} và {}", toast, eggs, coffee);
}

Kết quả:

Bữa sáng: bánh mì nướng, trứng và cà phê

▶ Tự chạy thử

Cà phê xong trước, nhưng tuple luôn trả về theo thứ tự bạn liệt kê các future. Cả bữa sáng mất khoảng 30ms (món chậm nhất), không phải 60ms.

2. Bình phương các số ở chế độ nền

Viết async fn square_later(n: u64) -> u64, hàm này ngủ 10 * n mili giây rồi trả về n * n. Trong main, tạo một task cho mỗi số từ 1 đến 4, rồi .await tất cả và in ra tổng. Gợi ý: đẩy từng handle của tokio::spawn(...) vào một Vec, rồi lặp qua các handle và cộng dồn handle.await.unwrap().

Xem lời giải

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

async fn square_later(n: u64) -> u64 {
    sleep(Duration::from_millis(10 * n)).await;
    n * n
}

#[tokio::main]
async fn main() {
    let mut handles = Vec::new();
    for n in 1..=4 {
        handles.push(tokio::spawn(square_later(n)));
    }

    let mut total = 0;
    for handle in handles {
        total += handle.await.unwrap();
    }
    println!("tổng = {total}");
}

Kết quả:

tổng = 30

▶ Tự chạy thử

Cả bốn task bắt đầu ngay và ngủ cùng lúc, nên việc này mất khoảng 40ms, không phải 10 + 20 + 30 + 40 = 100ms.

Tóm tắt

Async cho phép một luồng xử lý nhiều tác vụ bằng cách chuyển qua lại giữa chúng trong lúc chúng chờ.
Rust cần một runtime để chạy code async. Thêm tokio bằng cargo add tokio --features full và đánh dấu main bằng #[tokio::main].
Một async fn trả về một future. Future là lười: không có gì chạy cho tới khi bạn .await.
Dùng tokio::join! để chờ nhiều future cùng lúc, và tokio::spawn để khởi chạy task và nhận về một JoinHandle.
Đừng bao giờ dùng std::thread::sleep trong code async; hãy dùng tokio::time::sleep(...).await.
Async dành cho việc chờ, không dành cho việc nặng về CPU. Với việc đó, hãy dùng luồng.

Kiểm tra nhanh

1. Điều gì xảy ra khi bạn gọi một async fn mà không có .await?

2. #[tokio::main] làm gì?

3. Hai future ngủ 30ms và 20ms. tokio::join! trên cả hai mất khoảng bao lâu?

4. Vì sao std::thread::sleep gây vấn đề trong code async?

5. Async hợp nhất với công việc nào?

Điểm: 0 / 5

Đọc thêm

Fundamentals of Asynchronous Programming — The Rust Book (tiếng Anh): chương chính thức về future, async, .await và stream.
The tokio tutorial (tiếng Anh): hướng dẫn thực hành xây một server nhỏ giống Redis, đề cập tới task, trạng thái dùng chung và channel.
tokio trên docs.rs (tiếng Anh): tài liệu API cho tokio::spawn, tokio::time, tokio::sync và phần còn lại.
Asynchronous Programming in Rust (the async book) (tiếng Anh): cái nhìn sâu hơn về cách future và executor hoạt động bên dưới.

Async/Await trong Rust

Cài đặt tokio

Chương trình async đầu tiên của bạn

Future là lười

Chạy nhiều tác vụ cùng lúc với join!

Tạo tác vụ với tokio::spawn

Lỗi thường gặp

Async hay luồng?

Người ta xây dựng gì với nó

Luyện tập

1. Làm bữa sáng cùng một lúc

2. Bình phương các số ở chế độ nền

Tóm tắt

Kiểm tra nhanh

Đọc thêm

Chạy nhiều tác vụ cùng lúc với `join!`

Tạo tác vụ với `tokio::spawn`