理论学再多，不如上手敲代码！前面我们又是搭集群，又是讲概念，今天终于到了激动人心的编程实战环节！

我们将使用 Python 这个“万金油”语言，连接到我们亲手搭建的 Kafka 4.0 KRaft 集群，编写生产者（Producer）发送消息，再编写消费者（Consumer）接收消息。

更重要的是，我们会穿插一个 “开发运维扯皮角” 环节，专门聊聊那些在实际工作中，开发和运维因为 Kafka 问题经常battle的场景，以及如何从根源上解决它们。

准备工作：安装 Python Kafka 库

我们需要一个强大的 Python 库来和 Kafka 交互。kafka-python 是一个非常流行的选择。

pip install kafka-python

生产者（Producer）：把消息送上车

生产者负责将业务消息发送到指定的 Topic。下面是一个简单但包含了最佳实践的生产者代码。

# producer.pyfrom kafka import KafkaProducerimport jsonimport time# --- 最佳实践：配置要清晰 ---# 1. 集群地址列表，即使只有一个也要写成列表，方便扩展bootstrap_servers = ['192.168.1.11:9092', '192.168.1.12:9092', '192.168.1.13:9092']# 2. Topic 名称topic_name = 'hello-kraft'# 3. 序列化器：将 Python 字典转为 JSON 字符串再编码为字节流def json_serializer(data): return json.dumps(data).encode('utf-8')# --- 创建生产者实例 ---producer = KafkaProducer( bootstrap_servers=bootstrap_servers, value_serializer=json_serializer, # acks='all' 保证数据不丢失，是生产环境推荐的配置 acks='all', # 重试次数，应对网络抖动 retries=3)print("生产者已启动，开始发送消息...")# --- 发送消息 ---for i in range(10): message = { 'event_id': i, 'message': f'Hello Kafka from Python! Message #{i}', 'timestamp': time.time } try: # 发送！.get 会阻塞直到消息发送成功或失败 future = producer.send(topic_name, value=message) record_metadata = future.get(timeout=10) print(f"消息发送成功 -> Topic: {record_metadata.topic}, Partition: {record_metadata.partition}, Offset: {record_metadata.offset}") except Exception as e: print(f"消息发送失败: {e}") # 这里可以加入更复杂的错误处理逻辑，比如记录日志、告警 time.sleep(1)# --- 关闭生产者 ---producer.flushproducer.closeprint("生产者已关闭。")

️ 开发运维扯皮角 (Producer篇)

开发：“我程序报 KafkaTimeoutError！是不是 Kafka 集群挂了？”

运维：“我看了监控，集群好好的！是不是你代码有问题？”

排查思路：

网络先行：运维首先要确认，开发的应用服务器能否 telnet 192.168.1.11 9092 通。90% 的连接问题都是网络不通或防火墙。

检查 advertised.listeners：运维要再次确认 Kafka Broker 的 advertised.listeners 配置的是不是开发能访问的 IP。这是新手运维最常犯的错！

代码配置 acks：开发要检查 acks 配置。如果 acks='all'，意味着需要所有副本都确认收到才算成功，这对网络的稳定性和延迟要求更高。如果业务允许，可以降级为 acks=1（仅 Leader 确认）来提高发送速度，但会牺牲一点点可靠性。

消息大小：开发要确认发送的消息是否超过了 Kafka 的 message.max.bytes 限制（默认 1MB）。

消费者（Consumer）：从车上取包裹

消费者负责从 Topic 订阅并处理消息。

# consumer.pyfrom kafka import KafkaConsumerimport json# --- 配置 ---bootstrap_servers = ['192.168.1.11:9092', '192.168.1.12:9092', '192.168.1.13:9092']topic_name = 'hello-kraft'# 【关键】消费者组 ID，同一组的消费者会分摊 Partitiongroup_id = 'my-first-consumer-group'# --- 创建消费者实例 ---consumer = KafkaConsumer( topic_name, bootstrap_servers=bootstrap_servers, group_id=group_id, # auto_offset_reset='earliest' 表示从最早的消息开始消费 # 如果设置为 'latest'，则只消费启动后到达的新消息 auto_offset_reset='earliest', # 自动提交 offset，生产环境可能需要手动提交以保证精确一次处理 enable_auto_commit=True, auto_commit_interval_ms=5000, # 5秒提交一次 # 反序列化器 value_deserializer=lambda v: json.loads(v.decode('utf-8')))print(f"消费者已启动，正在监听 Topic '{topic_name}'...")# --- 循环消费消息 ---try: for message in consumer: print(f"收到消息 -> Partition: {message.partition}, Offset: {message.offset}") print(f" Key: {message.key}, Value: {message.value}") # 在这里编写你的业务处理逻辑 # ...except KeyboardInterrupt: print("停止消费...")finally: consumer.close print("消费者已关闭。")

️ 开发运维扯皮角 (Consumer篇)

开发：“我的服务好像重复消费了好多消息啊！Kafka 的 bug？”

运维：“集群没问题。你看看你的 group.id 是不是没设对？或者处理逻辑太慢了？”

排查思路：

group.id 是灵魂：开发必须保证，所有处理同一业务逻辑的消费者实例，都使用 相同的 group.id。如果每个实例都随机生成一个 group.id，那 Kafka 会认为它们都是独立的消费者，于是把所有消息都发给它们一遍，造成“伪重复消费”。

消费 rebalance：运维可以观察日志，看消费者组是否频繁发生 rebalance（重平衡）。当一个消费者加入或离开组时，会触发 rebalance，分区会重新分配。如果消费者的处理逻辑时间超过了 Kafka 的 session.timeout.ms（默认 45 秒），Broker 会认为它“假死”了，将它踢出组，从而引发 rebalance。这是导致重复消费的常见原因。

消息积压 (Lag)：运维的核心职责是监控 Consumer Lag。如果 Lag 持续增大，说明消费速度跟不上生产速度。此时需要：分析瓶颈：是消费者的处理逻辑太慢（比如调用外部 API、复杂计算）？还是资源不足（CPU、内存）？扩容：如果逻辑无法优化，最直接的办法就是 增加消费者实例。但前提是，你的 Topic 分区数必须大于等于消费者实例数，否则多出来的消费者实例会闲置，无法提高并行度。这就是为什么我们在上一篇强调“合理规划分区数”如此重要！

日本首相石破茂宣布辞去自民党总裁职务　　当地时间7日18时左右，日本首相、自民党总裁石破茂在首相官邸举行紧急记者会，宣布决定辞去自民党总裁职务。　　石破茂表示，有关临时总裁选举的相关手续事项，已向自民党干事长传达。在选出新总裁之前，将尽到自己的责任