流与消费者

本指南通过持久化消费者和临时消费者读取 JetStream 流，选择新消费者从哪里开始，读取投递的元数据，并成批接收消息。

流保存发布到它的 subject 上的消息，每条消息都有一个序号。消费者按自己的节奏读取流，并逐条确认消息。

示例是一个网店。它履行新订单，在看板上显示每个订单事件，并与银行核对已付款的订单。

流

流是服务器的状态。服务在启动时查找它，从不创建它，所以要用 NATS 命令行工具创建一次：

nats stream add ORDERS --subjects 'orders.*' --defaults

ORDERS 保存发布到 orders.created、orders.paid 以及 orders.* 匹配的其他任何 subject 上的每条消息。向这样的 subject 做一次普通发布，消息就会被保存，不论发布者是谁。订阅服务器上不存在的流，会让服务启动失败。

持久化消费者

宏
手写

src/main.rs
/// What the warehouse is asked to ship. The sequence is the order's place in the stream.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments.requested")]
struct Shipment {
    order: u64,
    sequence: Option<u64>,
}

#[subscriber(
    JetStreamSubject::new("orders.created", "ORDERS").durable("fulfilment"),
    reply
)]
async fn fulfil(order: &Order, Ctx(sequence): Ctx<StreamSequence>) -> Shipment {
    tracing::info!(order.id, ?sequence, "order sent to the warehouse");
    Shipment {
        order: order.id,
        sequence,
    }
}

src/main.rs
/// What the warehouse is asked to ship. The sequence is the order's place in the stream.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Shipment {
    order: u64,
    sequence: Option<u64>,
}

impl OutgoingDestination for Shipment {
    type Form = FixedName;

    const DESTINATION: &'static str = "shipments.requested";
}

impl MessageHeaders for Shipment {
    type Contract = NoHeaders;
}

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

struct Fulfil;

impl Handle<Order, Shipment, (), JetStreamContext> for Fulfil {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, JetStreamContext>,
    ) -> impl Future<Output = Result<Shipment, HandlerOutcome>> {
        let sequence = ctx.context(STREAM_SEQUENCE);
        tracing::info!(order.id, ?sequence, "order sent to the warehouse");
        ready(Ok(Shipment {
            order: order.id,
            sequence,
        }))
    }
}

JetStreamSubject 接收消费者读取的 subject 和保存它的流。 subject 就是消费者的过滤条件，所以 fulfil 只从 ORDERS 中读取 orders.created。两者需要不同时，filter_subject 设置另一个过滤条件。服务在启动时创建消费者，带上描述符的每一项设置。

durable 把消费者命名为 fulfilment。服务器保存它的位置，所以重启后的服务从停下的地方继续。服务的各个副本共用这一个持久化消费者，每条消息交给其中一个副本。这就是 JetStream 中与队列组相对应的机制。

处理器用 Ctx<StreamSequence> 读取投递在流中的位置。 JetStream 上下文还带有流和消费者的名字、消费者序号、投递次数，以及仍在等待的消息数。在手写形式中，处理器接收 Context<'_, JetStreamContext>，用 ctx.context(..) 读取同样的键。在 Core NATS 投递上，每个键读出的都是 None。

临时消费者

宏
手写

src/main.rs
#[subscriber(JetStreamSubject::new("orders.*", "ORDERS").deliver_policy(DeliverPolicy::New))]
async fn show(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order event on the dashboard");
    HandlerOutcome::ack()
}

src/main.rs
struct Show;

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

没有 durable 的描述符创建一个临时消费者。每个进程得到自己的消费者，订阅关闭后服务器删除它。所以服务的每个副本都显示每个订单事件。

DeliverPolicy 决定新消费者从哪里开始。默认的 All 从流保存的第一条消息开始，New 从下一条发布的消息开始。 Last、LastPerSubject、ByStartSequence 和 ByStartTime 分别从最后一条消息、每个 subject 的最后一条消息、某个序号或某个时间点开始。该策略只在创建消费者时生效一次；之后，持久化消费者从自己的位置继续。

批

宏
手写

src/main.rs
#[subscriber(
    JetStreamSubject::new("orders.paid", "ORDERS")
        .durable("reconciler")
        .pull_expires(NonZeroDuration::from_secs(nonzero!(1)))
)]
async fn reconcile(orders: &[Order]) -> HandlerOutcome {
    let total: u64 = orders.iter().map(|order| order.amount_cents).sum();
    tracing::info!(orders = orders.len(), total, "paid orders reconciled");
    HandlerOutcome::ack()
}

src/main.rs
struct Reconcile;

impl Handle<[Order]> for Reconcile {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        let total: u64 = orders.iter().map(|order| order.amount_cents).sum();
        tracing::info!(orders = orders.len(), total, "paid orders reconciled");
        ready(Ok(()))
    }
}

接收 &[Order] 的处理器得到一批消息，并为整批只运行一次。它返回的结果结算这批中的每条消息。

应用

宏
手写

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(fulfil);
            b.include(show);
            b.include(reconcile.batch(nonzero!(100)));
        },
    )
}

src/main.rs
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(
                subscriber(
                    JetStreamSubject::new("orders.created", "ORDERS").durable("fulfilment"),
                    Fulfil,
                )
                .reply()
                .build(),
            );
            b.include(
                subscriber(
                    JetStreamSubject::new("orders.*", "ORDERS").deliver_policy(DeliverPolicy::New),
                    Show,
                )
                .build(),
            );
            b.include(
                subscriber(
                    JetStreamSubject::new("orders.paid", "ORDERS")
                        .durable("reconciler")
                        .pull_expires(NonZeroDuration::from_secs(nonzero!(1))),
                    Reconcile,
                )
                .batch(nonzero!(100usize))
                .build(),
            );
        },
    )
}

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

挂载点指定批的大小：batch(100) 在一次拉取中向服务器请求最多一百条消息。描述符上的 pull_expires 决定一次拉取等待它们多久，这里是一秒，之后交出一个较小的批。默认值是五秒。

fulfil 的回复通过 Core NATS 发布，这是 Broker 的默认方式。向流发布则把回复保存到流中。 Core 订阅和 JetStream 订阅共用一个连接，所以一个服务可以自由混用两种模型。

max_ack_pending 限制一个消费者同时持有的未确认投递数；默认值是 1024。确认与投递保证讲解每次投递如何结算。

流​

持久化消费者​

临时消费者​

批​

应用​

流

持久化消费者

临时消费者

批

应用