Channel trong Rust: truyền tin nhắn giữa các luồng

Ở trang Lập trình đồng thời, các luồng (thread) chia sẻ dữ liệu bằng cách khóa nó với Arc<Mutex<T>>. Cách đó chạy được, nhưng còn một cách khác thường đơn giản hơn: thay vì chia sẻ dữ liệu, các luồng gửi tin nhắn cho nhau.

Hãy hình dung channel như một đường ống có hai đầu. Một luồng thả giá trị vào đầu gửi (sender). Một luồng khác lấy chúng ra ở đầu nhận (receiver), theo đúng thứ tự. Không ai chia sẻ gì cả: mỗi giá trị di chuyển từ luồng này sang luồng kia. Cộng đồng Go có một câu nói mà Rust mượn ở đây: “Đừng giao tiếp bằng cách chia sẻ bộ nhớ; hãy chia sẻ bộ nhớ bằng cách giao tiếp.”

Channel đầu tiên của bạn

Channel của thư viện chuẩn nằm trong std::sync::mpsc. Cái tên này nghĩa là multiple producer, single consumer (nhiều producer, một consumer): nhiều đầu gửi, một đầu nhận.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("xin chào từ worker");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main nhận được: {received}");
}

Kết quả:

main nhận được: xin chào từ worker

▶ Tự chạy thử

Từng bước một:

mpsc::channel() trả về một cặp: một Sender và một Receiver. Theo thói quen, người ta đặt tên chúng là tx (transmit, truyền) và rx (receive, nhận).
move giao tx cho luồng mới, giống như cách bạn chuyển dữ liệu vào luồng ở trang Lập trình đồng thời.
tx.send(value) đặt một giá trị vào channel. Nó trả về một Result, vì nó sẽ thất bại nếu đầu nhận đã không còn.
rx.recv() chờ cho tới khi có giá trị đến. Nó chỉ trả về Err khi mọi đầu gửi đều đã không còn và channel đã rỗng.

Để ý rằng chúng ta không cần join(). recv() đã chờ tin nhắn rồi, nên main không thể kết thúc quá sớm.

Nhận nhiều giá trị

Một Receiver hoạt động như một iterator. Vòng for chạy qua nó sẽ gọi recv() giúp bạn liên tục, và vòng lặp tự kết thúc khi mọi đầu gửi đã bị drop:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // tx bị drop ở đây, khi luồng kết thúc
    });

    for value in rx {
        println!("nhận {value}");
    }
    println!("channel đã đóng, xong hết");
}

Kết quả:

nhận 10
nhận 20
nhận 30
channel đã đóng, xong hết

▶ Tự chạy thử

Ở đây chỉ có một đầu gửi, nên các giá trị đến đúng theo thứ tự đã gửi. Quy tắc “vòng lặp dừng khi các đầu gửi không còn” rất quan trọng. Hãy nhớ nó cho phần tiếp theo.

Nhiều producer: clone đầu gửi

Để nhiều luồng cùng gửi được, hãy cho mỗi luồng một bản sao đầu gửi riêng bằng tx.clone(). Mọi bản sao đều đổ vào cùng một channel:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("worker {id} báo cáo")).unwrap();
        });
    }
    drop(tx); // drop bản gốc, nếu không vòng lặp bên dưới sẽ không bao giờ dừng

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

Kết quả:

worker 1 báo cáo
worker 2 báo cáo
worker 3 báo cáo

▶ Tự chạy thử

Hai điều cần để ý:

drop(tx) rất quan trọng. Mỗi luồng drop bản clone của nó khi chạy xong, nhưng main vẫn giữ bản gốc. Chừng nào còn một đầu gửi, đầu nhận vẫn tiếp tục chờ, và chương trình sẽ treo mãi mãi. Drop nó đi tức là nói “main sẽ không gửi gì nữa”.
Các luồng xong theo thứ tự bất kỳ. Tin nhắn có thể đến theo thứ tự worker 2, worker 1, worker 3. Chúng ta gom chúng lại và sắp xếp để kết quả in ra lần nào cũng giống nhau. Hãy bỏ sort() và chạy vài lần để thấy thứ tự thay đổi.

Không chờ mãi: `try_recv` và `recv_timeout`

recv() chặn lại cho tới khi có thứ gì đó đến. Đôi khi bạn không muốn chờ. Có hai “người anh em” giúp được:

try_recv() trả về ngay lập tức: một giá trị, Empty (chưa có gì), hoặc Disconnected (mọi đầu gửi đã không còn).
recv_timeout(duration) có chờ, nhưng chỉ tới một giới hạn thời gian. Nếu không có gì đến kịp, bạn nhận được Timeout.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::<i32>();

    // Chưa gửi gì: try_recv trả về ngay.
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // Đã có một giá trị đang chờ, nên lệnh này trả nó về mà không phải chờ.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // Đầu gửi đã không còn và channel đã rỗng.
    println!("{:?}", rx.try_recv());
}

