53-消息队列与事件流

消息队列与事件流 (Messaging and Event Streams)

面向 Rust 1.97.1 (stable, 2026-07)。本篇假设你熟悉 Go，凡有可比性处均给出 🐹 Go 对比。

热度: hot 高频 | common 常见 | occasional 偶尔 | advanced 进阶少见

本篇能解决什么：

• 你是否在 Go 里用过 Kafka / NATS / RabbitMQ，一到 Rust 就问「官方客户端是哪个」？
• 你是否分不清 Redis Pub/Sub、NATS、Kafka 各自该用在哪？
• 你是否想写 async 消费循环 + 优雅关停，却怕丢消息或重复消费？
• 你是否卡在重试、死信、幂等、背压这些「生产必问」？

术语速查表：

说明：本表覆盖本篇出现的所有专业名词与缩略词；正文首次出现时仍会就地解释一次。

热度索引：

Q1. Rust 里 Kafka 用什么？和 Go sarama 怎么对照？

Tags: hot Kafka rdkafka sarama 适用版本: rdkafka 等；以 crates.io 为准

一句话答案： 常见选择是 rdkafka（绑 librdkafka）。对标 Go 的 sarama / confluent：都是客户端库，不是标准库。生产还要自己管序列化、提交 offset、错误与再均衡。

解答：

1
2
3
4

# 示意
[dependencies]
rdkafka = { version = "0.36", features = ["cmake-build"] }
# 或按文档用动态链接 librdkafka

1
2
3
4
5

// 生产者直觉
// let producer: FutureProducer = ClientConfig::new()
//   .set("bootstrap.servers", "localhost:9092")
//   .create()?;
// producer.send(FutureRecord::to("topic").payload(&bytes), Timeout::Never).await;

1
2
3
4

fn main() {
    // 编译常依赖系统 librdkafka / cmake——CI 镜像要装好
    println!("rdkafka ≈ sarama role; ops still on you");
}

Go 对比：

1

import "github.com/IBM/sarama"

• Go 怎么做：纯 Go 客户端较多。
• Rust 为什么不同：rdkafka 走 C 库，绑定与链接是额外成本。
• Go 程序员易踩的坑：cargo add 后本机构建失败——缺原生依赖。

记忆点：

• Kafka 客户端 ≈ rdkafka。
• 链接/CI 环境要单独准备。

Q2. NATS / async-nats 最小 pub/sub？

Tags: hot NATS async-nats pubsub 适用版本: async-nats；需 Tokio

一句话答案： async-nats 连接后 publish / subscribe；适合轻量通知与请求-应答。比 Kafka 简单，比 Redis Pub/Sub 更「消息系统」一些（见 Q4）。

解答：

1
2
3

[dependencies]
async-nats = "0.37"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

1
2
3
4
5
6
7

let client = async_nats::connect("nats://127.0.0.1:4222").await?;
client.publish("events.hi", "hello".into()).await?;

let mut sub = client.subscribe("events.>").await?;
while let Some(msg) = sub.next().await {
    println!("{:?}", msg.payload);
}

1
2
3

fn main() {
    println!("NATS subject ≈ topic; wildcards events.>");
}

Go 对比：

1
2

nc, _ := nats.Connect(nats.DefaultURL)
nc.Publish("events.hi", []byte("hello"))

• API 心智非常接近。
• Go 程序员易踩的坑：在 async 里用阻塞客户端——选 async-nats。

记忆点：

• 轻量总线 → NATS + async-nats。
• subject 通配要设计好层级。

Q3. RabbitMQ / lapin 和 Go amqp 差在哪？

Tags: hot RabbitMQ lapin AMQP 适用版本: lapin；AMQP 0.9.1 直觉

一句话答案： Rust 常用 lapin 谈 AMQP：声明 exchange/queue、bind、publish、consume。和 Go amqp091-go 同模型；差异在 async API 与连接恢复要自己接。

解答：

1
2
3

[dependencies]
lapin = "2"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

1
2

// 示意：连 → channel → queue_declare → basic_publish / basic_consume
// 路由键、持久化、prefetch 与 Go 侧概念一一对应

1
2
3

fn main() {
    println!("exchange + queue + binding = Rabbit mental model");
}

Go 对比：

