Subject、通配符与队列组

本指南为 Core NATS subject 命名，用通配符读取多个 subject，在队列组中让各个副本分担工作，并在并行的工作分区中保持每个键的顺序。

subject 是由点分隔的一串 token，比如 orders.eu.created。 Core NATS 把消息交给此刻 subject 与之匹配的每个订阅，并且什么也不存储。

示例是一个订单服务。它打包欧洲的订单，为订单开账单，并审计每个地区的每个订单事件。

队列组

pack 读取 orders.eu.created。服务运行多个副本，而每个订单只能打包一次。

宏
手写

src/main.rs
#[subscriber(CoreSubject::new("orders.eu.created").queue_group("packers"))]
async fn pack(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order packed");
    HandlerOutcome::ack()
}

src/main.rs
struct Pack;

impl Handle<Order> for Pack {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order packed");
        ready(Ok(()))
    }
}

CoreSubject 读取一个 subject。 queue_group 把订阅放进 packers 组。服务器把每条消息交给组内的一个成员，所以每个副本打包一部分订单。对服务器来说，一个组名就是一个组，即使成员订阅的 subject 不同。组外的订阅，或者另一个组里的订阅，仍然收到属于自己的一份。

按客户划分的并行工作分区

bill 读取同一个 subject，但不加入任何组，所以 pack 收到的每个订单它都会收到。它最多同时为八个订单开账单，并保持每个客户的顺序。

宏
手写

src/main.rs
/// The customer an order belongs to, carried in the partition-key header.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct CustomerKey {
    #[serde(rename = "nats-partition-key")]
    customer: String,
}

/// The invoice of one order, keyed by the customer like the order it bills.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "invoices.eu", headers = CustomerKey)]
struct Invoice {
    order: u64,
    amount_cents: u64,
}

#[derive(OutSlot)]
#[publishes(Invoice)]
struct Invoices;

#[subscriber("orders.eu.created", workers(8, by_key))]
async fn bill(
    order: &Order,
    Headers(key): Headers<CustomerKey>,
    Out(invoices): Out<impl Publisher, Invoices>,
) -> HandlerOutcome {
    let invoice = Invoice {
        order: order.id,
        amount_cents: order.amount_cents,
    };
    if invoices
        .message(&invoice)
        .with_headers(&key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, customer = %key.customer, "order billed");
    HandlerOutcome::ack()
}

src/main.rs
/// The customer an order belongs to, carried in the partition-key header.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct CustomerKey {
    #[serde(rename = "nats-partition-key")]
    customer: String,
}

/// The invoice of one order, keyed by the customer like the order it bills.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Invoice {
    order: u64,
    amount_cents: u64,
}

impl OutgoingDestination for Invoice {
    type Form = FixedName;

    const DESTINATION: &'static str = "invoices.eu";
}

impl MessageHeaders for Invoice {
    type Contract = WithHeaders<CustomerKey>;
}

impl MessageInfo for Invoice {
    const NAME: &'static str = "Invoice";
}

struct Invoices;

impl OutSlot for Invoices {
    const NAME: &'static str = "Invoices";
    type Destination = Reads;
}

impl PublishedThrough<Invoices> for Invoice {}

struct Bill;

impl<Sent> Handle<Message<CustomerKey, Order>, (), Outs<(Sent,)>> for Bill
where
    Sent: OutEntry<Invoices, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Message<CustomerKey, Order>,
        outs: &Outs<(Sent,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let key = &order.headers;
        let invoice = Invoice {
            order: order.body.id,
            amount_cents: order.body.amount_cents,
        };
        if outs
            .get(Invoices)
            .message(&invoice)
            .with_headers(key)
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        tracing::info!(order.body.id, customer = %key.customer, "order billed");
        Ok(())
    }
}

workers(8, by_key) 运行八个工作分区。键相同的投递在同一个工作分区上依次运行。在手写形式中，同一设置是 workers_by_key。 NATS 没有原生的键，所以键以 nats-partition-key 消息头（PARTITION_KEY_HEADER）传递，由发送方设置。 CustomerKey 是一个消息头契约，只有一个以该名字命名的字段。 Headers<CustomerKey> 读取投递的键，with_headers(&key) 把它写到账单上。于是对下一个读者来说，同一客户的账单也保持顺序。没有该消息头的订单不满足消息头契约，由解码策略结算，默认会丢弃它。

NATS 消息头是文本：名字是不含冒号的可打印 ASCII 字符，值只有一行。不符合这些要求的消息头会让发布返回错误。

通配符

audit 读取每个地区的每个订单事件。

宏
手写

src/main.rs
#[subscriber(CoreWildcard::new("orders.>"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order event audited");
    HandlerOutcome::ack()
}

src/main.rs
struct Audit;

impl Handle<Order> for Audit {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order event audited");
        ready(Ok(()))
    }
}

CoreWildcard 读取模式匹配的每个 subject。 * 匹配一个 token：orders.*.created 读取每个地区的订单创建事件。 > 匹配末尾的一个或多个 token：orders.> 读取 orders 之下的每个 subject。

应用

宏
手写

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(pack);
            b.include(bill).out(Invoices, Publish).build();
            // A pattern is no subject to publish a retry copy to, so the mount site names one.
            b.include(audit).out_retry(Publish).to("orders.retry");
        },
    )
}

src/main.rs
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(
                subscriber(
                    CoreSubject::new("orders.eu.created").queue_group("packers"),
                    Pack,
                )
                .build(),
            );
            b.include(
                subscriber("orders.eu.created", Bill)
                    .workers_by_key(nonzero!(8usize))
                    .build(),
            )
            .out(Invoices, Publish)
            .build();
            // A pattern is no subject to publish a retry copy to, so the mount site names one.
            b.include(subscriber(CoreWildcard::new("orders.>"), Audit).build())
                .out_retry(Publish)
                .to("orders.retry");
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

模式读取多个 subject，所以它的投递的延迟副本没有唯一的 subject 可去。挂载点用 out_retry(Publish).to(..) 指定一个，这里是 orders.retry，模式同样读取它。每个基于模式的注册都要指定它，即使处理器从不重试；没有它，服务拒绝启动。重试与死信讲解这些副本。

至多一次

Core NATS 把消息投递给它到达时已经存在的订阅。在没有任何订阅匹配时发布的消息就此消失。投递也没有确认，所以处理器失败后，服务器不会再次交给它这条消息。如果每条消息都不能丢，就让 subject 归属一个流：流与消费者展示如何读取流。

示例文件中的测试在进程内运行应用，检查每个订阅收到的那一份、模式，以及每个客户的账单顺序。

队列组​

按客户划分的并行工作分区​

通配符​

应用​

至多一次​

队列组

按客户划分的并行工作分区

通配符

应用

至多一次