Kafka相关

Kafka 分布式消息队列 完整系统学习大纲

本文面向测试工程师及后端开发初学者。Kafka 是互联网公司异步架构的事实标准：下单后发券、支付后通知、日志采集、数据同步……背后几乎都有消息队列。而对测试来说，MQ 是最容易”看起来通过、实际上出问题”的环节——消息丢了没人知道、重复消费导致重复扣款、顺序错了业务直接乱套、积压几百万条拖垮下游。这些正是中高级面试的高频场景。 文中命令基于 Kafka 3.x / 4.x（KRaft 模式），Windows 用户请在 PowerShell（非 cmd）中执行。

一、Kafka 概述

1.1 什么是 Kafka？

• 定义：Kafka 是一个分布式、高吞吐、可持久化的发布订阅消息系统（官方定位是”分布式事件流平台”）。
• 历史：2011 年由 LinkedIn 开发并开源，2012 年进入 Apache 孵化器；因能支撑百万级 TPS 且数据可持久化，迅速成为大数据与微服务领域的标配。
• 核心思想：把消息当成只追加（append-only）的日志文件顺序写入磁盘，用”顺序读写 + 零拷贝 + 批量压缩”换极致吞吐，用”分区 + 副本”换水平扩展与高可用。

1.2 消息队列解决什么问题？

作用	说明	举例
解耦	生产者只管把消息丢进队列，不关心谁消费、消费几次	订单系统发”支付成功”事件，优惠券、积分、物流各自订阅
异步	主流程不等下游处理完就返回	注册成功后异步发欢迎邮件，接口响应从 800ms 降到 50ms
削峰填谷	突发流量先堆在队列里，下游按自己的节奏消费	秒杀 10 万请求涌入，库存服务每秒只处理 2000 条
可恢复	下游挂了消息不丢，恢复后继续消费	积分服务宕机 30 分钟，重启后从 offset 继续处理
广播	一份数据多个系统各取所需	埋点日志同时供实时计算、离线数仓、监控告警使用

一句话：消息队列的本质是”用异步换性能、用冗余换可靠”。它带来性能提升的同时，也引入了丢失、重复、顺序、延迟四类新的测试风险。

1.3 Kafka vs 其他主流 MQ

维度	Kafka	RabbitMQ	RocketMQ
定位	高吞吐日志 / 事件流	企业级消息代理	业务消息（阿里系）
单机吞吐	十万 ~ 百万级 TPS	万级 TPS	十万级 TPS
消息模型	发布订阅 + 分区并行消费	复杂路由（Exchange/Queue）	发布订阅 + 事务消息
消息堆积能力	极强（磁盘顺序写，堆积上亿条无压力）	较弱（内存为主，堆积会触发流控）	强
延迟	毫秒 ~ 几十毫秒	微秒 ~ 毫秒（最低）	毫秒级
顺序保证	分区内严格有序	单队列有序	队列有序
事务 / 延时消息	支持事务；延时需自己实现	支持（插件）	原生支持延时消息、事务消息
运维复杂度	中（依赖调优项较多）	低	中
典型场景	日志采集、大数据管道、事件驱动	业务异步、任务队列、复杂路由	电商交易、金融业务消息

一句话：要吞吐和堆积选 Kafka，要低延迟和复杂路由选 RabbitMQ，要事务消息/延时消息选 RocketMQ。测试工程师至少要知道被测系统用的是哪一个，因为三者的”丢失/重复”表现完全不同。

1.4 核心概念（务必先记牢）

概念	类比	说明
Broker	快递分拣中心	一个 Kafka 服务节点；多台 Broker 组成集群
Topic	快递品类（如”生鲜”）	消息的逻辑分类，生产者往 Topic 发，消费者从 Topic 收
Partition（分区）	同一品类下的多条流水线	一个 Topic 分成多个分区，分区是并行度和顺序性的最小单位
Offset	流水线上的编号	消息在分区内的唯一自增序号，消费者靠它记录消费进度
Producer	寄件人	发送消息的一方
Consumer	收件人	拉取消息的一方（Kafka 是拉模型，消费者主动 pull）
Consumer Group	收件团队	多个消费者组成一个组，组内分摊分区，组间互不影响
Replica	备份仓库	分区的副本，分 Leader（对外服务）与 Follower（同步备份）
ISR	“跟得上”的备份名单	In-Sync Replicas，与 Leader 保持同步的副本集合
Controller / KRaft	调度中心	集群管理节点；Kafka 4.0 起完全用 KRaft 取代 ZooKeeper

最关键的三个关系：

