Mô hình Actor trong Rust: Xây dựng hệ thống đồng thời với tokio và Channels

Mô hình Actor trong Rust: Xây dựng hệ thống đồng thời với tokio và Channels

Việc truyền thông điệp giúp cô lập trạng thái, ngăn chặn các lỗi điều kiện tranh chấp (race conditions) ngay từ thiết kế trong các kiến trúc đồng thời.

Chúng ta sẽ xây dựng gì

Chúng ta đang xây dựng một pipeline xử lý sự kiện đồng thời, trong đó các đơn vị logic riêng biệt, được gọi là các actor, giao tiếp nghiêm ngặt thông qua các kênh bất đồng bộ (asynchronous channels). Thay vì chia sẻ trạng thái có thể thay đổi (mutable state) được bảo vệ bởi mutexes, các actor duy trì các vùng nhớ cô lập và trao đổi dữ liệu thông qua các hàng đợi tin nhắn có kiểu (typed message queues).

Mẫu thiết kế này rất cần thiết để xây dựng các dịch vụ backend mạnh mẽ, có khả năng mở rộng, các pipeline dữ liệu phân tán và các nhóm worker có thông lượng cao, nơi sự cô lập giúp ngăn chặn việc làm hỏng dữ liệu và tình trạng tắc nghẽn (deadlocks). Chúng ta sẽ triển khai một trình tổng hợp log đơn giản định tuyến các sự kiện đầu vào đến các bộ xử lý chuyên biệt, từ đó chứng minh cách Mô hình Actor mở rộng quy mô trong Rust.

Bước 1 — Định nghĩa kiểu tin nhắn rõ ràng

Tin nhắn đóng vai trò là hợp đồng giữa các actor, thay thế việc truy cập bộ nhớ trực tiếp bằng các "bao bì" có kiểu để thực thi tính an toàn của dữ liệu.

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum LogEvent {
    Warning { message: String },
    Error { code: u32, message: String },
}

pub struct ProcessAck {
    pub id: u64,
    pub status: String,
}

Hệ thống kiểu mạnh (strong typing) của Rust đảm bảo rằng người gửi biết chính xác người nhận mong đợi gì, ngăn chặn các lỗi runtime do tải dữ liệu không khớp gây ra.

Bước 2 — Xây dựng trạng thái Actor cô lập

Một actor đóng gói trạng thái có thể thay đổi của chính nó trong một struct, đảm bảo rằng các phép ghi đồng thời không bao giờ can thiệp vào nhau ranh giới của các actor.

struct LogActor {
    buffer: Vec<String>,
    id: u64,
}

impl LogActor {
    async fn handle(&mut self, event: LogEvent) -> Result<ProcessAck, ()> {
        match event {
            LogEvent::Warning { message } => {
                println!("Warning {}: {}", self.id, message);
                Ok(ProcessAck { id: self.id, status: "processed".to_string() })
            }
            _ => Ok(ProcessAck { id: self.id, status: "ignored".to_string() }),
        }
    }
}

Sự cô lập này đảm bảo rằng một lỗi trong việc xử lý một mục nhập log không bao giờ làm hỏng trạng thái nội bộ của actor hoặc chặn các worker khác đọc dữ liệu.

Bước 3 — Thiết lập giao tiếp qua kênh

Giao tiếp diễn ra thông qua các kênh tokio::sync::mpsc, nơi người gửi truyền tin nhắn và người nhận xử lý chúng theo tốc độ riêng.

let (sender, mut receiver) = mpsc::channel(256);
let actor = LogActor { id: 1, buffer: Vec::new() };
let worker_handle = tokio::spawn(async move {
    while let Some(msg) = receiver.recv().await {
        let _ = actor.handle(msg).await;
    }
});

Việc sử dụng một kênh có giới hạn (bounded channel) ở đây cung cấp áp suất ngược (backpressure), ngăn chặn việc cạn kiệt bộ nhớ nếu người gửi đẩy dữ liệu nhanh hơn khả năng xử lý của actor.

Bước 4 — Triển khai vòng lặp sự kiện

Vòng đời của actor được định nghĩa bởi một vòng lặp while let chặn cho đến khi một tin nhắn mới đến, tại thời điểm đó nó thực thi một trình xử lý.

loop {
    match receiver.recv().await {
        Some(msg) => {
            // Thực hiện hành động
        }
        None => break, // Kênh đã đóng
    }
}

Từ khóa await đảm bảo I/O không chặn (non-blocking), cho phép runtime nhường quyền kiểm soát khi kênh trống mà không làm dừng toàn bộ hệ thống.

Bước 5 — Xử lý lỗi một cách linh hoạt

Các lỗi bên trong vòng lặp phải được bắt rõ ràng, đảm bảo rằng actor có thể ghi lại các lỗi và quyết định whether để phục hồi hay chấm dứt.

match actor.handle(msg).await {
    Ok(ack) => {
        // Xử lý thành công
    }
    Err(e) => {
        eprintln!("Error processing event {}: {}", ack.id, e);
        // Quyết định panic hoặc tiếp tục dựa trên mức độ nghiêm trọng
    }
}

Mẫu này đảm bảo rằng một tin nhắn xấu không gây ra sự cố cho toàn bộ pipeline, duy trì sự ổn định dưới tải cao (load).

Những điểm chính

Mô hình Actor trong Rust dựa vào sự cô lập thay vì đồng bộ hóa. Bằng cách định nghĩa các loại tin nhắn nghiêm ngặt, đóng gói trạng thái trong các struct actor và sử dụng các kênh tokio để giao tiếp, chúng ta xây dựng các hệ thống có khả năng mở rộng mà không có lỗi điều kiện tranh chấp. Các kênh có giới hạn đóng vai trò là các cơ chế backpressure tự nhiên, ngăn chặn các vấn đề về bộ nhớ. Mẫu này đặc biệt hiệu quả cho các kiến trúc hướng sự kiện (event-driven) và vi dịch vụ (microservices), nơi độ trễ và sự ổn định là tối quan trọng.

Tiếp theo là gì?

Trong bài viết tiếp theo, chúng ta sẽ khám phá cách triển khai các actor phân tán sử dụng gRPC và cách quản lý backpressure ở tầng mạng. Chúng ta cũng sẽ đề cập đến việc kiểm thử các hệ thống dựa trên actor để đảm bảo tính mạnh mẽ trong môi trường sản xuất.

Tài liệu tham khảo thêm

• Designing Data-Intensive Applications
• A Philosophy of Software Design

Phần của loạt bài Mẫu thiết kế kiến trúc (Architecture Patterns).