Concurrency an toàn trong Rust với Mutex và channel

1. Tổng quan

Rust gọi khả năng của mình là “fearless concurrency”. Compiler chặn data race ngay lúc build, trước khi chương trình kịp chạy. Bài này trình bày hai công cụ chính trong thư viện chuẩn: Mutex (kèm Arc) để chia sẻ state, và mpsc channel để truyền message giữa các thread. Đọc xong, bạn sẽ biết dùng từng loại, viết test cho chúng và tránh những bẫy hay gặp như deadlock hay giữ lock quá lâu.

Mọi ví dụ chỉ dùng thư viện chuẩn, không cần dependency ngoài. Bạn có thể tạo project bằng cargo new orders-demo rồi dán code vào src/main.rs.

2. Vấn đề: chia sẻ dữ liệu giữa các thread

Giả sử bạn có nhiều thread cùng xử lý đơn hàng và muốn đếm tổng số đơn. Cách viết ngây thơ như sau:

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

Đoạn code này không biên dịch được:

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

Compiler đưa ra hai lý do:

• thread::spawn yêu cầu closure là 'static. Thread con có thể sống lâu hơn main, nên nó không được mượn biến cục bộ.
• Kể cả khi thêm move, mỗi thread sẽ nhận một bản sao riêng của total_orders. Không ai thực sự cộng dồn vào một biến chung.

Để giải quyết, bạn cần hai thứ: chia sẻ ownership bằng Arc và đồng bộ truy cập bằng Mutex.

3. Chia sẻ state với Arc<Mutex<T>>

3.1. Đếm đơn hàng từ nhiều thread

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

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&total_orders);
        handles.push(thread::spawn(move || {
            for _ in 0..1_000 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }

    println!("Tổng đơn: {}", *total_orders.lock().unwrap());
}

Tổng đơn: 4000

Giải thích từng phần:

• Arc::new(Mutex::new(0u32)): Arc (Atomic Reference Counted) cho phép nhiều thread cùng sở hữu một giá trị. Mutex đảm bảo tại mỗi thời điểm chỉ một thread được truy cập giá trị bên trong.
• Arc::clone(&total_orders) chỉ tăng bộ đếm tham chiếu, không copy dữ liệu. Mỗi thread nhận một Arc riêng qua move.
• counter.lock() chặn thread hiện tại cho tới khi lấy được lock. Hàm trả về Result<MutexGuard<u32>, PoisonError<…>>.
• MutexGuard deref ra &mut u32, nên *... += 1 sửa trực tiếp giá trị. Guard bị drop ở cuối câu lệnh, và lock được nhả tự động tại đó.

Bạn không thể “quên unlock” trong Rust. Bạn cũng không thể đọc giá trị mà bỏ qua lock, vì dữ liệu nằm bên trong Mutex.

Lưu ý: Bạn không thể thay Arc bằng Rc. Rc không implement Send, và compiler sẽ báo lỗi ngay khi bạn đưa nó vào thread::spawn. Tương tự, bạn không thể thay Mutex bằng RefCell, vì RefCell không phải Sync.

3.2. Phạm vi của MutexGuard

Lock được giữ cho tới khi guard bị drop. Nếu bạn gán guard vào biến, lock sống tới hết scope của biến đó:

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

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // guard bị drop ở đây, lock được nhả

    if reserved {
        // Giả lập gọi PaymentService chậm, KHÔNG giữ lock
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("Khách {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

Khách 0: true
Khách 1: true
Khách 2: false

Thứ tự các dòng có thể khác nhau giữa các lần chạy. Kết quả luôn là đúng hai true và một false.

Block { ... } giới hạn thời gian giữ lock ở mức tối thiểu. Phần chậm (gọi thanh toán) chạy sau khi lock đã được nhả, nên các thread khác không phải chờ. Nếu cần nhả sớm hơn giữa chừng, bạn có thể gọi drop(stock) một cách tường minh.

3.3. Mutex bị poison

Nếu một thread panic trong khi đang giữ lock, Rust đánh dấu Mutex là poisoned. Lý do là dữ liệu có thể đang ở trạng thái dở dang:

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

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentService lỗi giữa chừng");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("Số dư: {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutex bị poison, số dư hiện tại: {}", *v);
        }
    }
}

thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentService lỗi giữa chừng
Mutex bị poison, số dư hiện tại: 70

Vì vậy, .unwrap() sau lock() thực chất có nghĩa là “nếu thread khác đã panic khi giữ lock thì panic theo”. Với phần lớn ứng dụng, đó là lựa chọn hợp lý. Nếu bạn biết cách khôi phục dữ liệu, into_inner() trả lại guard để bạn kiểm tra và sửa.

4. Truyền message với channel

Mutex trả lời câu hỏi “nhiều thread cùng sửa một dữ liệu”. Channel đi theo hướng ngược lại: mỗi dữ liệu chỉ có một chủ tại một thời điểm, và ownership được chuyển qua channel. Trong nhiều bài toán, cách này đơn giản hơn và ít lỗi hơn.

4.1. Nhiều producer, một consumer

std::sync::mpsc là viết tắt của multiple producer, single consumer:

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

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

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

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("khach-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // quan trọng: nhả sender gốc

    let mut revenue = 0;
    for order in rx {
        println!("Nhận đơn #{} của {}", order.id, order.customer);
        revenue += order.amount;
    }
    println!("Doanh thu: {revenue}");
}

Nhận đơn #100 của khach-1
Nhận đơn #300 của khach-3
Nhận đơn #200 của khach-2
Doanh thu: 900000

Giải thích:

• mpsc::channel() trả về cặp (Sender, Receiver). Mỗi producer nhận một bản tx.clone().
• tx.send(order) move order vào channel. Sau dòng này, thread gửi không còn dùng được order, và compiler sẽ báo lỗi nếu bạn cố dùng. Nhờ vậy, hai thread không bao giờ cùng sửa một Order.
• for order in rx lặp cho tới khi mọi Sender đã bị drop. Nếu thiếu dòng drop(tx), sender gốc vẫn còn sống trong main, và vòng lặp sẽ chờ mãi.

Thứ tự nhận phụ thuộc vào lịch chạy thread. Tổng doanh thu thì luôn là 900000.

4.2. Channel có giới hạn và backpressure

mpsc::channel() có buffer không giới hạn. Nếu producer nhanh hơn consumer, bộ nhớ sẽ phình dần. sync_channel(n) giới hạn buffer ở n phần tử, và send sẽ block khi buffer đầy:

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

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("Đã gửi đơn {id}");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("Xử lý đơn {id}");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}

