跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

队列与主题

本指南读取一个工作队列，在主题上广播，为暂时离开的读者保留广播，并向一个地址发布消息。

AMQP 1.0 规定了线上格式，地址的含义则交给 Broker 决定。 因此订阅要说明自己想要什么：AmqpAddress 为每种意图提供一个构造函数，链路把这个意图告诉 Broker。 本页所述内容都在 ActiveMQ Artemis 上验证过，本 crate 的测试也运行在它上面。

示例是一个订单服务。 它从队列中取出订单，并在一个主题上公布每个已接收的订单；账务、审计日志和总账都关注这个主题。

工作队列​

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    amount_cents: u64,
}

/// What every service that follows orders hears about a taken order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "order-events")]
struct OrderTaken {
    order: u64,
    amount_cents: u64,
}

#[subscriber(AmqpAddress::queue("orders"), reply)]
async fn take_order(order: &Order) -> OrderTaken {
    tracing::info!(order.id, "order taken");
    OrderTaken {
        order: order.id,
        amount_cents: order.amount_cents,
    }
}

AmqpAddress::queue 请求一个队列：链路带上 terminus 能力 queue。 Artemis 随后创建一个任播地址 orders，其后有一个队列。 一个队列的各个消费者相互竞争，每个订单只交给其中一个。 要更快地处理订单，就多运行几个服务副本：每个副本都是 orders 的又一个消费者。

take_order 用一个 OrderTaken 回复每个订单。 OrderTaken 用 #[outgoing(name = "order-events")] 指明自己的地址，所以不带参数的 reply 子句把它发往那里。

广播​

src/main.rs
#[subscriber(AmqpAddress::topic("order-events"))]
async fn bill(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, event.amount_cents, "invoice issued");
    HandlerOutcome::ack()
}

#[subscriber(AmqpAddress::topic("order-events"))]
async fn audit(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, "order recorded");
    HandlerOutcome::ack()
}

AmqpAddress::topic 请求一个主题：链路带上能力 topic。 Artemis 把 order-events 设为组播地址，并为每个订阅各建一个队列。 因此 bill 和 audit 各自收到每个 OrderTaken 的一份副本。

主题订阅与它的链路同生共存。 服务停止期间发布到 order-events 的消息不会为它保留。 不能错过任何消息的读者应读取队列。

等待读者的广播​

src/main.rs
/// Reads the queue `ledger` that the operator created on the `order-events` address.
#[subscriber(AmqpAddress::raw("order-events::ledger"))]
async fn post_to_ledger(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, "order posted to the ledger");
    HandlerOutcome::ack()
}

总账必须看到每个订单，包括它停止期间接收的订单。 运维人员在组播地址 order-events 上创建一个持久队列 ledger：

docker compose exec artemis /opt/activemq-artemis/bin/artemis queue create \
  --user artemis --password artemis --silent \
  --name ledger --address order-events --auto-create-address \
  --multicast --durable --preserve-on-no-consumers

此后发布到 order-events 的每条消息也会进入 ledger，无论 post_to_ledger 是否在运行。 order-events::ledger 是该队列在 Artemis 中的完全限定名：地址、两个冒号、队列名。 AmqpAddress::raw 按原样发送地址，不带任何能力，因此链路附加到这个队列上。 直接写 #[subscriber("order-events::ledger")] 效果相同。

应用​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(take_order).out_reply(Publish);
            b.include(bill);
            b.include(audit);
            b.include(post_to_ledger);
        },
    )
}

.out_reply(Publish) 指明回复的发布者，即 prelude 中名为 Publish 的 AmqpPublish。 发布在第一次使用时向回复的地址附加一条发送链路，并一直保留它。 Broker 拒收的消息会让发布返回 AmqpError::PublishNotAccepted 错误。

消息头与其他客户端​

常见的消息头放在 AMQP 的 properties 段中：content-type、correlation-id、reply-to 和 message-id。 其他所有消息头放在 application-properties 中。 本 crate 发出的消息是普通的 AMQP 消息，因此其他语言的客户端不借助 RustStream 也能读取它。

分区键以 PARTITION_KEY_HEADER 消息头传递，并成为消息的 group-id。 Artemis 用 group-id 实现消息分组：同一组的每条消息都交给队列的同一个消费者。 因此以客户为键的订单在多个副本之间也保持顺序。

其他 Broker 上的地址​

其他 Broker 用各自的方式为节点命名，raw 把它们的写法原样传过去。 这些写法见各自的文档： RabbitMQ 4.x、 Azure Service Bus、 ActiveMQ Classic（Amazon MQ 运行的就是它）。

测试路由​

示例文件中的测试在进程内运行应用。 一个订单发布一个 OrderTaken，orders 被调用一次，order-events 被调用两次，每个主题订阅各一次。 进程内 Broker 按原样投递 raw 地址，所以它的测试直接发布到 order-events::ledger。 测试 AMQP 服务介绍测试套件。