Kết quả:

Err(Empty)
Ok(7)
Err(Disconnected)

▶ Tự chạy thử

Chúng ta join() luồng trước khi nhận, nên giá trị đã nằm chờ sẵn và kết quả lần nào cũng giống nhau.

try_recv() rất tiện trong một vòng lặp còn việc khác phải làm, như vòng lặp game kiểm tra tin nhắn mới mỗi khung hình. recv_timeout() hợp với kiểu “bỏ cuộc nếu worker mất quá lâu”.

Channel giới hạn với `sync_channel`

Một channel() bình thường không có giới hạn kích thước. Nếu đầu gửi nhanh hơn đầu nhận nhiều, tin nhắn sẽ chất đống trong bộ nhớ. mpsc::sync_channel(n) tạo một channel chứa tối đa n tin nhắn. Khi đầy, send() sẽ chờ tới khi đầu nhận lấy bớt một cái ra. Điều này gọi là backpressure (áp lực ngược): một consumer chậm sẽ tự động làm chậm một producer nhanh.

use std::sync::mpsc;
use std::thread;

fn main() {
    // Mỗi lúc chỉ có chỗ cho 2 tin nhắn.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // chặn lại khi bộ đệm đầy
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("tổng = {total}");
}

Kết quả:

tổng = 15

▶ Tự chạy thử

Producer không bao giờ đi trước main quá hai số. Với sync_channel(0), không có bộ đệm nào cả: mỗi send() đều chờ tới khi đầu nhận sẵn sàng lấy giá trị, giống như trao tận tay một gói hàng.

Một worker pool nhỏ

Giờ hãy ghép mọi thứ lại. Worker pool là vài luồng lấy việc từ một hàng đợi chung và gửi kết quả về. Nó dùng hai channel: một cho công việc, một cho kết quả.

Có một điểm vướng: Receiver không clone được (nó là consumer duy nhất). Vì vậy các worker dùng chung đầu nhận công việc đặt sau Arc<Mutex<_>>, và lần lượt lấy việc từ đó:

use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // Receiver không clone được, nên các worker dùng chung một cái sau Arc<Mutex<_>>.
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // hết việc
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // báo cho các worker biết là hết việc rồi

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n} bình phương là {square}");
    }
}

Kết quả:

1 bình phương là 1
2 bình phương là 4
3 bình phương là 9
4 bình phương là 16
5 bình phương là 25
6 bình phương là 36

▶ Tự chạy thử

Cách nó tắt gọn gàng:

drop(job_tx) đóng channel công việc. Khi đó recv() của mỗi worker trả về Err, và worker thoát khỏi vòng lặp.
Khi một worker kết thúc, bản clone result_tx của nó bị drop. Chúng ta đã drop bản gốc từ trước, nên khi cả ba worker xong, channel kết quả sẽ đóng.
Lúc đó result_rx.iter() dừng lại, và chúng ta sắp xếp kết quả vì các worker xong theo thứ tự bất kỳ.

Một chi tiết nhỏ: job_rx.lock().unwrap().recv() được cố ý viết trong một câu lệnh let. Khóa được nhả ra ở cuối câu lệnh đó, nên các worker khác có thể giành việc tiếp theo trong khi worker này vẫn đang bình phương số của nó.

Channel hay `Arc<Mutex<T>>`

Cả hai đều an toàn. Chúng chỉ hợp với những dạng bài toán khác nhau:

Dùng channel khi…	Dùng `Arc<Mutex<T>>` khi…
Công việc chảy theo một chiều: producer tạo ra, consumer xử lý	Nhiều luồng cần đọc và cập nhật cùng một giá trị (một bộ đếm, một cache)
Bạn muốn một hàng đợi công việc, sự kiện hoặc dòng log	Dữ liệu lớn và việc sao chép hay di chuyển nó qua lại sẽ lãng phí
Mỗi giá trị chỉ có một chủ sở hữu tại một thời điểm	Các luồng cần xem trạng thái hiện tại, không phải một dòng cập nhật

Một quy tắc dễ nhớ: nếu bạn có thể mô tả chương trình là “những luồng này báo cho luồng kia một điều gì đó”, hãy dùng channel. Nếu là “những luồng này cùng sửa thứ này”, hãy dùng Mutex. Nhiều chương trình thực tế dùng cả hai, như worker pool của chúng ta.

Lỗi thường gặp