1. Topic 是逻辑概念，Partition 才是物理存储：消息实际存在分区的日志文件里。
2. 分区内有序，分区间无序：全局顺序需要特殊设计（见 3.8）。
3. 消费者组内的消费者数量不要超过分区数：多出来的消费者会一直空闲（见 3.4）。

1.5 整体架构

                          ┌──────────────────────────────┐
      Producer  ────────▶ │          Topic: order        │
      (订单服务)           │  ┌────────┬────────┬────────┐ │
                          │  │ P0     │ P1     │ P2     │ │  ← 3 个分区
                          │  │Leader  │Leader  │Leader  │ │
                          │  └───┬────┴───┬────┴───┬────┘ │
                          └──────┼────────┼────────┼──────┘
                                 │        │        │   ▲ 副本同步（Follower）
                    ┌────────────┘        │        └────────────┐
                    ▼                     ▼                     ▼
             Consumer Group A （订单处理组）
             ├─ C1 → 消费 P0
             ├─ C2 → 消费 P1        Consumer Group B （数据分析组）
             └─ C3 → 消费 P2        └─ C4 → 消费 P0+P1+P2（全量）

      Broker 1 ─ Broker 2 ─ Broker 3        KRaft Controller（元数据管理）

• Producer → Topic → Partition：消息落到一个具体分区的日志末尾。
• Consumer Group 内：一个分区同一时刻只能被组内一个消费者消费（保证不重复处理）。
• Consumer Group 间：各自维护自己的 offset，互不影响 = 天然广播。

1.6 典型应用场景

1. 异步解耦：支付成功后发”支付完成”事件，券、积分、物流、通知各自订阅。
2. 日志采集：应用日志 → Kafka → ELK / 数仓，是 Kafka 最经典的用途。
3. 削峰：大促流量先入队，下游按能力消费。
4. 事件溯源 / CQRS：用事件流重建业务状态。
5. 实时计算：Kafka → Flink / Spark Streaming 实时指标。
6. 测试造数与流量回放：把线上流量录到 Kafka，再回放给测试环境（很实用的测试技巧）。

二、环境搭建与快速上手

2.1 Docker 一键启动（推荐，KRaft 单机模式）

Kafka 官方镜像 apache/kafka 从 3.7 起提供，无需再单独起 ZooKeeper：

# docker-compose.yml
services:
  kafka:
    image: apache/kafka:4.0.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

docker compose up -d
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

⚠️ KAFKA_ADVERTISED_LISTENERS 是最容易踩的坑：它写的是”客户端真正能连到的地址”。填 localhost 则只能本机访问；填容器名则只能容器网络内访问。消费者连不上、一直报 Node -1 disconnected，八成是这个配错了。

2.2 常用管理命令

# 创建 Topic（3 分区，1 副本）
kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic order-topic --partitions 3 --replication-factor 1

# 查看 Topic 列表 / 详情
kafka-topics.sh --bootstrap-server localhost:9092 --list
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic order-topic

# 修改分区数（只能增不能减！）
kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic order-topic --partitions 6

# 删除 Topic
kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic order-topic

2.3 命令行收发消息（测试必备）

# 控制台生产者（回车发送一条，Ctrl+C 退出）
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic order-topic

# 带 key 发送（决定进哪个分区，验证顺序性时必用）
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic order-topic \
  --property "parse.key=true" --property "key.separator=:"

# 控制台消费者（从最新开始）
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic order-topic

# 从头消费（排查"消息是不是真发出去了"时用）
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic order-topic --from-beginning

# 指定消费组消费（会把 offset 提交到该组）
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic order-topic \
  --group test-group --from-beginning

2.4 消费组与积压（Lag）查看

# 查看消费组列表
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# 查看某个组的消费进度（重点看 LAG 列）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-consumer-group

# 重置 offset（重放消息，测试"重复消费"场景常用）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-consumer-group --topic order-topic --reset-offsets \
  --to-earliest --execute

输出示例：

GROUP                TOPIC        PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID
order-consumer-group order-topic  0          1200            1350            150   consumer-1-xxx
order-consumer-group order-topic  1          980             980             0     consumer-2-xxx

一句话：LAG = LOG-END-OFFSET − CURRENT-OFFSET，就是”还有多少条没消费”。压测和线上巡检最先看的指标就是它。

三、核心工作原理

3.1 分区与存储：为什么 Kafka 这么快？

一个分区在磁盘上就是一个只追加的日志文件，消息带 offset 顺序写入：

Partition 0 的日志文件（segment 分段存储）
┌────────────────────────────────────────────────────┐
│ offset 0 │ offset 1 │ offset 2 │ ... │ offset 9999 │ ← 顺序追加写
└────────────────────────────────────────────────────┘
   ↑ .log 数据文件        ↑ .index 稀疏索引（加速定位 offset）