• 协议相同；crate 名不同。
• Go 程序员易踩的坑：以为有 database/sql 那种 std AMQP——没有。

记忆点：

• Rabbit → lapin。
• 先搞清 exchange 类型再写代码。

Q4. Redis Pub/Sub 够用吗？何时必须上 Kafka？

Tags: hot Redis Kafka 选型 适用版本: 架构选型

一句话答案： 扇出通知、允许丢、不需长期重放 → Redis Pub/Sub（见 48）或 NATS 常够。要堆积、消费组、重放、审计日志 → Kafka（或同类日志系统）。别把 Redis Pub/Sub 当任务队列。

解答：

1
2

Redis Pub/Sub：无持久化保证、慢消费者易丢
Kafka：分区有序、可重放、运维更重

1
2
3

fn main() {
    assert_ne!("pubsub", "durable log");
}

Go 对比：

• 同样选型问题；语言不改变 CAP/运维成本。
• Go 程序员易踩的坑：用 Redis list 当「简易 Kafka」却无消费语义设计。

记忆点：

• 要重放/堆积 → 日志型（Kafka）。
• 只要广播 → Pub/Sub 可能够。

Q5. 消费者组 / offset 提交直觉是什么？

Tags: hot consumer-group offset 适用版本: Kafka 心智；其它系统有近似

一句话答案： 消费者组让多实例分摊分区；offset 记录「读到哪」。先处理成功再提交（或事务性提交）——提交太早会丢，太晚会重复。Rust 客户端 API 不同，但这条语义不变。

解答：

1
2
3

partition-0 ──► consumer A
partition-1 ──► consumer B   } same group_id
partition-2 ──► consumer A

1
2
3
4

# 提交策略直觉
# at-most-once：先 commit 再处理 → 可能丢
# at-least-once：先处理再 commit → 可能重复 → 要幂等（Q8）
# exactly-once：要事务/幂等外键等，成本高

1
2
3

fn main() {
    println!("group_id + commit policy = correctness story");
}

Go 对比：

• sarama/consumer-group 同语义。
• Go 程序员易踩的坑：换语言以为语义变简单了——没有。

记忆点：

• 组内分摊；提交策略决定丢还是重。
• 默认按 at-least-once + 幂等想。

Q6. 消息正文用 JSON 还是 Protobuf/Avro？

Tags: common serde protobuf schema 适用版本: 序列化选型

一句话答案： 内部快速迭代 → JSON + serde 往往够。跨语言/强 schema/体积敏感 → Protobuf（见 45）或 Avro + Schema Registry。队列里传的是字节，契约要单独管。

解答：

1
2

serde = { version = "1", features = ["derive"] }
serde_json = "1"

1
2
3

#[derive(Serialize, Deserialize)]
struct Event { id: String, ts: i64 }
// payload = serde_json::to_vec(&ev)?;

1
2
3

fn main() {
    println!("bytes on wire; schema is a product decision");
}

Go 对比：

• encoding/json vs protobuf 同一权衡。
• Go 程序员易踩的坑：Kafka 里混用无版本字段的 JSON。

记忆点：

• 有 schema 演进需求就别纯「随便 JSON」。
• 与 36/45 交叉。

Q7. async 消费循环 + 优雅关停怎么写？

Tags: common async shutdown select 适用版本: Tokio；见 31-async

一句话答案： select! 同时等「下一条消息」和「关闭信号」（CancellationToken / ctrl_c）；退出前停止拉取、等 in-flight 处理完再提交/断开。对标 Go ctx 取消 + WaitGroup。

解答：

1
2
3
4
5
6
7
8

loop {
  tokio::select! {
    biased;
    _ = token.cancelled() => break,
    msg = consumer.recv() => { handle(msg).await; }
  }
}
// 退出：commit / close / flush producer

1
2
3
4

fn main() {
    // 信号：SIGTERM（K8s）→ 取消 token → 排空
    println!("see 31-async Q15 CancellationToken");
}

1
2

# 与 49-deployment 健康检查配合：
# 收到 SIGTERM 后 readiness 先失败，再停消费

Go 对比：

1
2
3
4

select {
case <-ctx.Done():
case msg := <-msgs:
}

• 心智相同。
• Go 程序员易踩的坑：drop 连接就走，in-flight 未完成。

记忆点：