Đã gửi đơn 1
Đã gửi đơn 2
Xử lý đơn 1
Đã gửi đơn 3
Xử lý đơn 2
...

Producer gửi được hai đơn rồi phải chờ consumer rút bớt. Đây là cơ chế backpressure tự nhiên, rất hữu ích cho pipeline xử lý dữ liệu. Với sync_channel(0), mỗi lần send phải chờ tới khi có người recv (rendezvous).

5. Khi nào không cần Arc: scoped thread

Nếu các thread chắc chắn kết thúc trước khi hàm trả về, thread::scope cho phép bạn mượn dữ liệu trực tiếp, không cần Arc:

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // mọi thread trong scope đã join tại đây

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}

[101, 102, 103, 104]

Nhờ scope, compiler biết các thread không sống lâu hơn processed và orders. Vì thế mượn & là hợp lệ. Sau scope, into_inner() lấy dữ liệu ra khỏi Mutex mà không cần lock.

6. Viết test cho code concurrent

Bạn nên tách logic thành hàm thuần để dễ test. Đặt đoạn sau vào src/lib.rs:

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

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

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

    #[test]
    fn counts_every_order_exactly_once() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn sums_all_payments_from_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn empty_payments_returns_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}

cargo test

running 3 tests
test tests::empty_payments_returns_zero ... ok
test tests::sums_all_payments_from_channel ... ok
test tests::counts_every_order_exactly_once ... ok

test result: ok. 3 passed; 0 failed

Test empty_payments_returns_zero kiểm tra trường hợp biên: nếu quên drop(tx), test này sẽ treo vĩnh viễn thay vì trả về 0. Đó là dấu hiệu rõ ràng để phát hiện lỗi.

Có một điểm nhỏ trong count_orders: giá trị được gán vào biến result trước khi trả về. Lý do là guard tạm thời ở biểu thức cuối hàm có thể sống lâu hơn total, và compiler sẽ báo lỗi borrow.

7. Lỗi hay gặp

Deadlock do khóa ngược thứ tự. Thread A khóa orders rồi tới customers, trong khi thread B khóa customers rồi tới orders. Cả hai sẽ chờ nhau mãi mãi. Rust chỉ ngăn data race, không ngăn deadlock. Bạn nên quy ước một thứ tự khóa cố định, hoặc gộp hai dữ liệu vào chung một Mutex<(A, B)>.

Guard tạm thời sống lâu hơn bạn nghĩ. Xem đoạn code sau:

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // lock vẫn đang bị giữ ở đây!
}

Temporary trong biểu thức của while let (và match) sống tới hết thân vòng lặp. Mọi thread khác muốn đẩy đơn vào queue sẽ bị chặn suốt thời gian process chạy. Cách sửa là tách ra: let next = queue.lock().unwrap().pop(); rồi mới xử lý.

Gọi lại lock() trên cùng thread. std::sync::Mutex không phải reentrant. Gọi lock() hai lần trên cùng một thread sẽ deadlock hoặc panic.

Giữ lock trong khi chờ I/O hoặc recv(). Mọi thao tác blocking khi đang giữ guard đều biến Mutex thành nút thắt cổ chai. Bạn nên copy hoặc clone dữ liệu cần thiết ra ngoài, nhả lock, rồi mới làm việc chậm.

Quên drop Sender. Như đã thấy ở trên, consumer sẽ chờ vô hạn.

Lưu ý: Trong code async (Tokio), không giữ std::sync::MutexGuard qua .await. Hãy dùng tokio::sync::Mutex hoặc channel của Tokio. Ngoài ra, với một số nguyên đơn giản như bộ đếm, AtomicU32 nhanh hơn hẳn Mutex<u32>.

Khi nào chọn cái nào?

• Dùng Mutex khi nhiều thread cần đọc/sửa cùng một state lâu dài, ví dụ cache hay kho hàng. Nếu đọc nhiều hơn ghi rõ rệt, hãy cân nhắc RwLock.
• Dùng channel khi dữ liệu chảy từ nơi này sang nơi khác, ví dụ hàng đợi đơn hàng, pipeline hay worker pool. Channel giúp mỗi giá trị chỉ có một chủ, nên logic dễ suy luận hơn.

8. Kết luận

Rust biến các lỗi concurrency phổ biến thành lỗi biên dịch. Bạn không thể chia sẻ dữ liệu khi thiếu Arc, không thể sửa dữ liệu khi thiếu Mutex, và không thể dùng lại một giá trị đã gửi qua channel. Arc<Mutex<T>> phù hợp với shared state, còn mpsc channel phù hợp với việc chuyển ownership giữa các thread. sync_channel thêm backpressure, và thread::scope giúp bỏ Arc khi vòng đời các thread đã rõ ràng. Những gì compiler không bảo vệ được là deadlock, guard sống quá lâu và sender chưa được drop. Đó là những chỗ bạn cần review kỹ và viết test cẩn thận.