快的四个原因：

手段	说明
顺序写磁盘	避免随机写的磁头寻道，顺序写甚至比随机写内存还快
零拷贝（sendfile）	数据从磁盘直接进网卡，绕过用户态多次拷贝
批量发送 + 压缩	多条消息打包成一个 batch，减少网络往返
分区并行	多分区分布在不同 Broker，读写天然并行

3.2 生产者发送流程

send(record)
   ↓ 序列化（key/value → byte[]）
   ↓ 分区器 Partitioner（决定进 P0/P1/P2）
   ↓ 进入 RecordAccumulator 缓冲区（按分区+批次攒着）
   ↓ Sender 线程批量拉取，组装成 ProduceRequest
   ↓ 网络发送到 Broker Leader
   ↓ Broker 落盘 → 返回 ack（含 offset）
   ↓ 失败则按 retries 重试

3.3 分区策略

策略	触发条件	说明
指定分区	发送时显式给了 partition	直接发到该分区
按 key 哈希	指定了 key	hash(key) % 分区数，同 key 一定进同一分区（保证同 key 顺序）
粘性分区（默认）	无 key 无分区	Kafka 2.4+ 默认：先粘住一个分区攒批，攒够再换，吞吐更高
自定义 Partitioner	配置 partitioner.class	业务自定义路由（如按省份分分区）

// 指定 key 发送：保证同一订单的消息进同一分区 → 保证该订单的顺序
kafkaTemplate.send("order-topic", orderId, orderEvent);

3.4 消费者组与分区分配

Topic 有 3 个分区（P0/P1/P2）

情况1：Group A 只有 1 个消费者  → C1 消费 P0+P1+P2
情况2：Group A 有 3 个消费者    → C1→P0, C2→P1, C3→P2（最优）
情况3：Group A 有 4 个消费者    → 前 3 个各分一个，C4 空闲（浪费！）
情况4：Group A 有 3 个消费者，挂了 1 个 → 触发 Rebalance，重新分配

• 组内：分区被分摊，一条消息组内只被消费一次（负载均衡）。
• 组间：不同组各自独立消费全量，一条消息可以被多个组各消费一次（广播）。
• Rebalance：消费者增减或订阅 Topic 变化时触发，期间整个组暂停消费（生产上要避免频繁 Rebalance）。

⚠️ 分区数决定了消费并行度的上限。分区数太少 → 加再多消费者也没用；分区数改大后，同 key 的消息可能落到不同分区，原有顺序性会被破坏。

3.5 Offset 提交：消费进度记在哪？

消费者消费完要提交 offset，告诉 Kafka”这条我处理过了”。

提交方式	配置	语义	风险
自动提交	enable.auto.commit=true（默认）	每 5 秒提交一次拉取到的最大 offset	先提交后处理 → 处理失败则消息丢失
手动同步提交	enable.auto.commit=false + commitSync()	处理完再提交	处理成功但提交前宕机 → 重复消费
手动异步提交	commitAsync()	同上但不阻塞	吞吐高，失败不重试

// 推荐：业务处理成功后再提交（Spring Kafka 的 manual ack 模式）
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
    try {
        orderService.handle(record.value());   // 1. 先处理业务
        ack.acknowledge();                     // 2. 成功后才提交 offset
    } catch (Exception e) {
        log.error("消费失败，不提交 offset，等待重试", e);
    }
}

3.6 副本与高可用

Partition 0（replication-factor=3）
┌──────────────┐   同步    ┌──────────────┐
│  Leader      │ ────────▶ │  Follower    │  ← ISR 成员
│  Broker 1    │ ────────▶ │  Broker 2    │
└──────────────┘           └──────────────┘
       │                    ┌──────────────┐
       └───────────────────▶│  Follower    │  ← 落后太多，被踢出 ISR
                            │  Broker 3    │
                            └──────────────┘

• Leader：负责该分区的所有读写；Follower：只做数据同步备份。
• ISR（In-Sync Replicas）：与 Leader 差距在阈值内的副本集合；Leader 挂了只从 ISR 里选新 Leader。
• 关键配置：min.insync.replicas=N + 生产者 acks=all → 至少 N 个副本写成功才返回，这是防丢失的核心组合。
• unclean.leader.election.enable：默认 false，禁止从非 ISR 副本选主（选了会丢数据）。生产环境不要改成 true。

3.7 数据保留与清理

配置	默认值	说明
retention.ms / retention.hours	168 小时（7 天）	消息保留时长，到期删除
retention.bytes	-1（不限）	分区日志最大体积
log.cleanup.policy	delete	delete 按时间删；compact 按 key 保留最新值
log.retention.check.interval.ms	300000	清理检查周期

