3. 交易流Kafka消费者

任务说明

本案例用 kafka-python 编写一个消费者，从 Kafka 主题 transactions 实时拉取交易 JSON 数据，逐条打印并追加写入本地文件 transactions.log，用于验证数据管道连通性。运行前需保证 Kafka 已启动、主题存在且有生产者写入数据。

环境与依赖

• Python 3.8+，安装依赖：pip install kafka-python
• 本地 Kafka（默认 localhost:9092），主题 transactions 已创建
• 创建主题示例：kafka-topics.sh --create --topic transactions --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

消费者实现

核心逻辑：订阅主题、循环拉取消息、解析 JSON、打印并落盘。使用 enable_auto_commit=True 自动提交偏移量；auto_offset_reset='latest' 表示从最新消息开始消费。收到 KeyboardInterrupt 时优雅关闭。

import json
import signal
import sys
from kafka import KafkaConsumer

TOPIC = "transactions"
BOOTSTRAP = "localhost:9092"
LOG_FILE = "transactions.log"

running = True

def stop(signum, frame):
    global running
    running = False

signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)

consumer = KafkaConsumer(
    TOPIC,
    bootstrap_servers=BOOTSTRAP,
    group_id="txn-consumer-group",
    auto_offset_reset="latest",
    enable_auto_commit=True,
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    consumer_timeout_ms=1000,
)

print(f"[consumer] subscribed topic={TOPIC} bootstrap={BOOTSTRAP}")

count = 0
with open(LOG_FILE, "a", encoding="utf-8") as f:
    while running:
        records = consumer.poll(timeout_ms=1000)
        for tp, msgs in records.items():
            for msg in msgs:
                txn = msg.value
                line = json.dumps(txn, ensure_ascii=False)
                print(f"[recv] partition={msg.partition} offset={msg.offset} data={line}")
                f.write(line + "\n")
                f.flush()
                count += 1

consumer.close()
print(f"[consumer] stopped, total received={count}, log={LOG_FILE}")

运行与验证

1. 启动 Kafka 并确认主题存在。
2. 另开终端运行生产者写入测试数据（示例）：
   echo '{"txn_id":"t1","user_id":"u1","amount":99.5,"ts":1700000000}' | kafka-console-producer.sh --topic transactions --bootstrap-server localhost:9092
3. 运行消费者：python consumer.py
4. 观察终端是否打印 [recv] ... 行，并检查 transactions.log 是否追加了对应 JSON 行。

通过条件：生产者每发送一条消息，消费者终端出现一条 [recv] 输出，且 transactions.log 中新增一行内容与发送数据一致。按 Ctrl+C 后打印 [consumer] stopped 并显示累计条数。

常见问题

• 收不到消息：确认 auto_offset_reset 与生产者发送顺序；若先启动消费者再发送，latest 可正常接收。
• 连接失败：检查 bootstrap_servers 地址与 Kafka 监听配置（advertised.listeners）。
• JSON 解析报错：确认生产者发送的是合法 JSON 字符串。