Dùng một giá trị sau khi đã gửi nó. send() lấy quyền sở hữu, giống như khi truyền một giá trị vào hàm. Sau khi gửi, giá trị thuộc về đầu nhận:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("hi");
        tx.send(message).unwrap();
        println!("sent {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:25
   |
 8 |         let message = String::from("hi");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("sent {message}");
   |                         ^^^^^^^ value borrowed here after move

Thông báo lỗi (luôn bằng tiếng Anh) nói rằng message đã bị move vào send() nên không thể mượn lại để in. Đây là Rust đang bảo vệ bạn: nếu đầu gửi vẫn còn thay đổi được message, cả hai luồng sẽ cùng đụng vào một dữ liệu. Hãy in trước khi gửi, hoặc gửi một bản .clone().

Move đầu gửi vào một vòng lặp tạo luồng. Luồng đầu tiên lấy mất tx, nên không còn gì cho luồng thứ hai:

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

Trình biên dịch cho biết tx đã bị move vào closure ở lần lặp trước. Cách sửa là thêm let tx = tx.clone(); ở đầu vòng lặp, như trong ví dụ nhiều producer.

Một chương trình không bao giờ kết thúc. Lỗi này không có thông báo nào từ trình biên dịch, nên rất khó phát hiện. Nếu bạn lặp qua một đầu nhận (for x in rx) trong khi vẫn còn một đầu gửi nào đó sống, thường là bản tx gốc trong main, vòng lặp sẽ chờ mãi mãi. Khi chương trình của bạn “cứ treo đó”, hãy kiểm tra xem mọi đầu gửi đã được drop chưa.

Bước tiếp theo

std::sync::mpsc đáp ứng được nhiều việc, nhưng bạn sẽ gặp hai lựa chọn thay thế phổ biến:

crossbeam-channel (tiếng Anh) thêm channel nhiều consumer (đầu nhận có thể clone) và macro select! để chờ trên nhiều channel cùng lúc.
tokio::sync::mpsc là channel cho code bất đồng bộ. Nếu bạn dùng async/.await (xem Async/Await), hãy dùng nó thay cho bản của std, vì recv().await của nó không chặn cả luồng.

Tóm tắt

mpsc::channel() cho bạn một cặp (Sender, Receiver). send() đưa một giá trị vào, recv() chờ giá trị tiếp theo.
Clone đầu gửi để có nhiều producer. Lúc nào cũng chỉ có một đầu nhận.
Lặp qua một đầu nhận sẽ kết thúc khi mọi đầu gửi đã bị drop, nên nhớ drop() bản gốc.
try_recv() và recv_timeout() cho phép kiểm tra mà không phải chờ mãi. sync_channel(n) giới hạn hàng đợi ở n tin nhắn.
Tin nhắn từ nhiều luồng đến theo thứ tự bất kỳ: hãy sắp xếp hoặc cộng chúng lại nếu bạn cần kết quả ổn định.
Dùng channel khi dữ liệu chảy từ luồng này sang luồng khác. Dùng Arc<Mutex<T>> khi các luồng cùng chia sẻ và sửa một giá trị.

Kiểm tra nhanh

1. "mpsc" là viết tắt của gì?

2. Làm sao để ba luồng cùng gửi vào một channel?

3. Một vòng for value in rx không bao giờ kết thúc. Nguyên nhân có khả năng nhất là gì?

4. send() làm gì khi một sync_channel(2) đã chứa 2 tin nhắn?

Điểm: 0 / 4

Đọc thêm

Using Message Passing to Transfer Data Between Threads — The Rust Book (tiếng Anh): phần giới thiệu chính thức về channel, với nhiều ví dụ hơn.
std::sync::mpsc — tài liệu thư viện chuẩn (tiếng Anh): mọi phương thức của Sender, SyncSender và Receiver.
Channels — Rust by Example (tiếng Anh): một ví dụ ngắn, chạy được, với nhiều luồng producer.
From Single-Threaded to Multithreaded Server — The Rust Book (tiếng Anh): xây dựng một thread pool hoàn chỉnh với channel và Arc<Mutex<Receiver>>, từng bước một.
tokio::sync::mpsc — docs.rs (tiếng Anh): channel bất đồng bộ bạn sẽ dùng với Tokio, và nó khác bản của std ra sao.

Channel trong Rust: truyền tin nhắn giữa các luồng

Channel đầu tiên của bạn

Nhận nhiều giá trị

Nhiều producer: clone đầu gửi

Không chờ mãi: try_recv và recv_timeout

Channel giới hạn với sync_channel

Một worker pool nhỏ

Channel hay Arc<Mutex<T>>

Lỗi thường gặp

Bước tiếp theo

Tóm tắt

Kiểm tra nhanh

Đọc thêm

Không chờ mãi: `try_recv` và `recv_timeout`

Channel giới hạn với `sync_channel`

Channel hay `Arc<Mutex<T>>`