注意：Kafka 的消息不会因为被消费而删除，它按时间/大小保留。这也是 Kafka 能”重放消息”的原因——测试时想重复消费历史数据，重置 offset 就行。

3.8 顺序性：到底能不能保证？

场景	是否有序	说明
单分区	严格有序	分区内 offset 递增，先写先读
多分区、无 key	无序	消息被分散到不同分区
多分区、同 key	同 key 有序	同 key 路由到同一分区
消费者多线程处理	可能乱序	拉取有序，但线程池并发处理会打乱

保证顺序的三种做法：

1. Topic 只建 1 个分区（最简单，但牺牲并行度）。
2. 按业务 key 分区（推荐，如订单号、用户 ID），保证同一实体的顺序。
3. 消费端按 key 分发到单线程 / 内存队列处理，避免并发乱序。

四、可靠性：丢失、重复、顺序、积压

这一章是测试工程师最该吃透的部分。MQ 的 Bug 基本都出在这里，而且往往要等到线上对账才发现。

4.1 消息为什么会丢？三个环节逐个堵

环节	丢失原因	防护措施
生产者 → Broker	acks=0 不等确认就返回；网络抖动未重试	acks=all、retries>0、enable.idempotence=true
Broker 自身	Leader 写完后宕机，副本没同步就选了新主	replication-factor≥3、min.insync.replicas≥2、unclean.leader.election.enable=false
Broker → 消费者	自动提交 offset 后业务处理失败	关闭自动提交，业务成功后手动提交

生产者 acks 参数对照：

acks	行为	吞吐	可靠性
0	不等任何确认，”发了就算”	最高	最低（可能丢）
1（默认）	Leader 写入即确认	中	中（Leader 宕机且未同步 → 丢）
all / -1	ISR 中所有副本写入才确认	较低	最高

# 生产环境生产者推荐配置（可靠性优先）
acks=all
retries=3
enable.idempotence=true
max.in.flight.requests.per.connection=5
min.insync.replicas=2      # Broker 侧配置

4.2 为什么会重复消费？怎么防？

根因：

1. 消费成功但提交 offset 失败（宕机、超时）→ 下次从旧 offset 重新消费。
2. Rebalance 导致 offset 未及时提交。
3. 生产者重试：开启重试后，Broker 已写入但 ack 丢失，重试导致写入两条。

三层防护：

层级	手段	说明
生产者	enable.idempotence=true	单分区单会话幂等，靠 sequence number 去重（Kafka 0.11+）
生产者	事务（transactional.id）	跨分区原子写，配合消费者 isolation.level=read_committed
消费者（最关键）	业务幂等	唯一键去重、数据库 INSERT ... ON DUPLICATE KEY UPDATE、Redis SETNX、状态机判断

一句话：Kafka 只能保证”不丢”，”不重复”必须由业务侧幂等来兜底。这是面试必答点，也是测试必验点。

4.3 三种投递语义

语义	含义	实现要点	适用场景
At most once（至多一次）	可能丢，绝不重复	先提交 offset 再处理	允许丢失的日志、埋点
At least once（至少一次）	绝不丢，可能重复	先处理再提交 offset（Kafka 默认）	绝大多数业务 + 消费端幂等
Exactly once（精确一次）	不丢不重	幂等生产者 + 事务 + read_committed	金融、支付、对账

4.4 消息积压（Lag）怎么排查？

积压 = 生产速度 > 消费速度
   ↓
查 Lag：kafka-consumer-groups.sh --describe --group xxx
   ↓
定位瓶颈：
  ├─ 分区数太少 → 并行度不够（最常见！）
  ├─ 消费逻辑太慢 → 单条处理耗时长（查 DB、调外部接口）
  ├─ 消费者线程数不足 → 但别超过分区数
  ├─ 频繁 Rebalance → 消费反复暂停
  └─ 下游依赖故障 → 消费一直失败重试
   ↓
处理：加分区 + 加消费者 / 批量消费 / 异步化耗时逻辑 / 临时扩容

应急手段：写个”转发程序”，把积压 Topic 的消息批量搬到 N 倍分区的临时 Topic，启 N 倍消费者快速消化。

4.5 消费失败与死信队列

• 无限重试会阻塞整个分区（Kafka 没有原生重试队列，失败会一直重试同一条）。
• 正确做法：重试 N 次后投递到死信队列（DLT，Dead Letter Topic），主流程继续，死信单独告警和人工处理。

// Spring Kafka：3 次重试后进入死信 Topic（原 Topic 名 + ".DLT"）
@RetryableTopic(attempts = "3", backoff = @Backoff(delay = 1000, multiplier = 2))
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void listen(String message) {
    orderService.handle(message);
}

