使用 Python 生产 Kafka JSON 事件并检查交付

简介

控制台生产者验证了事件与 Key 的模型。应用还需要区分另一件事：调用发送 API 时，记录可能先进入客户端本地队列，之后才由 Broker 确认。如果程序在入队后就退出，可能在记录尚未交付时报告成功。

你将使用已安装的 confluent-kafka Python 客户端连接准备好的本地 Apache Kafka Broker。这个包是客户端库，无需 Confluent Cloud 账号。你将把字典序列化为 JSON/UTF-8 字节，发送订单 Key，并检查交付回调和最终队列状态。一个无法连接的本地地址将用于有时限的失败测试。

这个独立实验准备了一个三分区 Topic 和两份输入样例。课程后续活动要求你了解 Python 函数、循环、字典、JSON 和脚本运行；你可以通过 Python 学习路径补充这些基础。Kafka 安装和服务配置已准备好。请使用终端完成实验。

将字典转换为传输用的 Key 和值

这一步，你将在连接 Kafka 前检查转换过程。先查看六条事件的输入：

cd /home/labex/project

cat /home/labex/project/data/events.jsonl

每一行都是完整的 JSON 对象。json.loads 将文本转换为 Python 字典。json.dumps 将字典转换为 JSON 文本，encode("utf-8") 则将文本转换为客户端 API 可用的字节。Key 单独由 order_id 编码而来，不是 event_id。

保存并运行这个简短的预览程序。带引号的 here-document 会原样保留 Python 源码：

cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path

event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON

bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt

cat /home/labex/project/output/encoding-preview.txt

找到字典、JSON 文本和字节表示。Python 的 bytes 显示以 b 开头；这个显示标记不是传输值的一部分。第一个 Key 应是 order-4001 的编码。JSON 保留命名的事件字段，UTF-8 提供字符编码。

目前尚未发送任何事件。下一步会将这些字节提供给生产者，并检查 Broker 何时确认它们。

发布并确认每一次真实交付

这一步，你将运行一个完整的生产者，参数处理和交付回调已经提供。Producer 配置客户端；bootstrap.servers 提供初始地址。acks=all 请求当前同步副本的确认。这个单 Broker 环境只有一个副本，因此该设置不能展示多 Broker 容错能力。在分区数量不变时，提供的 murmur2_random 路由设置能让 Key 路由与之前的 Java 控制台示例保持兼容。

注意程序中的关键区别：produce 在本地将记录入队。poll 处理回调，flush 则在处理回调的同时等待队列中的工作完成。

回调是你交给客户端库的函数，交付结果可用时由客户端库调用。传入 callback=delivery 提供的是函数本身，不会立即调用它。poll(0) 无需等待就会处理已经可用的回调，flush 还会等待尚未完成的发送。回调会记录 Broker 确认的 Topic、分区、offset，或者记录失败。flush 之后，程序会检查剩余记录、失败情况和确认数量，再宣布成功。失败回调会被收集，并导致进程返回非零结果，而不是打印后就忽略。

提供的 argparse 部分将命令行选项映射为 args 属性：例如，--input 变成 args.input，--delivery-timeout-ms 变成 args.delivery_timeout_ms。Path(args.input).open(...) 打开选定的本地文件；循环从每个非空行读取一条 JSON 事件。使用这个客户端无需自行设计参数解析器。

保存完整程序。解析器和回调结构已经提供，序列化与 produce 调用会应用你刚刚检查过的字节：

cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer

parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
    parser.error("The delivery timeout must be positive.")

producer = Producer({
    "bootstrap.servers": args.bootstrap_server,
    "acks": "all",
    "message.timeout.ms": args.delivery_timeout_ms,
    "partitioner": "murmur2_random",
})
delivered = []
failures = []

def delivery(error, message):
    event_id = json.loads(message.value().decode("utf-8"))["event_id"]
    if error is not None:
        failures.append(str(error))
        print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
    else:
        delivered.append(event_id)
        print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")

