SQL 数据库

英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

ruststream-sqlx 通过 sqlx 把 SQL 数据库接入 RustStream 服务。它包含两个组件：

• 收件箱（inbox）：任务队列，存放在服务自己的 Postgres、MySQL/MariaDB 和 SQLite 表中， 作为 RustStream Broker 运行；
• 事务性发件箱（transactional outbox）：服务通过任意 RustStream Broker 发布的消息， 会记录在服务自己的一张表中，直到消费方处理完毕。

ruststream = { version = "0.7", features = ["macros"] }
ruststream-sqlx = { version = "0.7", features = ["inbox", "outbox", "postgres"] }
sqlx = { version = "0.9", features = ["runtime-tokio", "postgres", "derive", "chrono"] }
chrono = "0.4"
serde = { version = "1", features = ["derive"] }

inbox 和 outbox 两个 feature 分别开启两个组件，彼此独立。 驱动 feature 选择数据库：postgres、mysql（MySQL 和 MariaDB）、sqlite， 或者用于 AnyPool 的 any。

服务表中的任务队列

crates/ruststream-sqlx/examples/inbox.rs

use std::error::Error;

use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnectOptions, PgConnection, PgPool, Postgres};

/// One row of `email_jobs`: the struct names the table and the role of each column.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A publish to `emails` writes a row through the service's own statement.
impl Publish<Postgres> for SendEmail {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[derive(Deserialize, Serialize, Outgoing)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    println!("sending to {}", email.to);
    HandlerOutcome::ack()
}

fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
        SqlxBroker::new(pool).route::<SendEmail>("emails"),
        |b| {
            b.include(send);
        },
    )
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // sqlx builds a pool only inside a Tokio runtime, so the service builds it here, before the
    // app.
    let pool = PgPool::connect_with(PgConnectOptions::new()).await?;
    app(pool.clone()).run().await?;
    pool.close().await;
    Ok(())
}

#[derive(Inbox)] 描述队列表：表名，以及每一列承担的角色。 SqlxBroker 为服务连接池中的队列提供服务。订阅领取行，处理器的结果决定如何处置这一行： 确认会删除这一行，重新处理会把它放回队列。发布会用服务自己的语句写入一行。

订阅以三种方式之一领取行。在行锁方式下，处理器运行期间事务一直保持打开。 在租约方式下，订阅立即提交领取。在咨询锁方式下，订阅锁住这一行的键。 用 .transactional() 注册的处理器通过本次投递的事务写入数据， 确认会把处理器的写入与这一行一起提交。 每一列的角色开启一种行为：分组、行进入处理的顺序、延迟重新处理、 带死信转移的尝试次数上限，以及让已处理的行保留在表中的标记。

手写 API 是与 #[derive] 并列的另一种方式，描述的是同一张表：结构体实现一个 trait， 表的全部设置由一个带类型的构建器给出。编译器检查这份描述，与检查 #[derive] 一样严格。 两种方式生成相同的语句，开销也相同。

在编译期检查 SQL 的团队，可以在 #[derive] 中加上 checked。 这样 cargo sqlx prepare 会把 #[derive] 生成的语句与服务自己的查询一起检查。 测试在 TestApp 中针对真实数据库运行服务自己的应用，因为服务的 SQL 也属于测试要检查的内容。 SQLite 不需要服务器。

事务性发件箱

crates/ruststream-sqlx/examples/outbox.rs

use std::convert::Infallible;

use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream::memory::prelude::*;
use ruststream_sqlx::{Outbox, outbox};
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnectOptions, PgConnection, PgPool, Postgres};

/// One record of the `outbox` table.
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox")]
struct OrderOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

/// The service writes the record; the outbox sends its id with the message.
impl outbox::Publish<Postgres> for OrderOutbox {
    async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

#[derive(Deserialize)]
struct PlaceOrder {
    id: u64,
}

#[derive(Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct OrderPlaced {
    id: u64,
}

/// The reply to `orders` is recorded before it is sent.
#[subscriber("checkout", reply)]
async fn place(cmd: &PlaceOrder) -> OrderPlaced {
    OrderPlaced { id: cmd.id }
}

/// After the ack, the record is marked processed.
#[subscriber("orders")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
    println!("fulfilling order {}", order.id);
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let tracking = outbox! {
        "orders" => OrderOutbox,
    };
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // sqlx builds a pool only inside the Tokio runtime, which starts after this builder.
        .on_startup(async move |()| {
            let pool = PgPool::connect_with(PgConnectOptions::new()).await?;
            registry
                .set_pool(pool.clone())
                .expect("the outbox is deferred and gets one pool");
            Ok::<_, sqlx::Error>(pool)
        })
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .after_shutdown(async move |pool| {
            pool.close().await;
            Ok::<_, Infallible>(())
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(fulfil);
            // Once the subscriptions are open, the unprocessed records are sent again.
            b.after_startup(Publish, tracking.republish());
        })
}

#[derive(Outbox)] 描述服务的发件箱表，outbox! 按它跟踪的名称注册这张表。 发布中间件在发送每条被跟踪的消息之前先写入一条记录，并把记录 ID 放进消息头。 订阅中间件根据这个 ID 领取记录，处理器确认后把记录标记为已处理。 启动时，重新发布会再次发送所有尚未处理的记录。

消息至少投递一次。同时启动的两个服务实例，可能都会再次发送同一条记录。 因此，不能重复处理消息的消费方需要自己检查。

在 #[ruststream::app] 中，应用在 Tokio 运行时启动之前构建，而 sqlx 只能在运行时内创建连接池。 因此，注册表启动时没有连接池，由 on_startup 创建连接池，并通过 set_pool 交给它。 测试构建默认关闭发件箱，环境中设置 RUSTSTREAM_SQLX_OUTBOX=on 后才会开启。

发件箱的中间件包装收件箱的处理器，与包装其他处理器的方式相同。 记录通过单独的连接写入，因此即使投递的事务回滚，记录也会提交。 被跟踪并写入收件箱表的消息，只要表中有 headers 列、并由服务的发布语句写入， 就会保留它的记录 ID。其余内容见 事务性发件箱。

如何选择

收件箱适合属于服务数据的工作：任务与它所服务的数据写在同一个事务中， 队列放在服务已经在用的数据库里。发件箱适合服务通过另一个 Broker 发布、且不能丢失的消息。 一个服务可以同时使用两者。

其余内容在哪里

docs.rs 上的参考文档以这个 crate 自己的教程开篇，每个主题一节：

• 收件箱 Broker： 队列表、它的订阅，以及处理器的结果如何处置一行。
• 宏还是手写： 通过 InboxTable 和带类型的构建器手写描述队列表。
• 角色和 时间： 每一列开启什么，以及“当前时间”从哪里来。
• 行锁、租约和咨询锁和 事务模式： 订阅如何领取行，以及处理器如何通过本次投递的事务写入数据。
• 行模式、 头部布局和 批处理： 处理器从一行中得到什么。
• 数据库和 服务自己的方言： 每种数据库执行什么语句，以及为其他数据库编写方言。
• 编译期检查语句： checked 模式。
• 唤醒订阅： 轮询间隔，以及 Postgres 上的 LISTEN/NOTIFY。
• 在收件箱上测试服务： 在 TestApp 中运行服务自己的应用。
• 事务性发件箱： 投递保证、记录及其注册表、中间件、重新发布、连接池、测试和开销。

这个 crate 每条消息的开销，以及它每秒处理的消息数，与做同样工作的原生 sqlx 循环的对比，见基准测试页面。

处理器、路由器、编解码器和中间件都来自框架本身，它的入口页面从 RustStream 站点开始。