@DltHandler
public void handleDlt(String message) {
    log.error("进入死信队列，需人工处理：{}", message);
}

五、Spring Boot 集成

本站已有 Spring Boot 篇作基础，这里只给 Kafka 集成的最小可用配置。

5.1 依赖与配置

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

# application.properties
spring.kafka.bootstrap-servers=localhost:9092

# 生产者
spring.kafka.producer.acks=all
spring.kafka.producer.retries=3
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.properties.enable.idempotence=true

# 消费者
spring.kafka.consumer.group-id=order-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

# 监听器：手动提交 + 批量
spring.kafka.listener.ack-mode=manual
spring.kafka.listener.type=batch
spring.kafka.consumer.max-poll-records=500

5.2 生产者

@RestController
@RequiredArgsConstructor
public class OrderController {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @PostMapping("/order")
    public String createOrder(@RequestBody Order order) {
        // 用 orderId 作 key → 同一订单进同一分区 → 保证顺序
        kafkaTemplate.send("order-topic", order.getOrderId(), JSON.toJSONString(order))
                .addCallback(
                        result -> log.info("发送成功 partition={} offset={}",
                                result.getRecordMetadata().partition(),
                                result.getRecordMetadata().offset()),
                        ex -> log.error("发送失败", ex)
                );
        return "ok";
    }
}

5.3 消费者

@Component
@Slf4j
public class OrderConsumer {

    @KafkaListener(topics = "order-topic", groupId = "order-group", concurrency = "3")
    public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        try {
            for (ConsumerRecord<String, String> record : records) {
                log.info("收到消息 partition={} offset={} key={}",
                        record.partition(), record.offset(), record.key());
                orderService.handle(record.value());
            }
            ack.acknowledge();     // 全部处理成功后批量提交
        } catch (Exception e) {
            log.error("批量消费失败，不提交 offset", e);
            throw e;               // 交给重试/死信机制
        }
    }
}

5.4 常见配置项速查

配置	作用
spring.kafka.consumer.auto-offset-reset	无 offset 时从哪开始：earliest（最早）/ latest（最新，默认）/ none（报错）
spring.kafka.listener.ack-mode	record/batch/manual/manual_immediate 等提交策略
spring.kafka.listener.concurrency	消费线程数（不要超过分区数）
spring.kafka.consumer.max-poll-records	单次 poll 最大条数
spring.kafka.consumer.max-poll-interval-ms	两次 poll 最大间隔，超时触发 Rebalance

六、Python 操作 Kafka（测试脚本）

6.1 安装

pip install kafka-python
# 或性能更好、官方 C 库封装的客户端
pip install confluent-kafka

6.2 生产者与消费者

from kafka import KafkaProducer, KafkaConsumer
import json

BOOTSTRAP = "localhost:9092"
TOPIC = "order-topic"

def build_producer() -> KafkaProducer:
    return KafkaProducer(
        bootstrap_servers=BOOTSTRAP,
        key_serializer=lambda k: k.encode("utf-8"),
        value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
        acks="all",           # 等所有 ISR 副本确认
        retries=3,
        linger_ms=10,         # 攒 10ms 再发，提高吞吐
        batch_size=16384,
    )

def build_consumer(group_id: str) -> KafkaConsumer:
    return KafkaConsumer(
        TOPIC,
        bootstrap_servers=BOOTSTRAP,
        group_id=group_id,
        auto_offset_reset="earliest",   # 新组从头消费
        enable_auto_commit=False,       # 手动提交，便于测试控制
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
        consumer_timeout_ms=5000,       # 拉不到消息 5s 后退出
    )

if __name__ == "__main__":
    producer = build_producer()
    for i in range(10):
        producer.send(TOPIC, key=f"order-{i}", value={"order_id": f"order-{i}", "amount": 100 + i})
    producer.flush()
    print("发送完成")

    consumer = build_consumer("test-script-group")
    for msg in consumer:
        print(f"partition={msg.partition} offset={msg.offset} key={msg.key} value={msg.value}")

6.3 测试造数技巧

import uuid, random, time
from kafka import KafkaProducer

def generate_orders(count: int = 10000):
    """批量造订单消息，用于压测或功能验证"""
    producer = KafkaProducer(
        bootstrap_servers="localhost:9092",
        value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
        compression_type="gzip",
    )
    for i in range(count):
        producer.send("order-topic", value={
            "order_id": str(uuid.uuid4()),
            "user_id": random.randint(1, 1000),
            "amount": round(random.uniform(10, 5000), 2),
            "timestamp": int(time.time() * 1000),
        })
    producer.flush()

