Kafka 入门实战

学习目标

完成本文件后，你能：

1. 用 Docker 快速启动一个 Kafka 环境
2. 创建 Topic、生产消息、消费消息
3. 理解 Partition 和 Consumer Group
4. 使用 kafka-console 工具进行调试

---

一、Docker 快速启动

# docker-compose.yml — 单节点 Kafka + KRaft（无需 Zookeeper）
cat > docker-compose.yml << 'EOF'
version: '3'
services:
  kafka:
    image: bitnami/kafka:latest
    ports:
      - "9092:9092"
    environment:
      # KRaft 模式（Kafka 3.3+ 自带共识，不再需要 Zookeeper）
      KAFKA_CFG_NODE_ID: 1
      KAFKA_CFG_PROCESS_ROLES: controller,broker
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
EOF

docker-compose up -d

# 验证
docker-compose logs kafka | grep "started"
# [KafkaServer id=1] started

如果在 K8s 上，直接用 Strimzi Operator 或 Bitnami Helm Chart，不必手写 StatefulSet。

二、第一个 Topic

# 进入容器
docker exec -it $(docker-compose ps -q kafka) bash

# 创建 Topic：3 个分区，2 个副本（单节点副本只能是 1）
kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic test-topic \
  --partitions 3 --replication-factor 1

# 输出: Created topic test-topic.

# 查看 Topic 详情
kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic test-topic

# Topic: test-topic  PartitionCount: 3  ReplicationFactor: 1
#   Partition: 0    Leader: 1   Replicas: 1   Isr: 1
#   Partition: 1    Leader: 1   Replicas: 1   Isr: 1
#   Partition: 2    Leader: 1   Replicas: 1   Isr: 1

# 列出所有 Topic
kafka-topics.sh --bootstrap-server localhost:9092 --list

三、生产与消费

命令行生产

# 启动生产者（输入后按 Ctrl+D 结束）
kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic test-topic

> hello kafka
> this is message 2
> message 3
^D

命令行消费

# 从头消费（--from-beginning）
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic test-topic --from-beginning

# hello kafka
# this is message 2
# message 3
# （阻塞等待新消息...）

# 带 key 和 partition 信息
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic test-topic --from-beginning \
  --property print.key=true \
  --property print.partition=true

参数	含义
--from-beginning	从头消费（重置 offset 到最早）
--group my-group	指定消费者组（不指定则自动生成随机组）
--partition 0	只消费特定分区
--max-messages 10	只取 10 条消息

四、Consumer Group 演示

# 终端 1：启动消费者（组名为 demo-group）
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic test-topic --group demo-group

# 终端 2：再启动一个同组消费者
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic test-topic --group demo-group

# 终端 3：生产消息
kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic test-topic
> msg1
> msg2
> msg3
> msg4

# 观察：终端 1 和终端 2 各自收到不同消息（负载均衡）
# 终端 1: msg1, msg3
# 终端 2: msg2, msg4

查看消费者组状态

# 列出所有消费者组
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# 查看组详情（消费进度、lag）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group demo-group --describe

# GROUP      TOPIC       PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# demo-group test-topic  0          5               10              5    ← 积压 5 条
# demo-group test-topic  1          8               8               0
# demo-group test-topic  2          3               3               0

字段	含义	关注
CURRENT-OFFSET	当前消费到哪了	—
LOG-END-OFFSET	Topic 最新消息位置	—
LAG	积压量 = LOG-END - CURRENT	> 0 说明消费跟不上

五、删除 Topic

# 标记删除
kafka-topics.sh --bootstrap-server localhost:9092 \
  --delete --topic test-topic

# 确认已删除
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep test-topic
# （无输出 = 已删除）

六、关键配置速查

配置	默认值	含义
num.partitions	1	默认 Topic 分区数
default.replication.factor	1	默认副本数（生产至少 3）
log.retention.hours	168（7天）	消息保留时长
log.segment.bytes	1GB	日志段大小，达到后滚动
message.max.bytes	1MB	单条消息最大大小
min.insync.replicas	1	最少同步副本数（可靠性关键！）

生产环境必改：min.insync.replicas = 2（3 副本时），保证至少有一个副本同步成功才算写入完成。

七、常见入门错误

错误	原因	修复
Connection to node -1 failed	Broker 未启动或端口不通	docker-compose ps 确认 + telnet localhost 9092
Topic already exists	重复创建	--list 查看已有 Topic
LEADER_NOT_AVAILABLE	Partition Leader 选举中（刚创建）	等几秒再试
消费者收不到消息	offset 已经消费过了	加 --from-beginning
Consumer Group 消费不均	Partition 数 < 消费者数	多余的消费者会空闲

自检清单

• 能用 Docker 5 分钟内启动一个 Kafka
• 能创建 Topic 并指定分区数
• 能用 console producer/consumer 收发消息
• 能创建 Consumer Group 并理解组内负载均衡
• 能通过 --describe 查看 Lag 判断积压
• 知道 min.insync.replicas 的作用

---