主题与订阅

本指南让服务的几个部分读取同一个主题，所用的订阅有的由服务创建，有的归运维所有。

发布者向主题发送消息。 订阅是一种独立的资源，挂在一个主题上。它保留创建之后发布到该主题的每条消息。 主题的每个订阅都收到每条消息。 同一订阅的消费者分担它的消息。 服务读取的总是订阅。

示例是一个订单服务：计费、发货、审计和报表各自读取主题 orders。

每个消费者一个订阅

宏

src/main.rs

/// Billing's own subscription on `orders`: it receives every order.
#[subscriber(GooglePubSub::new("orders-billing").create_with_topic("orders"))]
async fn bill(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.amount_cents, "order billed");
    HandlerOutcome::ack()
}

/// Shipping's own subscription on the same topic: it receives every order too.
#[subscriber(GooglePubSub::new("orders-shipping").create_with_topic("orders"))]
async fn ship(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

手写

src/main.rs

/// Bills an order; mounted on billing's own subscription, it receives every order.
struct Bill;

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

/// Ships an order; mounted on shipping's own subscription, it receives every order too.
struct Ship;

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

bill 和 ship 在 orders 上各有自己的订阅，因此各自收到每个订单。 主题的新消费者就是一个新订阅，发布者一方不需要任何改动。 订阅只收到它存在之后发布的消息，所以后来加入的消费者看不到此前的订单。

GooglePubSub 用 Broker 所在项目中的短 id 指定订阅，也可以用完整的资源名称 projects/{project}/subscriptions/{id}。 create_with_topic 让服务拥有自己的拓扑：启动时，它创建缺失的主题和订阅，已经存在的则保持不动。 它创建的订阅启用消息排序，排序键一页会用到这一点。

归运维所有的订阅

在生产环境中，拓扑通常属于基础设施，在服务运行之前由 Terraform 或 gcloud 创建。 这时服务只需指定它读取什么。

宏

src/main.rs

/// A subscription operations created on `orders`; the service only reads it.
#[subscriber("orders-audit")]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order audited");
    HandlerOutcome::ack()
}

/// A subscription whose name the deployment decides, mounted with `.name(..)` below.
#[subscriber(GooglePubSub)]
async fn report(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order reported");
    HandlerOutcome::ack()
}

手写

src/main.rs

/// Audits an order; mounted on a subscription operations created.
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 audited");
        ready(Ok(()))
    }
}

/// Reports an order; mounted on the subscription the deployment names.
struct Report;

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

#[subscriber("orders-audit")] 按原样读取订阅 orders-audit。 启动时不创建任何东西。订阅不存在时，它的流返回 PubSubError::Receive 错误。 #[subscriber(GooglePubSub)] 固定订阅的种类，名称留空。 挂载点用 .name(..) 指定订阅，由部署决定名称，这里取自 REPORTS_SUBSCRIPTION。 在手写形式中，描述符在挂载处理器的地方构建，所以直接接收名称。

宏

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    let reports = env::var("REPORTS_SUBSCRIPTION").unwrap_or_else(|_| "orders-reports".to_owned());
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(bill);
            b.include(ship);
            b.include(audit);
            b.include(report.name(reports));
        },
    )
}

手写

src/main.rs

fn app() -> impl App {
    let reports = env::var("REPORTS_SUBSCRIPTION").unwrap_or_else(|_| "orders-reports".to_owned());
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(
                subscriber(
                    GooglePubSub::new("orders-billing").create_with_topic("orders"),
                    Bill,
                )
                .build(),
            );
            b.include(
                subscriber(
                    GooglePubSub::new("orders-shipping").create_with_topic("orders"),
                    Ship,
                )
                .build(),
            );
            b.include(subscriber("orders-audit", Audit).build());
            b.include(subscriber(GooglePubSub::new(reports), Report).build());
        },
    )
}

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

使用模拟器时，像运维那样，在服务启动之前创建这两个订阅：

for subscription in orders-audit orders-reports; do
  curl -X PUT localhost:8085/v1/projects/my-project/subscriptions/$subscription \
    -H 'content-type: application/json' \
    -d '{"topic": "projects/my-project/topics/orders"}'
done

服务至少启动过一次之后，主题 orders 就存在了，因为计费和发货会创建它。

服务的多个副本

服务的每个副本都在同一组订阅上打开流式拉取。 Pub/Sub 把订阅的每条消息交给其中一个副本，因此副本分担工作。 横向扩展就是启动更多副本，不需要改任何设置。 一个副本用 retry 交还的消息，可能回到另一个副本。

在一个副本内部，一个订阅一次处理一条消息。 #[subscriber] 属性上的 workers(n) 让它同时处理最多 n 条；按键分配的形式见排序键。

测试

示例文件中的测试向 orders 发布一个订单，并检查计费和发货都处理了它。 在进程内，服务只指定名称的订阅读取与它同名的主题，所以 orders-audit 和 orders-reports 的测试向这两个主题发布。 进程内 Pub/Sub 模拟了什么，见测试 Pub/Sub 服务。