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

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

主题、消费者组与顺序

本指南在消费者组中读取主题，用并行的工作分区保持每个键的顺序，并按键、按编号或轮流把记录放到分区上。

示例是一家银行的账簿服务。 它把转账记入账户账簿，筛查欺诈，为自己的记录维护索引，并为有争议的转账立案。

每个账户内的顺序​

Kafka 在一个分区内保持记录的顺序。 键相同的记录落到同一个分区，所以每个键的顺序得以保持。 post 读取以账户为键的转账，最多同时处理八条，而且不打乱这个顺序。

src/main.rs
/// The account a record belongs to, carried as the Kafka record key.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct AccountKey {
    #[serde(rename = "kafka-partition-key")]
    account: String,
}

#[derive(Debug, PartialEq, Deserialize)]
struct Transfer {
    id: u64,
    amount_cents: i64,
}

/// One line of an account's ledger, keyed by the account like the transfer it came from.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "bank.entries", headers = AccountKey)]
struct Entry {
    transfer_id: u64,
    amount_cents: i64,
}

#[derive(OutSlot)]
#[publishes(Entry)]
struct Entries;

#[subscriber(
    KafkaTopic::new("transfers").lane_key(LaneKey::RecordKey),
    workers(8, by_key)
)]
async fn post(
    transfer: &Transfer,
    Headers(key): Headers<AccountKey>,
    Out(entries): Out<impl Publisher, Entries>,
) -> HandlerOutcome {
    let entry = Entry {
        transfer_id: transfer.id,
        amount_cents: transfer.amount_cents,
    };
    if entries
        .message(&entry)
        .with_headers(&key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    tracing::info!(transfer.id, account = %key.account, "transfer posted");
    HandlerOutcome::ack()
}

workers(8, by_key) 运行八个工作分区；分区键相同的投递在同一个工作分区上依次运行。 LaneKey::RecordKey 以记录键作为分区键，所以同一账户的两笔转账绝不会同时运行，即使它们在同一个分区内。 默认的 LaneKey::Partition 以来源分区作为分区键，这也覆盖了没有键的记录。 在手写形式中，同一设置是 workers_by_key。

记录键以 kafka-partition-key 消息头（PARTITION_KEY_HEADER）到达处理器。 AccountKey 是一个消息头契约，只有一个以该名字命名的字段。 Headers<AccountKey> 读取投递的键，with_headers(&key) 把它写成账簿条目的键。 于是条目在 bank.entries 中也落到其账户的分区上，账簿的读者按顺序看到每个账户。 没有键的转账不满足消息头契约，由解码策略结算，默认会丢弃它。

消费者组​

一个消费者组对每条记录只读取一次，不论该分区分配给了组里的哪个成员。 每个组都读取自己的一份。 screen 在自己的组中读取同样的转账，所以 post 看到的每笔转账它都能看到。

src/main.rs
#[subscriber(
    KafkaTopic::new("transfers")
        .group("fraud-check")
        .start(StartOffset::Earliest)
        .assignment(Assignment::CooperativeSticky)
)]
async fn screen(transfer: &Transfer) -> HandlerOutcome {
    if transfer.amount_cents > 1_000_000 {
        tracing::warn!(transfer.id, "large transfer flagged");
    }
    HandlerOutcome::ack()
}

KafkaTopic::group 指定组。 没有指定组的订阅加入 Broker 的 default_group；post 就是这样。 Kafka 只在组内读取主题，所以两者都没有的服务会在启动时失败。

StartOffset 决定组在某个分区上没有已提交的偏移量时从哪里开始。 组第一次运行时，以及保留策略删除了它的偏移量之后，都属于这种情况。 Earliest 从最旧的记录开始，Latest 从日志末尾开始，默认的 Committed 把选择交给 librdkafka 的 auto.offset.reset，其值为 latest。

Assignment 选择组在成员之间分配分区的方式。 有成员加入或离开时，CooperativeSticky 只移动换了主人的分区，其余分区继续运行。

启动会等到组把分区分配给这个成员，最多等待 assignment_timeout（默认 60 秒）。 此后，发布到该主题的每条记录都会到达订阅。 给服务的每个订阅一个独立的组。 同一个组里读取不同主题的两个订阅会互相触发再均衡：第二个加入时，第一个会短暂失去它的分区。 在组重新分配分区之前，第一个成员不读取任何记录，之后从停下的位置继续。