sent = 0
with Path(args.input).open(encoding="utf-8") as source:
    for line in source:
        if not line.strip():
            continue
        event = json.loads(line)
        key = event["order_id"].encode("utf-8")
        value = json.dumps(event, ensure_ascii=False).encode("utf-8")
        producer.produce(args.topic, key=key, value=value, callback=delivery)
        sent += 1
        producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
    print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
    raise SystemExit(1)
print("Delivered records:", len(delivered))

PYTHON

运行生产者并保存两条输出流。> 将标准输出写入日志；2>&1 将标准错误发送到同一目的地，因此回调失败和连接诊断信息也会保留。

bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1

cat /home/labex/project/output/producer-deliveries.log

预期会看到六行 DELIVERED 回调和 Delivered records: 6。每次回调中的实际分区和 offset 可能不同。仅凭 produce 成功返回无法证明这些确认已发生。

使用已经学过的控制台消费者独立检查 Broker：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt

cat /home/labex/project/output/python-first-observed.txt

找到 py1 到 py6，确认 Key 与其 order_id 值匹配。独立读取将应用的交付报告与真实 Broker 记录关联起来；不要求固定的分区编号或全局显示顺序。

修改新事件的一个字段并复用生产者

这一步，你将复用同一个应用发送一条新事件。准备好的 next-event 样例具有新的 event_id，即 py7，因此它尚未发布。检查其 channel 字段：

cat /home/labex/project/data/next-event.jsonl

在单独的工作文件中，只将这条未发布事件的 channel 从 online 改为 store。事件仍对应 order-4001，并以 sequence 3 接在之前的事件之后：

cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT

bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1

cat /home/labex/project/output/changed-delivery.log

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt

cat /home/labex/project/output/python-final-observed.txt

预期会有一次新的成功回调，独立读取会得到七条保留事件。py7 的 channel 应为 store，而 py1 到 py6 保持原始内容。order-4001 的相关事件在路由到的同一分区内按 sequence 顺序排列；跨分区的显示顺序可能不同。

生产者实现没有改变。通过传入输入路径，它能够序列化并交付新数据，无需在程序中嵌入固定的练习答案。

观察无法连接的地址在限定时间内返回失败

这一步，你将确认程序不会将入队误认为交付。即使 localhost:1 没有 Kafka Broker，同一客户端仍可创建本地队列。提供的交付超时选项将消息完成时限设为 3000 毫秒，flush 再增加一小段有限的余量。真实 Broker 仍位于 localhost:9092。

将单事件文件发送到这个故意写错的地址。下一条命令预期会失败。

bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1

立即保存状态，以免被其他命令替换。$? 是上一条命令的退出状态：零表示成功，非零表示失败。printf 将它保存为一行文本：

printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt

cat /home/labex/project/output/unreachable-exit-status.txt

cat /home/labex/project/output/unreachable-producer.log

预期会看到非零状态、连接诊断信息、DELIVERY_FAILED 回调和 Delivery incomplete，而不是最终成功数量。即使失败消息结束后队列已经排空，程序仍会检查回调失败。

失败地址不应向真实 Topic 添加第二条 py7。之前成功发送的全部七条事件仍然可用。修正端点并重试业务事件需要单独决定，不能假定入队记录已经得到确认；本实验中有意制造的失败没有向真实 Broker 发送任何内容。接下来，你将在 Python 消费者中解码接收到的 JSON、保存数据并提交读取进度。

总结

你将字典和订单 Key 序列化为 UTF-8 字节，通过真实的 Python 客户端发布，并将成功回调与 Broker 中的实际位置匹配。修改后的新事件复用了同一程序，而无法连接的地址在限定时间内返回了非零失败，没有错误地报告成功。接下来，Python 消费者将明确展示解码、文件输出和先写入后提交的行为。