Kafka：完整订单生产消费 CLI
本轮仅静态审阅，以下命令与练习均未执行；预期结果不是本轮运行证据。早期真实服务验证仅为历史记录，不能代替当前环境验证。
安装与项目合同
前置：按 起步准备隔离服务并确认管理请求成功，Python 3.12 可执行。下面使用 python 表示你的 Python 3.12 命令；Windows 可用 py -3.12，历史验收使用过仓库工具链的绝对 Python 路径，本轮未执行安装或测试。依赖安装到本项目 .venv 目录，main.py 只把该目录加入模块路径，无需激活 shell 环境；不要把 .venv 提交到 Git。
txt
confluent-kafka==2.12.2sh
python -m pip install --target examples/kafka/.venv -r examples/kafka/requirements.txt
python examples/kafka/test_contract.pyconfluent-kafka 2.12.2 是基于 librdkafka 的成熟驱动，通过协议协商连接 Kafka 4.1.2；不是 Kafka 的同版本 Python 官方库。固定顶层依赖没有额外 Python 运行依赖，wheel 包含对应本机库，Windows/Linux 应各自安装 wheel，不能共享二进制目录。
全部生产消费源码
程序包含 topic 创建、生产确认、按组读取、显式提交、超时失败、连接配置和退出码。所有学习引用都指向同一份示例源码，本轮未导入或执行。
python
"""Bounded Kafka CLI. Execute from any directory; dependencies may live in .venv."""
import argparse
import json
import os
import pathlib
import sys
import time
sys.path.insert(0, os.environ.get('MESSAGING_PYTHON_DEPS', str(pathlib.Path(__file__).with_name('.venv'))))
from confluent_kafka import Consumer, Producer
from confluent_kafka.admin import AdminClient, NewTopic
def require_delivery(errors, remaining):
if errors or remaining:
raise RuntimeError(f'delivery failed: errors={errors}, remaining={remaining}')
def config(server):
return {'bootstrap.servers': server, 'socket.timeout.ms': 3000}
def produce(server, topic, key, value, partition=-1):
errors, sent = [], []
producer = Producer({**config(server), 'enable.idempotence': True,
'acks': 'all', 'message.timeout.ms': 5000})
def delivered(error, message):
if error:
errors.append(str(error))
else:
sent.append({'partition': message.partition(), 'offset': message.offset(),
'key': key, 'value': value})
producer.produce(topic, key=key, value=value, partition=partition, on_delivery=delivered)
remaining = producer.flush(8)
require_delivery(errors, remaining)
return sent[0]
def consume(server, topic, group, count, commit=False, timeout=15, sink=None):
consumer = Consumer({**config(server), 'group.id': group,
'enable.auto.commit': False, 'enable.auto.offset.store': False,
'auto.offset.reset': 'earliest', 'session.timeout.ms': 6000})
rows, messages, deadline = [], [], time.monotonic() + timeout
try:
consumer.subscribe([topic])
while len(rows) < count and time.monotonic() < deadline:
message = consumer.poll(.3)
if message is None:
continue
if message.error():
raise RuntimeError(str(message.error()))
row = {'partition': message.partition(), 'offset': message.offset(),
'key': message.key().decode() if message.key() else None,
'value': message.value().decode()}
rows.append(row)
messages.append(message)
if len(rows) != count:
raise TimeoutError(f'expected {count} messages, received {len(rows)}')
# Complete and flush the observable batch before committing any offsets.
# A commit error after output can still lead to replay; the sink must tolerate duplicates.
if sink is not None:
sink(rows)
if commit:
if sink is None:
raise ValueError('committing requires an explicit successful sink')
for message in messages:
consumer.commit(message=message, asynchronous=False)
return rows
finally:
consumer.close()
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--server', default=os.environ.get('KAFKA_SERVER', '127.0.0.1:29092'))
parser.add_argument('--topic', default='orders')
sub = parser.add_subparsers(dest='command', required=True)
sub.add_parser('create')
send = sub.add_parser('produce')
send.add_argument('--key', default='customer-1')
send.add_argument('--value', required=True)
send.add_argument('--partition', type=int, default=-1)
receive = sub.add_parser('consume')
receive.add_argument('--group', required=True)
receive.add_argument('--count', type=int, default=1)
receive.add_argument('--commit', action='store_true')
receive.add_argument('--timeout', type=float, default=15)
args = parser.parse_args()
if args.command == 'create':
admin = AdminClient(config(args.server))
futures = admin.create_topics(
[NewTopic(args.topic, num_partitions=3, replication_factor=1)], request_timeout=8)
for future in futures.values():
future.result(10)
result = {'topic': args.topic, 'partitions': 3}
elif args.command == 'produce':
result = produce(args.server, args.topic, args.key, args.value, args.partition)
else:
if args.count < 1:
parser.error('--count must be positive')
consume(args.server, args.topic, args.group, args.count, args.commit, args.timeout,
sink=lambda rows: print(json.dumps(rows, ensure_ascii=False), flush=True))
return
print(json.dumps(result, ensure_ascii=False))
if __name__ == '__main__':
try:
main()
except Exception as error:
print(f'{type(error).__name__}: {error}', file=sys.stderr)
sys.exit(1)完整运行
使用新的 topic 名称避免旧组位移影响实验；create 重跑同名 topic 预期报已存在。每次 produce 成功输出 JSON，包含 key、value、partition、offset。consume 输出实际读到的 JSON 数组。
sh
python examples/kafka/main.py --topic course-orders create
python examples/kafka/main.py --topic course-orders produce --key customer-1 --value order-created --partition 0
python examples/kafka/main.py --topic course-orders consume --group course-accounting --count 1 --commit
python examples/kafka/main.py --server 127.0.0.1:1 --topic course-orders produce --value offline全新 topic、空消费组且服务正常时，前三条预期退出码为 0，消费内容为 order-created；最后一条必须非零并报告 delivery failed。本 CLI 不吞掉连接异常，不在终端打印伪成功。生产成功只说明交付确认，不证明某个消费组已处理。
历史验证器与完整工程文件
历史运行验证器为仓库 migration/knowledge-rebuild/m4-next/messaging/verify.py，本轮不执行。它会启动、停止和重启两种服务并改写结果文件，不属于静态检查命令。
该历史验证器设计为同时验证 Kafka 与 RabbitMQ，需要先按另一个项目安装依赖。它用随机 topic/组名、显式消息数量和就绪探测，比较三个分区的原始值与 offset；验证新组独立重读、未提交重放、提交恢复和重启后四条记录仍在。结果逐命令写到 migration/knowledge-rebuild/m4-next/messaging/results.json。它创建教学数据并重启任务实例，不可对生产环境使用。
下面补齐工程其余文件；Compose 已在 起步同源展示。契约测试会导入真实驱动；批次测试会连接 broker 并创建 topic，无自动删除 topic，须使用隔离实例。这些文件均未在本轮执行。
python
import importlib.util
import pathlib
import unittest
class Contract(unittest.TestCase):
def test_delivery_error_is_not_success(self):
path = pathlib.Path(__file__).with_name('main.py')
spec = importlib.util.spec_from_file_location('lab', path)
lab = importlib.util.module_from_spec(spec)
spec.loader.exec_module(lab)
with self.assertRaises(RuntimeError):
lab.require_delivery(['broker rejected record'], 0)
with self.assertRaises(RuntimeError):
lab.require_delivery([], 1)
lab.require_delivery([], 0)
if __name__ == '__main__':
unittest.main()python
"""Live regression: an incomplete batch must not consume the group's offsets."""
import json
import pathlib
import subprocess
import sys
import unittest
import uuid
class BatchContract(unittest.TestCase):
def test_partial_timeout_has_no_commits(self):
cli = pathlib.Path(__file__).with_name('main.py')
topic = 'batch-' + uuid.uuid4().hex[:10]
def run(*args):
return subprocess.run([sys.executable, str(cli), '--topic', topic, *args],
capture_output=True, text=True, timeout=25)
self.assertEqual(run('create').returncode, 0)
self.assertEqual(run('produce', '--value', 'must-replay').returncode, 0)
partial = run('consume', '--group', topic, '--count', '2', '--commit', '--timeout', '5')
self.assertNotEqual(partial.returncode, 0)
self.assertEqual(partial.stdout, '')
replay = run('consume', '--group', topic, '--count', '1', '--commit', '--timeout', '8')
self.assertEqual(replay.returncode, 0, replay.stderr)
self.assertEqual(json.loads(replay.stdout)[0]['value'], 'must-replay')
if __name__ == '__main__':
unittest.main()md
# Kafka 4.1.2 laboratory
Static review only: no commands, tests, imports, installation or services below were executed. Earlier live results are historical evidence only.
Teaching procedure (not executed), from the repository root with Docker, Compose v2 and Python 3.12:
```sh
docker compose -p showdocs-m4b-kafka -f examples/kafka/compose.yaml up -d
python -m pip install --target examples/kafka/.venv -r examples/kafka/requirements.txt
docker exec showdocs-m4b-kafka-broker /opt/kafka/bin/kafka-topics.sh --bootstrap-server broker:9092 --list
python examples/kafka/test_contract.py
python examples/kafka/test_batch.py
python examples/kafka/main.py --topic my-orders create
python examples/kafka/main.py --topic my-orders produce --value order-1 --partition 0
python examples/kafka/main.py --topic my-orders consume --group my-reader --commit
```
`test_batch.py` requires the real broker and proves that incomplete batches do not commit offsets. `consume` waits for the requested count, flushes JSON output, then commits. Output may be repeated if a later commit fails; this is not an exactly-once sink. CLI exceptions return 1, argument errors return 2. Do not treat stdout as success unless exit status is also zero.
`compose.yaml` pins the image digest. An init container adjusts ownership of only this laboratory volume; the broker runs as its image's non-root user. Two listeners support Windows host `127.0.0.1:29092` and Linux Compose-network `broker:9092`. Use `KAFKA_SERVER` to set the default endpoint and `MESSAGING_PYTHON_DEPS` to select separately installed Linux dependencies.
For the complete two-product suite, install RabbitMQ dependencies too, then run `python migration/knowledge-rebuild/m4-next/messaging/verify.py`. Linux clients: `pwsh -File migration/knowledge-rebuild/m4-next/messaging/verify-linux.ps1`. The Linux verifier uses read-only source mounts and disposable dependency directories; it needs outbound pip package access.
课程正文入口：`docs/middleware/messaging/kafka/plus/practice/project.md`。 Stop with `docker compose -p showdocs-m4b-kafka -f examples/kafka/compose.yaml down`; adding `-v` irreversibly removes the teaching logs and group offsets. One node is not HA; restart retention is not backup or power-loss validation.
Readiness may need bounded retries (stop and inspect logs after about 80 seconds). Use fresh topic/prefix and consumer group names: existing records/commits change expected results. Expected, not observed in this review: successful produce/consume exits 0 and echoes original values; bad endpoints and empty bounded reads exit nonzero. The historical migration verifier starts/stops/restarts services and writes results: do not execute it during static review.--timeout 是重复轮询的期限，不是进程硬截止；连接、单次网络操作、提交与关闭仍可能另耗时间。CLI 的 value/body 合同是 UTF-8 文本，外部二进制或 null/tombstone 记录不在合同内，解码失败不确认位移。先按 练习自行预测结果，再核对答案。
继续阅读
- 知识专题
- 知识专题
- 知识专题