并发：工作者与线程

打包是计算任务，所以它有两个专属线程；预约快递要等待网络，所以四个预约同时进行。

默认情况下，一个订阅一次处理一次投递，顺序与 Broker 交付的顺序相同。 工作者让订阅在应用的运行时上同时处理多次投递，适合等待 I/O 的处理器。 线程给订阅分配专属线程，适合做计算的处理器。否则这样的处理器会阻塞为其他所有订阅服务的运行时。 两者都是订阅的设置，处理器的主体不需要改动。 使用工作者或线程的订阅不再保持投递的顺序。workers(n, by_key) 保持每个键内的顺序。

在线程上打包

pack 把已付款的订单拆分成包裹，并以打包方案作为回复。 超出商店上限的已付款订单，在规划任何包裹之前就被丢弃。 threads(2) 随订阅启动两个线程，投递轮流分给它们。 解码、主体、回复和结算都在收到该投递的线程上运行。

宏

shipping.rs

/// Packs a paid order into parcels.
// Packing is computation: two threads of its own keep it off the app's runtime, which serves I/O.
#[subscriber("payments.settled", threads(2), reply)]
async fn pack(paid: &Paid) -> Result<Packed, HandlerOutcome> {
    if !within_bounds(&paid.lines) {
        tracing::warn!(
            order = paid.order,
            "a paid order past the shop's bounds, dropped"
        );
        return Err(HandlerOutcome::drop());
    }
    Ok(Packed {
        order: paid.order,
        customer: paid.customer.clone(),
        parcels: plan(&paid.lines),
    })
}

手写

shipping.rs

/// Packs a paid order into parcels.
struct Pack;

impl Handle<Paid, Packed, (), (), Shop> for Pack {
    fn handle(
        &self,
        paid: &Paid,
        _outs: &(),
        _ctx: &mut Context<'_, (), Shop>,
    ) -> impl Future<Output = Result<Packed, HandlerOutcome>> {
        if !within_bounds(&paid.lines) {
            tracing::warn!(
                order = paid.order,
                "a paid order past the shop's bounds, dropped"
            );
            return ready(Err(HandlerOutcome::drop()));
        }
        ready(Ok(Packed {
            order: paid.order,
            customer: paid.customer.clone(),
            parcels: plan(&paid.lines),
        }))
    }
}

在工作者上预约

dispatch 为已打包的订单预约快递，并把快递单号告诉顾客。 workers(4) 让最多四个预约同时进行，下一次投递等待空闲的工作者。

宏

shipping.rs

/// Books a courier for a packed order and tells the customer.
// Booking waits on the courier's API, so four bookings run at once.
#[subscriber("shipping.packed", workers(4))]
async fn dispatch(packed: &Packed, Out(notices): Out<impl Publisher, Notices>) -> HandlerOutcome {
    let tracking = book_courier(packed).await;
    let notice = Notice {
        customer: packed.customer.clone(),
        text: format!("Order {} is on its way: {tracking}.", packed.order),
    };
    match notices.message(&notice).publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

手写

shipping.rs

/// Books a courier for a packed order and tells the customer.
struct Dispatch;

impl<Entry> Handle<Packed, (), Outs<(Entry,)>, (), Shop> for Dispatch
where
    Entry: OutEntry<Notices, Wire: Publisher>,
{
    async fn handle(
        &self,
        packed: &Packed,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_, (), Shop>,
    ) -> Result<(), HandlerOutcome> {
        let tracking = book_courier(packed).await;
        let notice = Notice {
            customer: packed.customer.clone(),
            text: format!("Order {} is on its way: {tracking}.", packed.order),
        };
        let sent = outs.get(Notices).message(&notice).publish().await;
        sent.map_err(|_| HandlerOutcome::retry())
    }
}

路由器

发货路由器读取商店内部的主题，所以它解码 CBOR。 回复使用默认编解码器，除非它的发布位置另行指定，所以 pack 的回复指定了 CBOR。 槽位使用路由器的编解码器，除非它另行指定，所以通知指定了 JSON。 在手写形式中，这两项设置是挂载链上的步骤：threads 和 workers。

计算步骤运行在自己的线程上，应用运行时只需等待 I/O，因此商店把它放在调用 main 的线程上运行：#[ruststream::app(flavor = "current_thread")]，手写形式为 run_main_on(AppRuntime::CurrentThread, app)。后面一章中带 HTTP 的入口用 #[tokio::main(flavor = "current_thread")] 做同样的事。

宏

shipping.rs

/// The shipping side's subscribers. Both read the shop's own topics, so both decode CBOR.
pub(crate) fn routes() -> impl RouterDef<MemoryBroker, Shop> {
    Router::new()
        .with_codec(INTERNAL)
        .include(pack)
        // A reply encodes with the default codec unless its position names one.
        .out_reply(Publish)
        .codec(INTERNAL)
        .build()
        .include(dispatch)
        // A slot encodes with the router's codec unless it names one; notices stay JSON.
        .out(Notices, Publish)
        .codec(JsonCodec)
        .build()
}

手写

shipping.rs

/// The shipping side's subscribers. Both read the shop's own topics, so both decode CBOR.
pub(crate) fn routes() -> impl RouterDef<MemoryBroker, Shop> {
    Router::new()
        .with_codec(INTERNAL)
        .include(
            subscriber(Paid::DESTINATION, Pack)
                .describe("Packs a paid order into parcels.")
                // Packing is computation: two threads of its own keep it off the app's runtime.
                .threads(nonzero!(2))
                .reply()
                .build(),
        )
        // A reply encodes with the default codec unless its position names one.
        .out_reply(Publish)
        .codec(INTERNAL)
        .build()
        .include(
            subscriber(Packed::DESTINATION, Dispatch)
                .describe("Books a courier for a packed order and tells the customer.")
                // Booking waits on the courier's API, so four bookings run at once.
                .workers(nonzero!(4))
                .build(),
        )
        // A slot encodes with the router's codec unless it names one; notices stay JSON.
        .out(Notices, Publish)
        .codec(JsonCodec)
        .build()
}

每次投递的计算量达到约十微秒时，线程才划算。处理器更轻时，线程的开销超过收益。 轮询诊断能找出阻塞运行时的处理器。