• select! + 取消令牌。
• 关停要排空，不是直接杀进程。

Q8. 死信 / 重试 / 幂等怎么建模？

Tags: hot DLQ retry idempotency 适用版本: 业务设计；与具体 MQ 配置结合

一句话答案： 瞬时错误 → 有限次重试 + 退避；超限 → DLQ（死信队列）；业务上用 幂等键（消息 ID / 业务唯一键）防重复。Rust 不替你做这些，要写进处理器。

解答：

1
2
3
4
5
6
7

handle(msg):
  if already_processed(msg.id): ack; return
  match do_work(msg) {
    Ok => mark_processed(msg.id); ack
    Err(transient) if retries < N => nack/requeue with delay
    Err(_) => publish_to_dlq(msg); ack_original
  }

1
2
3
4

fn main() {
    let idempotency_key = "order-42-paid";
    assert!(!idempotency_key.is_empty());
}

1
2

# 存储：Redis SET NX / DB unique 约束
# 见 48-caching、43-database

Go 对比：

• 同样要自己做；框架只给钩子。
• Go 程序员易踩的坑：只靠「恰好不重复」上线。

记忆点：

• 重试有上限；失败进 DLQ。
• at-least-once ⇒ 必须幂等。

Q9. 和 gRPC 流式怎么分工？

Tags: common gRPC streaming 边界 适用版本: 架构边界

一句话答案： 同步 RPC / 强类型服务间调用 → gRPC（45）。异步解耦、多订阅者、削峰 → MQ。gRPC stream ≠ 消息队列：连接生命周期和背压模型不同。

解答：

1

下单 API ──gRPC/HTTP──► Order 服务 ──Kafka──► 库存 / 积分 / 邮件

1
2
3

fn main() {
    println!("RPC for request/response; MQ for fan-out async");
}

Go 对比：

• 同样分工。
• Go 程序员易踩的坑：用 gRPC stream 当公司级事件总线。

记忆点：

• 请求响应 → RPC；事件扩散 → MQ。

Q10. 本地怎么起 Kafka/Rabbit/NATS？

Tags: common docker local-dev 适用版本: Docker Compose 惯例

一句话答案： 开发机用 Docker Compose 起单节点；连接串写进 .env（见 44-configuration）。别在文档里假设「本机已装 broker」。

解答：

1
2
3
4
5
6

# 示意片段
services:
  nats:
    image: nats:2
    ports: ["4222:4222"]
  # kafka / rabbitmq 官方或 bitnami 镜像按团队模板

1
2

NATS_URL=nats://127.0.0.1:4222
KAFKA_BROKERS=127.0.0.1:9092

1
2
3

fn main() {
    println!("compose up ≈ local brokers; teardown between tests if needed");
}

Go 对比：

• 一样用容器。
• Go 程序员易踩的坑：CI 没起依赖导致 flaky。

记忆点：

• 本地依赖容器化。
• URL 走配置，别写死。

Q11. 背压 / 批量发送常见写法？

Tags: common backpressure batch 适用版本: 客户端配置 + 应用层

一句话答案： 生产者：限制 in-flight、批量 flush；消费者：prefetch / 并发 worker 有界。应用层用有界 channel 接业务（见 29/31）。无界缓冲 = 延迟内存爆。

解答：

1
2

source → bounded mpsc(1024) → N workers → broker
         ↑ 满则等待/丢弃策略要显式

1
2
3
4

fn main() {
    let bound = 1024usize;
    assert!(bound > 0);
}

Go 对比：

• buffered channel 同思路。
• Go 程序员易踩的坑：make(chan T, 1_000_000) 当解决方案。

记忆点：

• 有界队列 + 有界并发。
• 批量换吞吐，换来延迟与复杂。

Q12. 选型表：Redis vs NATS vs Kafka vs RabbitMQ？

Tags: common cheatsheet 选型 适用版本: 架构决策

一句话答案： 按「要不要持久/重放、路由复杂度、运维成本」选；语言只决定客户端 crate。Rust：Redis 见 48，NATS→async-nats，Kafka→rdkafka，Rabbit→lapin。

解答：

1
2
3

fn main() {
    println!("pick the semantics, then pick the crate");
}

记忆点：

• 先语义后实现。
• 本篇给地图；运维细节看各中间件文档。