Kafka 是一个高性能、分布式、可扩展的消息队列系统，常被用于：
日志收集
用户行为埋点
异步任务削峰
订单事件流转
数据同步
流式计算
微服务解耦
大数据实时处理
简单理解：Kafka 就像一个高性能的“消息中转站”。
生产者把消息写进去，消费者从里面读取消息，中间通过 Topic 进行分类。
例如一个电商系统：
用户下单
↓
订单服务发送消息到 Kafka
↓
库存服务消费消息，扣减库存
↓
支付服务消费消息，生成支付单
↓
通知服务消费消息，发送短信/邮件
这样做的好处是：订单服务不需要直接调用库存、支付、通知服务，各个系统通过 Kafka 解耦。
一、Kafka 核心概念速览
在学习 Kafka 之前，必须先理解以下概念。
1. Topic（主题）
Topic 是消息的分类。例如：
order_topic 订单消息
user_topic 用户消息
log_topic 日志消息
payment_topic 支付消息
生产者和消费者通过 Topic 进行消息交互。
2. Producer（生产者）
负责将消息发送到指定的 Topic。比如订单服务就是生产者。
3. Consumer（消费者）
负责从 Topic 中读取消息。比如库存服务、支付服务、通知服务都是消费者。
4. Broker（Kafka 服务器节点）
Kafka 是分布式的，每个服务节点称为一个 Broker。例如：
broker-1: 192.168.1.10:9092
broker-2: 192.168.1.11:9092
broker-3: 192.168.1.12:9092
5. Partition（分区）
一个 Topic 可以分成多个 Partition，以提升并发和吞吐量。例如：
order_topic
├── partition-0
├── partition-1
└── partition-2
6. Offset（偏移量）
消息在分区中的唯一位置标识。消费者消费到哪里，就是通过 Offset 记录的。
partition-0:
offset=0 message-1
offset=1 message-2
offset=2 message-3
offset=3 message-4
7. Consumer Group（消费者组）
多个消费者可以组成一个消费者组。
例如 order_topic 有 3 个分区：
二、典型应用场景
1. 异步解耦
订单服务不再直接调用下游，而是发送消息到 Kafka，下游服务各自消费，系统耦合度更低。
2. 削峰填谷
高并发请求先写入 Kafka，后端消费者按自己能力慢慢处理，避免压垮数据库。
秒杀场景：100000 请求到达 → 写入 Kafka → 库存服务按每秒 1000 条处理。
3. 日志采集
应用服务 → Kafka → Flink / Elasticsearch / ClickHouse。
4. 数据同步
用户信息修改后，用户服务 → Kafka → 搜索服务、推荐系统、数据仓库同步数据。
三、Python 操作 Kafka 的库选择
常用两个库：
1. kafka-python
安装：pip install kafka-python
纯 Python 实现，简单易用，适合入门学习和中小型项目，性能一般。
2. confluent-kafka
安装：pip install confluent-kafka
基于 C 库 librdkafka，性能高、稳定性好，更适合生产环境。API 稍复杂，但本文也会详细讲解。
本文会同时展示两个库的用法：用 kafka-python 入门，用 confluent-kafka 讲解生产级用法。
四、环境准备：用 Docker 启动 Kafka
docker-compose.yml：
version: "3.8"
services:
zookeeper:
image: bitnami/zookeeper:3.9
container_name: zookeeper
ports:
- "2181:2181"
environment:
- ALLOW_ANONYMOUS_LOGIN=yes
kafka:
image: bitnami/kafka:3.7
container_name: kafka
ports:
- "9092:9092"
environment:
- KAFKA_BROKER_ID=1
- KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
- KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
- ALLOW_PLAINTEXT_LISTENER=yes
depends_on:
- zookeeper
启动并验证：
docker-compose up -d
dockerps
进入容器创建 Topic：
dockerexec-it kafka bash
kafka-topics.sh \
--create\
--topic order_topic \
--bootstrap-server 127.0.0.1:9092 \
--partitions3\
--replication-factor 1
查看 Topic 列表与详情：
kafka-topics.sh --list --bootstrap-server 127.0.0.1:9092
kafka-topics.sh --describe--topic order_topic --bootstrap-server 127.0.0.1:9092
输出：
Topic: order_topic Partition: 0 Leader: 1 Replicas: 1 Isr: 1
Topic: order_topic Partition: 1 Leader: 1 Replicas: 1 Isr: 1
Topic: order_topic Partition: 2 Leader: 1 Replicas: 1 Isr: 1
五、kafka-python 快速上手
5.1 最简单的生产者
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=["127.0.0.1:9092"])
topic ="order_topic"
message ="hello kafka"
producer.send(topic, message.encode("utf-8"))
producer.flush()
print("消息发送成功:", message)
输出：
注意：Kafka 发送的是 bytes，字符串需要编码。
5.2 发送 JSON 消息（带序列化器）
import json
from kafka import KafkaProducer
producer = KafkaProducer(
bootstrap_servers=["127.0.0.1:9092"],
value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8")
)
order_data ={
"order_id":"ORDER_10001",
"user_id":9527,
"product_name":"MacBook Pro",
"amount":12999.00,
"status":"CREATED"
}
producer.send("order_topic", order_data)
producer.flush()
print("订单消息发送成功:", order_data)
输出：
订单消息发送成功: {'order_id':'ORDER_10001', 'user_id':9527, ...}
5.3 获取发送结果（Topic、Partition、Offset）
future = producer.send(topic, order_data)
record_metadata = future.get(timeout=10)
print("topic:", record_metadata.topic)# order_topic
print("partition:", record_metadata.partition)# 1
print("offset:", record_metadata.offset)# 0
5.4 最简单的消费者
from kafka import KafkaConsumer
consumer = KafkaConsumer(
"order_topic",
bootstrap_servers=["127.0.0.1:9092"],
auto_offset_reset="earliest",
enable_auto_commit=True,
group_id="order_consumer_group"
)
formessagein consumer:
print(f"topic={message.topic}, partition={message.partition}, offset={message.offset}")
print("value:", message.value.decode("utf-8"))
输出示例：
topic: order_topic, partition: 1, offset: 0
value: {"order_id":"ORDER_10002", "user_id":10086, "amount":299.99, "status":"PAID"}
5.5 带 JSON 反序列化的消费者
consumer = KafkaConsumer(
"order_topic",
bootstrap_servers=["127.0.0.1:9092"],
auto_offset_reset="earliest",
enable_auto_commit=True,
group_id="order_json_group",
value_deserializer=lambda v: json.loads(v.decode("utf-8"))
)
formessagein consumer:
order = message.value
print(f"订单号: {order['order_id']}, 金额: {order['amount']}")
5.6 消息 Key 与分区有序性
Kafka 中，相同 Key 的消息会被路由到同一个分区，从而保证该 Key 下的消息顺序。
发送带 Key 的消息：
producer = KafkaProducer(
bootstrap_servers=["127.0.0.1:9092"],
key_serializer=lambda k: k.encode("utf-8"),
value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8")
)
orders =[
{"order_id":"ORDER_20001", "status":"CREATED"},
{"order_id":"ORDER_20001", "status":"PAID"},
{"order_id":"ORDER_20001", "status":"SHIPPED"},
{"order_id":"ORDER_20002", "status":"CREATED"},
{"order_id":"ORDER_20002", "status":"CANCELLED"},
]
fororderin orders:
key = order["order_id"]
future = producer.send("order_topic", key=key, value=order)
metadata = future.get(timeout=10)
print(f"发送成功 order_id={key}, status={order['status']}, partition={metadata.partition}, offset={metadata.offset}")
输出：
发送成功 order_id=ORDER_20001, status=CREATED, partition=2, offset=0
发送成功 order_id=ORDER_20001, status=PAID, partition=2, offset=1
发送成功 order_id=ORDER_20001, status=SHIPPED, partition=2, offset=2
发送成功 order_id=ORDER_20002, status=CREATED, partition=0, offset=0
发送成功 order_id=ORDER_20002, status=CANCELLED, partition=0, offset=1
ORDER_20001 的所有消息都在分区 2，ORDER_20002 都在分区 0，这就是 Key 的核心价值。
六、消费者组与 Offset 管理
6.1 消费者组演示
一个 Topic 有 3 个分区，启动 3 个消费者并使用相同的 group_id="order_group_demo"，Kafka 会自动分配分区：
consumer = KafkaConsumer(
"order_topic",
bootstrap_servers=["127.0.0.1:9092"],
group_id="order_group_demo",
auto_offset_reset="earliest",
enable_auto_commit=True,
value_deserializer=lambda v: json.loads(v.decode("utf-8"))
)
同时运行多个实例，不同实例会看到不同分区的消息。消费者组内是负载均衡，组间是广播。
6.2 auto_offset_reset
注意：只在该消费者组尚未任何 offset 提交时生效。若想从头重新消费，需换一个新的 group_id 或手动重置 offset。
6.3 自动提交 vs 手动提交
自动提交（enable_auto_commit=True）简单，但存在风险：
- 消费者拉取消息后自动提交 offset → 业务处理失败 → 重启后该消息丢失（不会再消费）。
手动提交（enable_auto_commit=False）更可靠：
consumer = KafkaConsumer(
"order_topic",
bootstrap_servers=["127.0.0.1:9092"],
group_id="order_manual_group",
auto_offset_reset="earliest",
enable_auto_commit=False,
value_deserializer=...
)
formessagein consumer:
try:
order = message.value
# 模拟业务处理
if order.get("status")=="ERROR":
raise Exception("订单状态异常")
print("处理成功:", order["order_id"])
consumer.commit()# 处理成功再提交
except Exception as e:
print("处理失败，不提交 offset:", e)
失败的消息不会被提交，下次可重新消费，避免丢失。
七、进阶：批量消费、重试与死信队列
7.1 批量消费（poll）
使用 poll() 代替迭代器，可以一次拉取多条消息，提高吞吐：
while True:
records = consumer.poll(timeout_ms=1000, max_records=10)
if not records:
continue
for tp, messages in records.items():
formsgin messages:
print(f"处理 offset={msg.offset}, value={msg.value}")
consumer.commit()
7.2 异常重试机制
max_retry =3
success = False
forretryin range(1, max_retry+1):
try:
process_order(order)
success = True
break
except Exception as e:
print(f"第{retry}次失败: {e}")
time.sleep(1)
if success:
consumer.commit()
else:
print("超过最大重试次数，不提交 offset")
但这种方式如果一直失败且不提交 offset，重启后会重复消费该坏消息。因此需要死信队列。
7.3 死信队列（Dead Letter Queue）
当消息处理失败超过最大重试次数后，将其写入专门的“死信 Topic”，再提交 offset，从而避免阻塞后续消费。
创建死信 Topic：
kafka-topics.sh --create--topic order_dead_letter_topic --bootstrap-server 127.0.0.1:9092 --partitions3 --replication-factor 1
业务代码：
import json
from kafka import KafkaConsumer, KafkaProducer
consumer = KafkaConsumer("order_topic", ..., enable_auto_commit=False)
producer = KafkaProducer(
bootstrap_servers=["127.0.0.1:9092"],
value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8")
)
formessagein consumer:
order = message.value
# ... 重试逻辑 ...
if not success:
dead_msg ={
"original_topic": message.topic,
"original_partition": message.partition,
"original_offset": message.offset,
"original_value": order,
"error_message": str(last_error),
"retry_count": max_retry
}
producer.send("order_dead_letter_topic", dead_msg)
producer.flush()
consumer.commit()# 提交 offset，不再阻塞
print("消息已写入死信队列")
这样既避免了“坏消息”阻塞，也保留了问题消息供后续排查。
八、生产级库 confluent-kafka 与封装实践
8.1 基础 Producer
import json
from confluent_kafka import Producer
conf ={"bootstrap.servers":"127.0.0.1:9092"}
producer = Producer(conf)
def delivery_report(err, msg):
if err:
print("发送失败:", err)
else:
print(f"发送成功 topic={msg.topic()}, partition={msg.partition()}, offset={msg.offset()}")
order ={"order_id":"ORDER_40001", "user_id":10001, "amount":599.00, "status":"CREATED"}
producer.produce(
topic="order_topic",
key=order["order_id"],
value=json.dumps(order, ensure_ascii=False).encode("utf-8"),
callback=delivery_report
)
producer.flush()
8.2 基础 Consumer
from confluent_kafka import Consumer
consumer = Consumer({
"bootstrap.servers":"127.0.0.1:9092",
"group.id":"confluent_order_group",
"auto.offset.reset":"earliest",
"enable.auto.commit": False
})
consumer.subscribe(["order_topic"])
while True:
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
print("异常:", msg.error())
continue
value = json.loads(msg.value().decode("utf-8"))
print(f"收到消息 offset={msg.offset()}, value={value}")
consumer.commit(msg)
8.3 封装 KafkaProducerClient
项目中不应到处直接创建 Producer，建议封装为统一的服务类：
import json
import logging
from confluent_kafka import Producer
class KafkaProducerClient:
def __init__(self, bootstrap_servers):
self.producer = Producer({
"bootstrap.servers": bootstrap_servers,
"client.id":"python-order-producer",
"acks":"all",
"retries":3,
"linger.ms":10,
"batch.num.messages":1000
})
def _delivery_report(self, err, msg):
if err:
logging.error("发送失败: %s", err)
else:
logging.info("发送成功 topic=%s partition=%s offset=%s",
msg.topic(), msg.partition(), msg.offset())
def send_json(self, topic, key, value):
try:
self.producer.produce(
topic=topic,
key=str(key).encode("utf-8")if key else None,
value=json.dumps(value, ensure_ascii=False).encode("utf-8"),
callback=self._delivery_report
)
self.producer.poll(0)
except BufferError:
self.producer.flush()
except Exception as e:
logging.exception("发送异常: %s", e)
raise
def close(self):
self.producer.flush()
8.4 封装 KafkaConsumerClient
class KafkaConsumerClient:
def __init__(self, bootstrap_servers, group_id, topics):
self.consumer = Consumer({
"bootstrap.servers": bootstrap_servers,
"group.id": group_id,
"auto.offset.reset":"earliest",
"enable.auto.commit": False,
"max.poll.interval.ms":300000,
"session.timeout.ms":10000
})
self.consumer.subscribe(topics)
def process_message(self, value):
# 业务处理逻辑
pass
def start(self):
try:
while True:
msg = self.consumer.poll(1.0)
if msg is None or msg.error():
continue
value = json.loads(msg.value().decode("utf-8"))
self.process_message(value)
self.consumer.commit(msg)
except KeyboardInterrupt:
pass
finally:
self.consumer.close()
九、实战：自动化测试平台的任务调度
一个典型的自动化测试平台流程：
用户点击执行 → Django 后端创建任务 → 写入 Kafka → Worker 消费任务 → 执行测试 → 结果入库 → 通知用户。
测试任务消息格式：
{
"task_id":"TASK_202605110001",
"project":"cloud-platform",
"env":"test",
"executor":"vito",
"test_type":"api",
"case_ids":[1001, 1002, 1003],
"priority":"high",
"created_at":"2026-05-11 10:40:00"
}
Django 配置 (settings.py)：
KAFKA_CONFIG ={
"BOOTSTRAP_SERVERS":"127.0.0.1:9092",
"TEST_TASK_TOPIC":"test_task_topic",
}
Kafka 客户端 (kafka_client.py)：
from confluent_kafka import Producer
from django.conf import settings
import json
class KafkaClient:
def __init__(self):
self.producer = Producer({"bootstrap.servers": settings.KAFKA_CONFIG["BOOTSTRAP_SERVERS"]})
def send_message(self, topic, key, value):
self.producer.produce(
topic=topic,
key=str(key).encode("utf-8"),
value=json.dumps(value, ensure_ascii=False).encode("utf-8"),
callback=self.delivery_report
)
self.producer.poll(0)
@staticmethod
def delivery_report(err, msg):
if err:
print("失败:", err)
else:
print(f"成功 topic={msg.topic()}, partition={msg.partition()}, offset={msg.offset()}")
def flush(self):
self.producer.flush()
视图 (views.py)：
def create_test_task(request):
task ={
"task_id": f"TASK_{uuid.uuid4().hex[:12].upper()}",
"project":"cloud-platform",
...
}
client = KafkaClient()
client.send_message(settings.KAFKA_CONFIG["TEST_TASK_TOPIC"], task["task_id"], task)
client.flush()
return JsonResponse({"code":0, "message":"测试任务创建成功", "data": task})
Worker 消费任务并执行：
consumer = Consumer({
"bootstrap.servers":"127.0.0.1:9092",
"group.id":"test_worker_group",
"auto.offset.reset":"earliest",
"enable.auto.commit": False
})
consumer.subscribe(["test_task_topic"])
while True:
msg = consumer.poll(1.0)
if msg is None or msg.error():
continue
task = json.loads(msg.value().decode("utf-8"))
# 执行 pytest / playwright ...
result ={"task_id": task["task_id"], "total":3, "passed":3, "failed":0}
print("测试完成:", result)
consumer.commit(msg)
十、消息规范与项目结构建议
推荐的消息格式：
{
"message_id":"MSG_202605110001",
"event_type":"ORDER_CREATED",
"version":"1.0",
"trace_id":"TRACE_ABC123456",
"created_at":"2026-05-11 11:20:00",
"data":{
"order_id":"ORDER_90001",
"user_id":10001,
"amount":299.99,
"status":"CREATED"
}
}
message_id：幂等性判断
event_type：路由不同处理逻辑
version：兼容消息升级
trace_id：全链路追踪
created_at：排查延迟
data：实际业务数据
推荐的项目结构：
kafka_demo_project/
├── config/settings.py # Kafka 配置
├── kafka_client/
│ ├── __init__.py
│ ├── producer.py
│ ├── consumer.py
│ └── serializers.py
├── services/
│ ├── order_service.py
│ └── test_task_service.py
├── workers/
│ └── test_task_worker.py
├── logs/
├── requirements.txt
└── main.py
requirements.txt：
十一、运维常用命令与排查手段
Topic 管理：
kafka-topics.sh --list --bootstrap-server 127.0.0.1:9092
kafka-topics.sh --create--topic test_topic --bootstrap-server 127.0.0.1:9092 --partitions3 --replication-factor 1
kafka-topics.sh --describe--topic test_topic --bootstrap-server 127.0.0.1:9092
命令行生产/消费：
kafka-console-producer.sh --topic test_topic --bootstrap-server 127.0.0.1:9092
kafka-console-consumer.sh --topic test_topic --bootstrap-server 127.0.0.1:9092 --from-beginning
消费进度查看：
kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --list
kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe--group order_group_demo
输出中的 LAG 表示积压量，如果持续增大，表示消费跟不上。
十二、生产环境核心配置速查
Producer 推荐配置
Consumer 推荐配置
十三、最佳实践总结
Producer：
Consumer：
Topic 设计：
十四、常见问题解答
1. Kafka 为什么这么快？
顺序写磁盘、零拷贝、批量发送、分区并行、页缓存。
2. 能保证不丢消息吗？
配置合理可以接近不丢：acks=all，多副本，手动提交 offset。
3. 如何保证消息顺序？
只保证同一分区内有序，使用业务 Key（如订单 ID）将相同业务的消息发往同一分区。
4. 为什么会出现重复消费？
消费者处理成功但提交失败、崩溃、Rebalance 等。解决核心是幂等设计 + 手动提交。
5. 消息积压怎么办？
检查 LAG、消费者状态、下游数据库性能，增加分区和消费者实例，采用批量消费优化。
十五、结语
Kafka 在 Python 项目中的应用非常广泛，尤其适合异步任务、事件驱动、日志采集和测试平台调度等场景。
入门：掌握 Producer / Consumer、Topic / Partition / Offset、消费者组、Offset 提交。
进阶：理解消息顺序、重复消费与幂等、死信队列、批量消费、生产级封装。
生产：优选 confluent-kafka，注重配置、幂等、封装与监控。
最终记住 Kafka 的核心思想：
**生产者只负责把消息可靠地写进去，消费者按自己的能力慢慢处理，系统之间通过消息解耦。
**
