Rust 中 Actor 并发模型的实践与使用

简介

Actor 模型是一种并行计算模型，提供了一种用于构建并发、分布式系统的形象办法。在 Actor 模型中，计算被示意为独立的、轻量级的计算单元，称为 Actor，能够发送和接管音讯并进行本地计算。

作为一种通用的消息传递编程模型，被广泛用于构建大规模可伸缩分布式系统。其核心思想是独立保护隔离状态，并基于消息传递实现异步通信。

Actor 模型组成

• 存储：每个 Actor 持有一个邮箱（mailbox），实质上是一个队列，用于存储音讯。
• 通信：每个 Actor 能够发送音讯至任何 Actor，应用异步消息传递，不保障音讯达到指标 Actor 时的程序。
• 计算：Actor 能够通过解决音讯来更新外部状态，对于内部而言，Actor 的状态是隔离的isolated state。

优劣势

• 每个 Actor 独立运行，因而程序天然是并行的
• 每个 Actor 都齐全独立于其余实例，不存在共享内存和竞争条件的问题，能够防止并发编程中的一些难点
• 通过消息传递进行通信，每个 Actor 都有一个邮箱来接管音讯，每次只解决一个音讯，以确保状态的一致性和线程平安
• 消息传递是异步的，发送方不须要期待接管方的响应，从而防止了锁和同步的开销
• 提供高度的并发性和可扩展性，可能无效地解决多核 CPU 和分布式系统中的并发编程问题
• 提供良好的容错性和可恢复性，因为每个 Actor 都有本人的状态和行为，能够更容易地实现零碎的容错和复原

Actor 模型的实现方式

Actor 基于线程的任务调度

为每一个 Actor 调配一个独立的执行过程（线程），独占该线程，能够是操作系统线程，协程或者虚拟机线程。如果以后 Actor 的邮箱为空，Actor 会阻塞以后线程，期待接管新的音讯。

因为线程数量受到系统资源零碎的限度，因而 Actor 的数量也会受到限制。

Actor 基于事件驱动的任务调度

只有在事件触发（即接管音讯）时，才为 Actor 的任务分配线程并执行，当事件处理完毕，即退出线程。该形式能够应用很少的线程来执行大量 Actor 产生的工作，也是当初大部分 Actor 模型所采纳的调度形式。

这种实现与 run loop、event loop 机制十分类似

• 如下示例了三种方式的 Actor 系统示例：
  • 使用 Tokio 实现 Actor 系统
  • Actor 模型实现 之 actix 框架
  • Actor 模型实现 之 ghost_actor 框架

使用 Tokio 实现 Actor 系统

Actor 分为两部分：任务 Task 和 handle。 该任务是独立生成的Tokio任务，实际上执行 Actor 的职责，而 handle 是一种允许你与该任务进行通信的结构。

​ 让我们考虑一个简单的 Actor 。 Actor 在内部存储一个计数器，该计数器用于获取某种唯一ID。 Actor 的基本结构如下所示：

use tokio::sync::{
   oneshot, mpsc};

struct MyActor {
   
    receiver: mpsc::Receiver<ActorMessage>,
    next_id: u32,
}
enum ActorMessage {
   
    GetUniqueId {
   
        respond_to: oneshot::Sender<u32>,
    },
}

impl MyActor {
   
    fn new(receiver: mpsc::Receiver<ActorMessage>) -> Self {
   
        MyActor {
   
            receiver,
            next_id: 0,
        }
    }
    fn handle_message(&mut self, msg: ActorMessage) {
   
        match msg {
   
            ActorMessage::GetUniqueId {
    respond_to } => {
   
                self.next_id += 1;

                // The `let _ =` ignores any errors when sending.
                // `let _ =` 忽略了发送的任何 error
                // This can happen if the `select!` macro is used
                // to cancel waiting for the response.
                // 当 `select!` 宏被用到时将会停止接受响应
                let _ = respond_to.send(self.next_id);
            }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41