七、测试视角：MQ 到底怎么测（重点）

7.1 测试点清单

类型	测试点	验证方法
功能	消息能正确发送 / 接收	生产一条，消费端断言内容一致
功能	字段完整、序列化正确	对比发送的 JSON 与消费到的 JSON
功能	多消费者组广播	两个组同时消费，都收到全量
可靠性	消息不丢失	发 N 条 → 消费者全部处理完 → 对账条数
可靠性	重复消费幂等	同一条消息消费两次，断言业务只生效一次
可靠性	顺序性	同 key 发 1~N，断言消费顺序为 1~N
可靠性	消息积压恢复	停消费者 → 累积消息 → 重启 → Lag 归零且无丢失
异常	Broker 宕机 / 重启	杀掉 Broker，验证生产者重试与消费者重连
异常	消费失败重试与死信	构造消费异常，验证重试次数与死信 Topic
异常	消息体过大	发送超限消息，验证有明确报错且不拖垮服务
性能	吞吐量与 Lag	压测脚本 + 监控 Lag 是否持续增长
安全	敏感信息脱敏	检查消息体内有无明文密码、身份证

7.2 消息丢失测试

def test_no_message_loss(producer, group_id, expected_count=1000):
    """发送 N 条，验证消费者能完整收到 N 条"""
    unique_tag = str(uuid.uuid4())
    for i in range(expected_count):
        producer.send("test-topic", value={"tag": unique_tag, "seq": i})
    producer.flush()

    received = []
    consumer = KafkaConsumer(
        "test-topic",
        bootstrap_servers=BOOTSTRAP,
        group_id=group_id,
        auto_offset_reset="earliest",
        enable_auto_commit=False,
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
        consumer_timeout_ms=10000,
    )
    for msg in consumer:
        if msg.value.get("tag") == unique_tag:
            received.append(msg.value["seq"])

    assert len(received) == expected_count, f"消息丢失：期望 {expected_count}，实际 {len(received)}"
    assert set(received) == set(range(expected_count)), "消息缺失或重复"

关键做法：用唯一 tag 标记本轮测试数据，避免被历史消息干扰；最后按集合比对能同时发现丢失和重复。

7.3 幂等（重复消费）测试

def test_idempotent_consume(order_service, message):
    """同一条消息处理两次，业务结果应只生效一次"""
    order_service.handle(message)
    first_balance = order_service.get_balance(message["user_id"])

    order_service.handle(message)      # 故意重复消费
    second_balance = order_service.get_balance(message["user_id"])

    assert first_balance == second_balance, "重复消费导致重复扣款！"
    assert order_service.count_orders(message["order_id"]) == 1, "订单被重复创建"

必测场景：

1. 同一条消息连续消费两次。
2. 重置 offset 后重新消费历史消息（用 2.4 的 --reset-offsets --to-earliest）。
3. 消费者重启后从上次 offset 继续，验证边界那条是否被重复处理。

7.4 顺序性测试

def test_order_by_key(producer):
    """同一 key 的消息必须按顺序消费"""
    tag, key = str(uuid.uuid4()), "order-fixed-key"
    group = f"pytest-order-{tag}"
    for seq in range(1, 51):
        producer.send("order-topic", key=key, value={"tag": tag, "seq": seq})
    producer.flush()

    consumer = KafkaConsumer(
        "order-topic",
        bootstrap_servers=BOOTSTRAP,
        group_id=group,
        auto_offset_reset="earliest",
        enable_auto_commit=False,
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
        consumer_timeout_ms=10000,
    )
    seqs = [msg.value["seq"] for msg in consumer if msg.value.get("tag") == tag]
    consumer.close()

    assert seqs == list(range(1, 51)), f"同 key 顺序错乱：{seqs[:20]}"

注意：只有同 key 才保证有序。如果被测代码发消息时没带 key，那”顺序性”这个需求本身就不成立，属于设计缺陷，要提 Bug 而不是硬测。

7.5 积压测试

# 1. 停掉消费者，用生产者脚本灌 50 万条
python generate_orders.py --count 500000

# 2. 观察 Lag 增长
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group order-consumer-group

# 3. 重启消费者，观察 Lag 下降速度（消费速率），计算预计追平时间
watch -n 5 'kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-consumer-group | grep -v PARTITION'

# 4. 断言：最终 LAG == 0 且业务侧处理条数 == 发送条数

验收标准建议：给出明确的业务指标，例如”10 万条积压需在 5 分钟内追平，且期间不允许出现消息丢失”。

7.6 异常场景测试

