批处理

本指南一次把多条消息交给处理器，适用于按批做比逐条做更省的工作，例如向数据仓库的一次写入。

Pub/Sub 逐条投递消息，并单独结算每条消息。 服务在自己这一侧把一个订阅的投递收集成批，为这一批调用一次处理器，然后结算其中的每条消息。

示例把页面点击加载到分析数据仓库中。

批处理器

宏

src/main.rs

/// Loads a batch of clicks into the warehouse in one write. A click from a user the warehouse
/// does not know yet comes back on its own; a click without a page is dropped.
#[subscriber(
    GooglePubSub::new("clicks-loader")
        .create_with_topic("clicks")
        .batch_wait(Duration::from_millis(500))
)]
async fn load(clicks: &[Click]) -> Vec<HandlerOutcome> {
    tracing::info!(count = clicks.len(), "clicks loaded");
    clicks.iter().map(settle).collect()
}

fn settle(click: &Click) -> HandlerOutcome {
    if click.page.is_empty() {
        HandlerOutcome::drop()
    } else if click.user == UNKNOWN_USER {
        HandlerOutcome::retry()
    } else {
        HandlerOutcome::ack()
    }
}

手写

src/main.rs

/// Loads a batch of clicks into the warehouse in one write. A click from a user the warehouse
/// does not know yet comes back on its own; a click without a page is dropped.
struct Load;

impl Handle<[Click]> for Load {
    fn handle(
        &self,
        clicks: &[Click],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = clicks.len(), "clicks loaded");
        // `Err` carries one outcome per click, in the order of the batch.
        ready(Err(clicks.iter().map(settle).collect()))
    }
}

fn settle(click: &Click) -> HandlerOutcome {
    if click.page.is_empty() {
        HandlerOutcome::drop()
    } else if click.user == UNKNOWN_USER {
        HandlerOutcome::retry()
    } else {
        HandlerOutcome::ack()
    }
}

切片参数 &[Click] 让 load 成为批处理器。 描述符上的 batch_wait 是未满的批等待更多消息的时间，这里是半秒。 批一满就交出；等待结束时，批带着已有的消息交出。

load 为每次点击返回一个结果，顺序与批中一致。 来自数据仓库尚不认识的用户的点击会重试：Pub/Sub 再次投递这次点击，它单独到达，或进入后面的某一批。 没有页面的点击被丢弃，其余的得到确认。 单个 HandlerOutcome 以同样的方式结算整批。 在手写形式中，Ok(()) 确认整批，Err 逐个点击携带结果。

挂载点

宏

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("clicks", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            // At most a hundred clicks per call; a click refused five times goes to `clicks-dead`.
            b.include(load.batch(nonzero!(100)))
                .max_attempts(nonzero!(5u32))
                .dead_letter("clicks-dead");
        },
    )
}

手写

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("clicks", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            // At most a hundred clicks per call; a click refused five times goes to `clicks-dead`.
            b.include(
                subscriber(
                    GooglePubSub::new("clicks-loader")
                        .create_with_topic("clicks")
                        .batch_wait(Duration::from_millis(500)),
                    Load,
                )
                .batch(nonzero!(100))
                .build(),
            )
            .max_attempts(nonzero!(5u32))
            .dead_letter("clicks-dead");
        },
    )
}

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

batch(nonzero!(100)) 是批的大小，批处理器没有它就无法编译。 批中的每条消息都持有租约，直到整批结算，所以批的大小要小于订阅的 max_outstanding，默认 1000。 死信声明单独统计每次点击：一次点击经过五次投递后转入 clicks-dead，不论它进过多少批。 死信主题的细节见确认与重试。

测试

测试用 tokio::join! 同时发布点击，使它们进入同一批，并用 assert_batch_sizes 和 batches 读回各批。 重试的点击表现为只有一条消息的批：先是两条的批，然后陌生用户的点击单独出现四次，之后它就在 clicks-dead 上了。 测试套件的用法见测试 Pub/Sub 服务。