Coerce-rs

Coerce-rs 是一个异步（async/await）Actor运行时和分布式系统框架，为Rust语言提供。它允许通过极少的代码即可构建高度可扩展、容错能力强的现代actor驱动的应用程序。

Crates

crate	目的	最新版本
coerce	Coerce的主要运行时和框架
coerce-redis	Redis的actor持久化提供程序。允许从Redis读取和写入事件源和快照。
coerce-macros	有用的宏，允许快速实现快照、可序列化远程消息等。
coerce-k8s	Kubernetes发现提供程序，根据可配置的Pod选择标签自动发现托管在Kubernetes中的集群节点

在您的项目中使用Coerce

在您的项目中使用Coerce的第一步是添加coerce crate依赖项，这可以通过在Cargo.toml中添加以下内容来完成

[dependencies]
coerce = { version = "0.8", features = ["full"] }

可选：启用 tracing/valuable

Coerce支持tracing/valuable，这可以用于在actor上下文中丰富日志信息。这是一个不稳定的功能，可以通过添加coerce/tracing-unstable功能和以下部分到您的.cargo/config.toml文件来启用

[build]
rustflags = ["--cfg", "tracing_unstable"]

注意：如果您的项目已经依赖于tracing crate，您还需要启用valuable功能！

特性

Actors

• 类型安全的actors
• 监督/子进程创建
• 位置透明 ActorRef<A> 类型（ActorRef 可能包含一个 LocalActorRef<A> 或一个 RemoteActorRef<A>）
• 开箱即用的指标

远程通信

• 在集群中的任何位置与 actor 进行通信
• actor 可以本地部署或部署到其他远程节点
• Protobuf 网络协议
• actor 驱动的网络层

分布式分片

• actor ID 可以解析到特定的分片，这些分片可以分布在整个 Coerce 节点集群中
• 自动负载均衡，分片将在集群中公平分配
• 节点丢失时自动恢复，actor 可以在其他健康节点上自动重启

持久性

• 日志记录/事件溯源
• 快照
• 可插拔的存储提供程序（内存和 redis 立即可用，MySQL 计划中）

分布式 PubSub

• actor 可以从集群中的任何位置订阅可编程主题
• 系统级主题提供用于接收更新的系统状态（例如新节点加入、节点丢失等）

HTTP API

• 易于访问的指标和信息，可用于诊断

构建和测试 Coerce 库

构建 Coerce 很简单。您只需要安装最新的 Rust 稳定版或夜间版，以及 Cargo。

# Clone the repository
git clone https://github.com/leonhartley/coerce-rs && cd coerce-rs

## run Cargo build to build the entire workspace, including the examples and the tests
cargo build

## Alternatively, if you'd like to build the library, dependencies and run the tests
cargo test --all-features

如何运行示例

分片聊天示例

ActorSystem

每个 actor 都属于一个 ActorSystem。

async/await Actors

actor 只是一个计算单元的另一种说法。它可以有可变状态，可以接收消息并执行操作。然而，它一次只能做一件事。这很有用，因为它可以减少对线程同步的需要，通常通过锁定（使用 Mutex、RwLock 等）来实现。

这是如何在 Coerce 中实现的？

Coerce 使用 Tokio 的 MPSC 通道（tokio::sync::mpsc::channel），每个创建的 actor 都会启动一个任务来监听来自 Receiver 的消息，处理并等待消息的结果。每个引用（ActorRef<A: Actor>）都包含一个 Sender<M> where A: Handler<M>，可以克隆。

pub struct EchoActor {}

#[async_trait]
impl Actor for EchoActor {}

pub struct EchoMessage(String);

impl Message for EchoMessage {
    type Result = String;
}

#[async_trait]
impl Handler<EchoMessage> for EchoActor {
    async fn handle(
        &mut self,
        message: EchoMessage,
        _ctx: &mut ActorContext,
    ) -> String {
        message.0.clone()
    }
}

pub async fn run() {
    let mut actor = new_actor(EchoActor {}).await.unwrap();

    let hello_world = "hello, world".to_string();
    let result = actor.send(EchoMessage(hello_world.clone())).await;

    assert_eq!(result, Ok(hello_world));
}

pub struct EchoActor {}

#[async_trait]
impl Actor for EchoActor {}

pub struct EchoMessage(String);

impl Message for EchoMessage {
    type Result = String;
}

pub struct PrintTimer(String);

impl TimerTick for PrintTimer {}

#[async_trait]
impl Handler<PrintTimer> for EchoActor {
    async fn handle(&mut self, msg: PrintTimer, _ctx: &mut ActorContext) {
        println!("{}", msg.0);
    }
}

pub async fn run() {
    let mut actor = new_actor(EchoActor {}).await.unwrap();
    let hello_world = "hello world!".to_string();

    // print "hello world!" every 5 seconds
    let timer = Timer::start(actor.clone(), Duration::from_secs(5), TimerTick(hello_world));

    // timer is stopped when handle is out of scope or can be stopped manually by calling `.stop()`
    sleep(Duration::from_secs(20)).await;
    timer.stop();
}