用 Pub/Sub 广播

本指南用 Redis Pub/Sub 把消息广播给服务的每个运行实例。

示例是一个网店前台。价格变动时，每个实例都刷新自己的价格缓存；它还通过模式跟踪订单事件，并宣布已发货的订单。

每个订阅者都收到消息

宏
手写

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "prices.changed")]
struct PriceChanged {
    sku: String,
    price_cents: u64,
}

#[subscriber(RedisPubSub::new("prices.changed").buffer(nonzero!(1_000)))]
async fn refresh_price(change: &PriceChanged) -> HandlerOutcome {
    tracing::info!(change.sku, change.price_cents, "price cache refreshed");
    HandlerOutcome::ack()
}

#[subscriber(RedisPubSub::new("prices.changed"))]
async fn reindex_price(change: &PriceChanged) -> HandlerOutcome {
    tracing::info!(change.sku, "search index updated");
    HandlerOutcome::ack()
}

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct PriceChanged {
    sku: String,
    price_cents: u64,
}

outgoing!(PriceChanged, "prices.changed");

struct RefreshPrice;

impl Handle<PriceChanged> for RefreshPrice {
    fn handle(
        &self,
        change: &PriceChanged,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(change.sku, change.price_cents, "price cache refreshed");
        ready(Ok(()))
    }
}

struct ReindexPrice;

impl Handle<PriceChanged> for ReindexPrice {
    fn handle(
        &self,
        change: &PriceChanged,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(change.sku, "search index updated");
        ready(Ok(()))
    }
}

RedisPubSub 订阅一个频道。发布到频道的消息到达此刻已连接的每个订阅：refresh_price 和 reindex_price 都会收到它，在每个运行实例中都是如此。消息不在订阅之间分摊，也不需要确认；处理器返回的结果不结算任何东西。

Redis 不保存副本。没有订阅连接时发布的消息就此丢失，发布给正在重启的实例的消息也一样。只有 Redis 把频道路由到订阅之后，订阅才会打开，所以服务启动之后发布的消息不会丢失。

慢订阅者

消息一发布，Redis 就推送它，不管处理器是否就绪。订阅暂存处理器还没取走的消息，上限是它的缓冲区：除非 buffer 设置了别的大小，否则是 32 条消息。缓冲区满了以后，每条新消息都会挤掉最旧的一条，订阅会记录一条警告，写明频道和丢失的数量。 refresh_price 把缓冲区设为一千条消息，所以一阵密集的价格变动会排队等它处理，而不会被截断。缓冲区在订阅打开时预留，每条消息约 600 字节，与载荷大小无关。

模式

宏
手写

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct OrderEvent {
    order_id: u64,
}

#[subscriber(RedisPubSubPattern::new("orders.events.*"))]
async fn track_order(event: &OrderEvent, Ctx(channel): Ctx<keys::Channel>) -> HandlerOutcome {
    tracing::info!(event.order_id, channel, "order event");
    HandlerOutcome::ack()
}

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct OrderEvent {
    order_id: u64,
}

outgoing!(OrderEvent);

struct TrackOrder;

impl Handle<OrderEvent, (), (), PubSubContext> for TrackOrder {
    fn handle(
        &self,
        event: &OrderEvent,
        _outs: &(),
        ctx: &mut Context<'_, PubSubContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let channel = ctx.context(keys::Channel);
        tracing::info!(event.order_id, channel, "order event");
        ready(Ok(()))
    }
}

RedisPubSubPattern 用 PSUBSCRIBE 订阅一个通配模式，这里是 orders.events. 下的每个频道。处理器通过 Ctx<keys::Channel> 从 Pub/Sub 上下文读取消息到达的频道。 keys::FromPattern 表示消息是否通过模式匹配而来。

发布到频道

宏
手写

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order_id: u64,
}

#[subscriber(
    RedisStream::new("shipments").group("notifier"),
    reply("orders.events.shipped")
)]
async fn announce_shipment(shipment: &Shipment) -> OrderEvent {
    OrderEvent {
        order_id: shipment.order_id,
    }
}

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Shipment {
    order_id: u64,
}

outgoing!(Shipment, "shipments");

struct AnnounceShipment;

impl Handle<Shipment, OrderEvent> for AnnounceShipment {
    fn handle(
        &self,
        shipment: &Shipment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<OrderEvent, HandlerOutcome>> {
        ready(Ok(OrderEvent {
            order_id: shipment.order_id,
        }))
    }
}

announce_shipment 从流中读取发货记录，并在频道 orders.events.shipped 上以一个 OrderEvent 作为回复。

宏
手写

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("storefront", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(refresh_price);
            b.include(reindex_price);
            // A glob is no channel a publish can reach, so the registration names one for copies.
            b.include(track_order)
                .out_retry(pubsub::Publish::default())
                .to("orders.events.retry");
            b.include(announce_shipment)
                .out_reply(pubsub::Publish::default());
        },
    )
}

src/main.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("storefront", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(
                subscriber(
                    RedisPubSub::new("prices.changed").buffer(nonzero!(1_000)),
                    RefreshPrice,
                )
                .build(),
            );
            b.include(subscriber(RedisPubSub::new("prices.changed"), ReindexPrice).build());
            // A glob is no channel a publish can reach, so the registration names one for copies.
            b.include(subscriber(RedisPubSubPattern::new("orders.events.*"), TrackOrder).build())
                .out_retry(pubsub::Publish::default())
                .to("orders.events.retry");
            b.include(
                subscriber(
                    RedisStream::new("shipments").group("notifier"),
                    AnnounceShipment,
                )
                .reply()
                .to("orders.events.shipped")
                .build(),
            )
            .out_reply(pubsub::Publish::default());
        },
    )
}

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

回复经由 pubsub::Publish 发出，它发送 PUBLISH。对于本服务没有任何订阅读取的名字，Broker 的默认发布者会把它当作流的键，并向它发送 XADD。模式订阅并不按名字读取 orders.events.shipped，所以注册要指定 Pub/Sub 策略。同一个服务中的模式订阅会收到这条通知。

模式没有指定可以发布副本的频道。因此，track_order 的注册用 out_retry(..).to("orders.events.retry") 指定一个频道，没有它服务无法启动。这个频道匹配该模式，所以重试的事件会回到处理器，在每个运行实例中都是如此。

集群上的分片频道

在集群上，经典的 PUBLISH 发往每个节点。 PubSubMode::Sharded 改用 SSUBSCRIBE 和 SPUBLISH：频道位于拥有其哈希槽的节点上，所以流量分散到整个集群。它需要 Redis 7 或更高版本，也不支持模式订阅。两种方式互不相通，所以订阅和发布策略要指定同一种方式：RedisPubSub::new(..).mode(PubSubMode::Sharded) 和 pubsub::Publish::new().mode(PubSubMode::Sharded)。

连接断开时

连接断开、Redis 重启或故障转移之后，Broker 会重新连接，每个 Pub/Sub 订阅都会重新订阅。断开期间发布的消息不会投递。重新分片把分片订阅的槽移到另一个节点之后，这个订阅不会恢复。

何时改用流

Pub/Sub 适合只在当下有用的内容：要刷新的缓存、实时仪表盘、发给每个实例的通知。如果每个读者都必须收到每条消息，包括它停机期间发布的消息，就使用流，并为每个读者配一个消费者组。每个组读取每个条目，条目一直留在流中，直到被裁剪。

示例文件中的测试发布一次价格变动和一次发货，并检查每个订阅都收到了发布的内容。

每个订阅者都收到消息​

慢订阅者​

模式​

发布到频道​

集群上的分片频道​

连接断开时​

何时改用流​