2. 交易流Kafka生产者

任务说明

本案例实现一个交易流 Kafka 生产者：用 kafka-python 把模拟交易数据序列化为 JSON，实时发送到指定主题，并支持发送速率控制与失败重试。运行后可通过 Kafka 消费者或控制台消费命令验证消息是否按预期到达。

环境与依赖

• Python 3.8+
• 安装依赖：pip install kafka-python
• 本地或可访问的 Kafka 集群，默认地址 localhost:9092，主题 transactions

核心实现

生产者使用 KafkaProducer 的 value_serializer 自动把字典序列化为 JSON 字节；send 返回的 future 通过 get(timeout) 同步确认，失败时按指数退避重试；速率控制通过 rate（每秒条数）计算发送间隔实现。

import json
import time
import random
import uuid
from datetime import datetime, timezone

from kafka import KafkaProducer
from kafka.errors import KafkaError

def build_producer(bootstrap_servers="localhost:9092"):
    return KafkaProducer(
        bootstrap_servers=bootstrap_servers,
        value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
        acks="all",
        retries=0,          # 重试由业务层控制，便于观察
        linger_ms=10,
    )

def make_transaction():
    return {
        "tx_id": str(uuid.uuid4()),
        "user_id": f"u{random.randint(1000, 9999)}",
        "amount": round(random.uniform(1, 5000), 2),
        "currency": "CNY",
        "merchant": random.choice(["m_shop", "m_game", "m_travel", "m_food"]),
        "ts": datetime.now(timezone.utc).isoformat(),
    }

def send_with_retry(producer, topic, record, max_retries=5, base_delay=0.5):
    for attempt in range(1, max_retries + 1):
        try:
            future = producer.send(topic, value=record)
            meta = future.get(timeout=5)
            return meta
        except KafkaError as e:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** (attempt - 1))
            print(f"[retry {attempt}] {e}, sleep {delay:.2f}s")
            time.sleep(delay)

def run(topic="transactions", rate=5.0, total=20, bootstrap_servers="localhost:9092"):
    producer = build_producer(bootstrap_servers)
    interval = 1.0 / rate
    sent = 0
    try:
        for _ in range(total):
            record = make_transaction()
            meta = send_with_retry(producer, topic, record)
            sent += 1
            print(f"sent {sent}/{total} -> partition={meta.partition} offset={meta.offset}")
            time.sleep(interval)
    finally:
        producer.flush()
        producer.close()
    print(f"done, total sent = {sent}")

if __name__ == "__main__":
    run()

运行方式

# 1. 确保 Kafka 已启动，并创建主题（若未自动创建）
kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic transactions --partitions 3 --replication-factor 1

# 2. 启动生产者
python producer.py

# 3. 另开终端消费验证
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic transactions --from-beginning

验证方法与通过条件

• 生产者终端按 rate 设定的节奏逐条打印 sent i/total，并给出分区与偏移量。
• 消费者终端能收到与生产者条数一致的 JSON 消息，字段包含 tx_id、user_id、amount、currency、merchant、ts。
• 把 bootstrap_servers 指向不可用地址时，应看到 [retry ...] 日志并按指数退避重试，最终抛出异常而非静默丢弃。
• 调整 rate（如 1 或 20）可观察到发送间隔随之变化，说明速率控制生效。

常见问题

• 连接被拒：确认 Kafka 监听地址与 bootstrap_servers 一致，容器场景注意 advertised.listeners。
• 主题不存在：开启 auto.create.topics.enable 或手动创建主题。
• 消息乱序：多分区下同一 key 才保证分区内有序，本案例未设置 key，仅用于演示吞吐。