多个主题与模式​

KafkaTopics 通过一个消费者和一个组读取多个主题。 KafkaTopics::pattern 读取名字匹配某个正则表达式的所有主题，该表达式必须以 ^ 开头。 archive 读取银行的所有主题，并记下每条记录所在的位置。

src/main.rs
/// Where a record of one of the bank's topics sits.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "archive.index")]
struct Archived {
    topic: String,
    partition: i32,
    offset: i64,
}

#[subscriber(KafkaTopics::pattern("^bank\\..*").group("archive"), reply)]
async fn archive(
    _record: &serde_json::Value,
    Ctx(topic): Ctx<Topic>,
    Ctx(partition): Ctx<Partition>,
    Ctx(offset): Ctx<Offset>,
) -> Archived {
    Archived {
        topic: topic.to_string(),
        partition,
        offset,
    }
}

处理器从上下文读取投递的坐标：Topic、Partition 和 Offset。 所有匹配的主题共用一个载荷类型；archive 解码任意 JSON。 启动后创建的主题，会在 librdkafka 下一次刷新元数据时加入模式。

KafkaPartitions 读取它指定的分区，不加入任何组。 它从不再均衡，适合每个分区一个进程的部署，或检查单个分区的工具。

把记录放到分区上​

open_case 把一笔有争议的转账变成 bank.cases 上的一个案件。 该主题有三个分区：0 和 1 平分普通案件，分区 2 存放紧急案件。

src/main.rs
/// The partition of `bank.cases` kept for urgent cases.
const URGENT: i32 = 2;

#[derive(Debug, PartialEq, Deserialize)]
struct Dispute {
    transfer_id: u64,
    urgent: bool,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "bank.cases")]
struct Case {
    transfer_id: u64,
}

#[derive(OutSlot)]
#[publishes(Case)]
struct Cases;

#[subscriber(KafkaTopic::new("card-disputes").group("case-intake"))]
async fn open_case(
    dispute: &Dispute,
    Out(cases): Out<impl Publisher<Options = KafkaOptions>, Cases>,
) -> HandlerOutcome {
    let case = Case {
        transfer_id: dispute.transfer_id,
    };
    let opened = if dispute.urgent {
        cases.message(&case).partition(URGENT).publish().await
    } else {
        cases.message(&case).publish().await
    };
    if opened.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

partition(n) 把一条记录固定到某个分区。 它是 KafkaPublishSteps 的一个步骤，所以槽位要以 Publisher<Options = KafkaOptions> 为约束才能提供它。 固定的分区优先于记录键，记录键优先于 librdkafka 的分区器。 主题中不存在的分区会让发布失败。

应用​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(KafkaBroker::new(["localhost:9092"]).default_group("ledger"))
}

fn service(broker: KafkaBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(broker, |b| {
        b.include(post).out(Entries, Publish::default()).build();
        b.include(screen);
        // A set of topics has no single topic to send a retry copy to, so the mount site names it.
        b.include(archive)
            .out_retry(Publish::default())
            .transform(ToSourceTopic);
        // Partitions 0 and 1 share the regular cases evenly; partition 2 stays for urgent ones.
        b.include(open_case)
            .out(Cases, Publish::default())
            .transform(RoundRobin::partitions(2))
            .build();
    })
}

RoundRobin::partitions(2) 把每个没有指定分区的案件依次放到分区 0 和 1 中的下一个。 librdkafka 把没有键的记录放到随机分区上，这会让一些消费者比另一些更忙；轮转则把记录均匀地分开。 带键或固定了分区的记录保持原来的位置。 数量是固定的：它等于轮转应使用的分区数，绝不超过主题拥有的分区数。

读取一组主题的订阅没有唯一的主题可以接收重试副本，所以应用用 ToSourceTopic 指定一个；重试与死信主题讲解了这一点。

示例文件中的测试在进程内运行应用。 进程内 Kafka 给每个主题一个分区，所以检查多分区放置的测试要连接集群运行：

KAFKA_TEST_URL=localhost:9092 cargo test