场景	操作	预期
消费者宕机	kill 消费进程	消息堆积在 Topic；重启后从 offset 继续，不丢不重
消费者处理变慢	消费端 sleep 5s	Lag 上涨；恢复后追平；不触发频繁 Rebalance
Broker 宕机	停掉一台 Broker（多副本时）	分区 Leader 切换；生产者重试成功；无消息丢失
网络抖动	tc netem 模拟丢包延迟	生产者重试生效；消费者重连成功
消息体超限	发 10MB 消息	报 RecordTooLargeException；有明确日志；服务不崩
下游依赖故障	Mock 数据库超时	消费失败进重试 → 死信队列；主流程不被阻塞
重复启动同名消费组	起两个同 group.id 的实例	触发 Rebalance；分区重新分配；无重复消费

7.7 pytest 自动化示例（可直接复用）

# test_kafka_mq.py
import json, uuid, pytest, time
from kafka import KafkaProducer, KafkaConsumer

BOOTSTRAP = "localhost:9092"
TOPIC = "test-order-topic"


@pytest.fixture(scope="module")
def producer():
    p = KafkaProducer(
        bootstrap_servers=BOOTSTRAP,
        key_serializer=lambda k: k.encode("utf-8"),
        value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
        acks="all",
        retries=3,
    )
    yield p
    p.close()


def new_consumer(group_id: str) -> KafkaConsumer:
    return KafkaConsumer(
        TOPIC,
        bootstrap_servers=BOOTSTRAP,
        group_id=group_id,
        auto_offset_reset="latest",
        enable_auto_commit=False,
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
        consumer_timeout_ms=8000,
    )


def wait_messages(group_id: str, expect: int, timeout: int = 30):
    """轮询拉取，直到拿到 expect 条或超时"""
    consumer = new_consumer(group_id)
    result, deadline = [], time.time() + timeout
    try:
        while len(result) < expect and time.time() < deadline:
            for msg in consumer:
                result.append(msg.value)
                if len(result) >= expect:
                    break
    finally:
        consumer.close()
    return result


def test_produce_and_consume(producer):
    """用例1：基本收发，字段保持一致"""
    tag = str(uuid.uuid4())
    group = f"pytest-{tag}"
    payload = {"tag": tag, "order_id": "SO001", "amount": 99.9}

    producer.send(TOPIC, key="SO001", value=payload).get(timeout=10)

    msgs = [m for m in wait_messages(group, 1) if m.get("tag") == tag]
    assert msgs, "未消费到消息"
    assert msgs[0]["order_id"] == "SO001"
    assert msgs[0]["amount"] == 99.9


def test_no_loss_under_batch(producer):
    """用例2：批量 500 条，验证不丢失不重复"""
    tag = str(uuid.uuid4())
    group = f"pytest-{tag}"
    count = 500
    for i in range(count):
        producer.send(TOPIC, key=f"key-{i}", value={"tag": tag, "seq": i})
    producer.flush()

    seqs = [m["seq"] for m in wait_messages(group, count) if m.get("tag") == tag]
    assert len(seqs) == count, f"消息丢失：期望 {count} 实际 {len(seqs)}"
    assert sorted(seqs) == list(range(count)), "消息重复或乱序"


def test_same_key_ordering(producer):
    """用例3：同 key 消息消费顺序一致"""
    tag, key = str(uuid.uuid4()), "fixed-key"
    group = f"pytest-{tag}"
    for seq in range(1, 31):
        producer.send(TOPIC, key=key, value={"tag": tag, "seq": seq})
    producer.flush()

    seqs = [m["seq"] for m in wait_messages(group, 30) if m.get("tag") == tag]
    assert seqs == list(range(1, 31)), f"同 key 顺序错乱：{seqs}"

运行：

pip install pytest kafka-python
pytest test_kafka_mq.py -v

7.8 测试 MQ 的常见坑

坑	现象	解决
消费组 ID 重复	两个测试脚本用同一 group，互相抢消息	每次测试生成唯一 group id（如用 uuid）
auto-offset-reset 用错	新组从 latest 开始，历史消息”收不到”	测试造数后消费，用 earliest
断言太早	刚 send 就去消费，异步还没到	用轮询 + 超时等待，别 sleep 死等
没 flush	脚本退出后消息还在缓冲区，实际没发出去	生产者结束前 producer.flush()
消息被其他组消费掉	误以为丢消息	用 --describe 确认已有消费组，区分是”消费过”还是”丢失”
改了分区数破坏顺序	原本有序的场景开始乱序	回归测试时确认分区数未被变更
环境共用 Topic	多人同时测一个 Topic 数据互相干扰	测试用独立 Topic 前缀，如 test_xxx

八、性能优化与监控

8.1 吞吐优化手段

