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

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

消费者组与崩溃恢复

本指南在工作者之间、服务之间共享一个 Redis 流，并找回停止的工作者没有处理完的条目。

示例是一个支付服务。 付款到达流 payments。 账簿组记录每笔付款，风控组审查每笔付款。

一组工作者​

src/main.rs
/// Longer than the slowest posting ever takes: an entry idle this long belongs to no live worker.
const MIN_IDLE: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    id: u64,
    account: String,
    amount_cents: u64,
}

#[subscriber(
    RedisStream::claiming("payments", MIN_IDLE).group("ledger"),
    workers(8, by_key)
)]
async fn post_payment(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let earlier_deliveries = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("0");
    if payment.account.starts_with("migrating-") {
        tracing::warn!(
            payment.id,
            earlier_deliveries,
            "account locked, payment left pending"
        );
        return HandlerOutcome::retry();
    }
    tracing::info!(payment.id, earlier_deliveries, "payment posted");
    HandlerOutcome::ack()
}

Redis 流要通过消费者组读取。 组在流中持有一个游标，游标之后的每个条目交给组内的一个消费者。 每个订阅是一个消费者，所以服务的每个运行实例都向 ledger 增加一个消费者，各实例分摊付款。 消费者得到一个生成的名字，在每个进程中依次为 ruststream-0、ruststream-1 等。 consumer 可以为它命名，例如按主机命名，这样 XINFO CONSUMERS 就能区分各个实例。

交给消费者的条目进入组的待处理条目列表。 处理器确认它之后，它才离开这个列表。 如果工作者先停止了，无论是崩溃还是部署，条目都会保持待处理：Redis 保存着它，其他消费者也不会自动收到它。

RedisStream::claiming 在一次调用中读取两部分：先是待处理至少 min_idle 的条目，然后是新条目。 因此，接收新付款的订阅也接管停止的工作者留下的付款。 它需要 Redis 8.4 或更高版本；在更旧的服务器上服务无法启动，错误会指出订阅和两个版本号。

这个订阅上的 retry 不追加副本。 条目以原有的 id 保持待处理，空闲满 min_idle 后，下一次读取会认领它。 post_payment 用这一点处理正在迁移的账户：付款在待处理列表中等待，每 30 秒回来一次。

被认领的条目带有 DELIVERY_COUNT_HEADER：这次之前的投递次数，新条目为零。 IDLE_MS_HEADER 保存条目空闲了多久。 重试、延迟与死信按这个次数限定投递。

选择 min_idle​

min_idle 没有默认值。 空闲这么久的条目被视为已被放弃，所以把它设得比处理器最长的运行时间还长。 值太短，会从仍在处理的工作者手中拿走条目，付款就会被记录两次。 值太长，崩溃后的恢复就会推迟同样的时间。

第二个组​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments.screened")]
struct Screened {
    payment_id: u64,
    suspicious: bool,
}

#[subscriber(
    RedisStream::new("payments")
        .group("fraud")
        .start_id(StreamStart::Beginning),
    reply
)]
async fn screen_payment(payment: &Payment) -> Screened {
    Screened {
        payment_id: payment.id,
        suspicious: payment.amount_cents > 1_000_000,
    }
}

每个组有自己的游标和自己的待处理列表，所以每个组都读取每个条目。 风控组收到账簿组收到的每笔付款，两者互不拖慢。 实践中，每个读取流的服务都有一个自己的组。

如果组还不存在，服务会在启动时创建它。 start_id 指定新组从哪里开始：默认的 StreamStart::New 只读取此后添加的条目，StreamStart::Beginning 读取流中仍保存的每个条目。 风控组晚于账簿组加入，所以它从开头开始，审查流中已有的付款。 已存在的组保留自己的游标，start_id 不会移动它；start_at 会。

服务​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(post_payment);
            b.include(screen_payment).out_reply(Publish);
        },
    )
}

screen_payment 在流 payments.screened 上以一个 Screened 作为回复。

按账户保持顺序​

workers(8, by_key) 在八个分区上运行 post_payment。 分区键相同的付款进入同一个分区，依次处理，所以一个账户的付款按添加的顺序记录。 Redis 没有分区，所以键放在 PARTITION_KEY_HEADER 消息头中。 生产者在发布时用 partition_key 步骤设置它。 没有键的付款进入任意分区。 顺序只在一个消费者内部成立；服务的两个实例各自记录自己分到的那部分付款。

Redis 8.4 之前的恢复​

src/main.rs
#[subscriber(RedisStream::new("payments").group("ledger"))]
async fn post_new_payment(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.id, "payment posted");
    HandlerOutcome::ack()
}

#[subscriber(RedisStream::reclaim("payments", MIN_IDLE).group("ledger"))]
async fn recover_payment(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("1");
    tracing::warn!(
        payment.id,
        deliveries,
        "payment recovered from a stopped worker"
    );
    HandlerOutcome::ack()
}

fn app_before_8_4() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(post_new_payment);
            b.include(recover_payment);
            b.include(screen_payment).out_reply(Publish);
        },
    )
}

旧版服务器没有认领读取，所以恢复要靠同一个组上的第二个订阅。 RedisStream::new 读取新条目。 RedisStream::reclaim 用 XAUTOCLAIM 只读取待处理至少 min_idle 的条目，从而回收停止的工作者持有的条目。 回收的投递会把自己计入 DELIVERY_COUNT_HEADER，所以第一次恢复报告的值至少为 1。

在 new 订阅上，retry 向流追加一份条目的副本，并确认原条目。 副本立即回来，排在期间添加的条目之后。 像上面迁移那样的等待，在这里要用 retry_after，重试、延迟与死信讲解了它。

测试恢复​

示例文件中的测试在暂停的时钟上于进程内运行服务。 其中一个测试为正在迁移的账户发布一笔付款，用 tb.advance 把时钟推进 min_idle，然后看到账簿第二次收到这笔付款，而流上没有副本。 测试 Redis 服务讲解了测试套件。