方向	参数	说明
生产者攒批	batch.size（默认 16KB）、linger.ms（默认 0）	调大 linger.ms 到 5~100ms 可显著提升吞吐（牺牲一点延迟）
压缩	compression.type=snappy/lz4/gzip	减少网络传输；lz4 性价比最高
缓冲区	buffer.memory（默认 32MB）	攒批空间，满了会阻塞 send
并行度	分区数	分区越多并行越高，但文件句柄和 Rebalance 成本上升
消费批量	max.poll.records	一次拉多条，批量入库
副本同步	replica.fetch.max.bytes	副本同步跟不上会导致 ISR 频繁伸缩

8.2 压测命令

# 生产者压测：发 100 万条，每条 1KB
kafka-producer-perf-test.sh \
  --topic perf-topic \
  --num-records 1000000 --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers=localhost:9092 acks=1 linger.ms=10 compression.type=lz4

# 消费者压测
kafka-consumer-perf-test.sh \
  --bootstrap-server localhost:9092 --topic perf-topic \
  --messages 1000000 --show-detailed-stats

8.3 必看的监控指标

指标	含义	告警建议
Consumer Lag	未消费消息数	持续增长或超过阈值即告警（最重要）
Under Replicated Partitions	副本未同步的分区数	应长期为 0
ISR 收缩次数	副本频繁掉出 ISR	频繁收缩说明 Broker 负载或网络有问题
消息生产 / 消费 TPS	吞吐	与基线对比，突降要排查
请求处理延迟	Produce/Fetch 耗时	P99 突增需关注
Broker 磁盘、JVM GC	基础资源	磁盘 80% 预警

九、常见问题排查（FAQ）

1. 消费不到消息怎么办？

按顺序排查：

1. Topic 名是否写错（大小写、环境前缀）。
2. 消费组是否已经有 offset：用 --describe --group 看 CURRENT-OFFSET 是否已接近 LOG-END-OFFSET。
3. auto.offset.reset 是 latest 还是 earliest：新组默认从最新开始，历史消息不会被消费。
4. 生产者是否真的发送成功：用 kafka-console-consumer.sh --from-beginning 确认 Topic 里有数据。
5. advertised.listeners 是否配置正确：表现为能连上但拉不到数据或一直重连。

2. 出现重复消费怎么定位？

• 先看是不是业务侧重复投递（生产者重试导致发了两条）：对比两条消息的 offset 是否不同。
• 再看是不是 offset 未提交：消费者重启后从旧 offset 消费，同一条被处理两次。
• 最后确认消费端幂等是否失效：去重键是否正确、并发下是否有竞态。
• 结论：Kafka 层面只能尽量减小，彻底解决靠业务幂等。

3. Lag 一直涨怎么处理？

1. 确认消费速率是否低于生产速率（--describe 连续观察）。
2. 分区数是否成为瓶颈（消费者数 > 分区数则多余的空闲）。
3. 消费逻辑是否有慢调用（DB 查询、外部 HTTP）。
4. 是否频繁 Rebalance（看消费者日志 rebalance 关键字）。
5. 应急：临时扩容消费者 + 批量消费；必要时转发到高分区临时 Topic。

4. 分区数怎么定？

经验公式：分区数 ≈ 目标吞吐 / 单分区吞吐，单分区吞吐一般按 几千 ~ 1 万条/秒 估算。 注意三点：分区只能增不能减；分区过多会增加文件句柄、Rebalance 耗时；改分区会破坏同 key 的顺序性。 建议一开始就留出余量（如 6、12），而不是后期频繁调整。

5. Kafka 如何保证不丢消息？（面试必背）

三个环节一起说：

• 生产者：acks=all + retries>0 + 幂等生产者。
• Broker：replication-factor≥3 + min.insync.replicas≥2 + 关闭 unclean.leader.election。
• 消费者：关闭自动提交，业务处理成功后再手动提交 offset。
• 补充：Exactly Once 需要事务 + isolation.level=read_committed；真正的”不重复”靠业务幂等。

十、学习路径建议

第一阶段（能跑起来）
  概念 → Docker 起服务 → 命令行收发 → 看懂分区/消费组/offset

第二阶段（能排查问题）
  原理（存储/副本/ISR）→ 可靠性三件事（丢失/重复/顺序）→ Lag 排查

第三阶段（能用于测试）
  Spring Boot 集成 → Python 脚本造数与断言 → pytest 自动化 MQ 用例

第四阶段（能调优）
  压测与参数调优 → 监控告警 → 线上问题定位

一句话总结：Kafka 的学习重点不在”怎么发消息”，而在”消息会不会丢、会不会重、会不会乱、会不会堵”。把这四类问题的成因和对策讲清楚，Kafka 就算过关了。