Python Kafka SASL认证生产者消费者断线重连异常隔离

脚本专家

在实际的后端开发中，Kafka 作为高吞吐、高可靠的消息中间件，被广泛用于业务解耦和数据异步传输。线上 Kafka 集群基本都会开启 SASL 安全认证，防止未授权访问。本文基于 Python 实现带 SASL 账号密码认证的 Kafka 生产者和消费者完整逻辑，同时解决了断线掉线、单条消息异常阻塞、会话超时等生产环境常见问题。

整体设计方案采用配置与代码分离、异常容错、服务稳定常驻、消息不阻塞的原则。所有 Kafka 集群地址、认证方式、消费分组、超时参数等均统一写入 config.ini 配置文件，业务代码仅专注于消息发送和消费逻辑。这样后续集群变更或参数调优时无需修改业务代码。

生产者核心实现基于官方 kafka-python 库，适配 SASL_PLAINTEXT 认证模式。通过读取配置中的账号、密码、加密机制和安全协议等参数，严格匹配线上集群认证规则。消息发送采用同步模式，通过 future.get() 阻塞获取发送结果，精准捕获 topic、分区和偏移量信息，方便日志排查。同步模式相比异步更适合对可靠性要求高的场景，确保每条消息都成功投递。同时配置了重试机制（retries=3），自动应对临时网络波动。自定义 JSON 序列化时指定 ensure_ascii=False，保证中文正常传输。

消费者的核心优化包括常驻式自动重连机制、规避会话超时掉线、单条消息异常隔离和手动提交偏移量。原生的 Kafka 消费者在断连后无法自动恢复，这里通过外层死循环加异常捕获和休眠重试来实现：一旦检测到连接异常或未知运行时异常，服务不会宕机，休眠5秒后自动重建消费者，保障7×24小时运行。在消费者配置中删除了 consumer_timeout_ms 参数，避免因超时被集群剔除，依赖原生长轮询自动维持心跳。

单条消息异常隔离的实现是在消费循环中用 try-except 包裹每条消息的处理逻辑，当某条消息格式错误或业务处理异常时，仅跳过本条并打印日志，不会中断整体消费流程。使用手动提交 offset（consumer.commit() 在每条消息成功处理后调用），确保消息处理成功后才标记消费完成，避免业务处理失败但偏移量已提交导致消息丢失。

配置文件的解耦设计将所有核心参数收纳在 config.ini 中，分为 [KAFKA_BASE]、[KAFKA_PRODUCER]、[KAFKA_CONSUMER] 三个模块，涵盖集群地址、topic、认证信息、超时时间、消费分组、初始偏移量策略等。开发、测试、生产环境只需切换配置文件即可快速适配。

以下是完整的代码实现和配置文件示例。

依赖安装：

2. pip install kafka-python==2.3.0

复制代码

配置文件 config.ini：

2. [KAFKA_BASE]
3. bootstrap_servers = 192.168.0.10:9000
4. topic = JavaToPython
6. [KAFKA_PRODUCER]
7. security_protocol = SASL_PLAINTEXT
8. sasl_mechanism = PLAIN
9. sasl_plain_username = admin
10. sasl_plain_password = admin-secret
11. max_request_size = 10485760
12. send_timeout = 3000
14. [KAFKA_CONSUMER]
15. group_id = JavaToPythonGroup
16. security_protocol = SASL_PLAINTEXT
17. sasl_mechanism = PLAIN
18. sasl_plain_username = admin
19. sasl_plain_password = admin-secret
20. auto_offset_reset = latest
21. enable_auto_commit = False

复制代码

生产者代码 kafka_producer.py：

2. import configparser
3. import json
4. from kafka import KafkaProducer
5. from kafka.errors import KafkaError
6. import os
8. script_dir = os.path.dirname(os.path.abspath(__file__))
9. ini_path = os.path.join(script_dir, "config.ini")
10. conf = configparser.ConfigParser()
11. load_files = conf.read(ini_path, encoding="utf-8")
12. print("成功加载的配置文件：", load_files)
13. base_conf = conf["KAFKA_BASE"]
14. prod_conf = conf["KAFKA_PRODUCER"]
16. producer = KafkaProducer(
17.     bootstrap_servers=base_conf["bootstrap_servers"].split(","),
18.     security_protocol=prod_conf["security_protocol"],
19.     sasl_mechanism=prod_conf["sasl_mechanism"],
20.     sasl_plain_username=prod_conf["sasl_plain_username"],
21.     sasl_plain_password=prod_conf["sasl_plain_password"],
22.     max_request_size=int(prod_conf["max_request_size"]),
23.     retries=3,
24.     value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8")
25. )
27. def send_sync_msg(msg_data: dict):
28.     try:
29.         future = producer.send(topic=base_conf["topic"], value=msg_data)
30.         meta = future.get(timeout=int(prod_conf["send_timeout"]) / 1000)
31.         print(f"发送成功 | topic:{meta.topic} 分区:{meta.partition} offset:{meta.offset}")
32.         return True
33.     except KafkaError as e:
34.         print(f"发送失败：{str(e)}")
35.         return False
37. if __name__ == "__main__":
38.     test_msg = {"resourceId": 184501, "type": "文档", "resourceType": "txt"}
39.     for i in range(2):
40.         print(f"===== 第{i+1}次调用 =====")
41.         send_sync_msg(test_msg)
42.         producer.flush()

复制代码

消费者代码 kafka_consumer.py：

2. import configparser
3. import json
4. import time
5. from kafka import KafkaConsumer
6. from kafka.errors import KafkaError
7. import os
9. script_dir = os.path.dirname(os.path.abspath(__file__))
10. ini_path = os.path.join(script_dir, "config.ini")
11. conf = configparser.ConfigParser()
12. load_files = conf.read(ini_path, encoding="utf-8")
13. print("成功加载的配置文件：", load_files)
14. base_conf = conf["KAFKA_BASE"]
15. cons_conf = conf["KAFKA_CONSUMER"]
17. def create_consumer():
18.     consumer = KafkaConsumer(
19.         base_conf["topic"],
20.         bootstrap_servers=base_conf["bootstrap_servers"].split(","),
21.         group_id=cons_conf["group_id"],
22.         security_protocol=cons_conf["security_protocol"],
23.         sasl_mechanism=cons_conf["sasl_mechanism"],
24.         sasl_plain_username=cons_conf["sasl_plain_username"],
25.         sasl_plain_password=cons_conf["sasl_plain_password"],
26.         auto_offset_reset=cons_conf["auto_offset_reset"],
27.         enable_auto_commit=conf.getboolean("KAFKA_CONSUMER", "enable_auto_commit"),
28.         value_deserializer=lambda x: json.loads(x.decode("utf-8"))
29.     )
30.     return consumer
32. def consume_loop():
33.     print(f"消费任务启动，group:{cons_conf['group_id']} topic:{base_conf['topic']}")
34.     while True:
35.         try:
36.             consumer = create_consumer()
37.             print("Kafka消费者连接成功，持续监听新消息...")
38.             for msg in consumer:
39.                 try:
40.                     data = msg.value
41.                     print("=" * 60)
42.                     print(f"topic: {msg.topic}")
43.                     print(f"partition: {msg.partition}")
44.                     print(f"offset: {msg.offset}")
45.                     print(f"headers: {msg.headers}")
46.                     print(f"消息内容: {data}")
47.                     resource_id = data["resourceId"]
48.                     print("resourceId =", resource_id)
49.                     # 在此处编写业务处理逻辑
50.                     consumer.commit()
51.                 except Exception as single_err:
52.                     print(f"单条消息处理异常，跳过本条：{single_err}")
53.                     continue
54.         except KafkaError as e:
55.             print(f"Kafka连接异常，5秒后重连：{str(e)}")
56.         except Exception as e:
57.             print(f"消费循环未知异常，5秒后重启：{str(e)}")
58.         time.sleep(5)
60. if __name__ == "__main__":
61.     consume_loop()

复制代码

这套方案完全适配线上 SASL 认证集群，解决了原生 kafka-python 封装常见的中文乱码、服务掉线、消息阻塞、消息丢失和配置混乱等问题。代码轻量化、无冗余依赖，可直接用于数据同步、业务消息推送、日志采集等场景。同时代码结构清晰，可基于现有逻辑快速扩展消息重试、异常消息落盘、消费监控告警等功能。

热心网友4

非常感谢分享这么详细的实战方案！配置与代码分离的思路很好，尤其是把Kafka集群、认证参数都放在config.ini里，环境切换确实方便很多。 几个细节想请教一下： 1. 生产者用了同步future.get()，可靠性高，但在高吞吐场景下会不会因为等待ack而影响整体发送速率？有没有考虑过异步+回调记录错误的方式？ 2. 消费者里手动commit放在每条消息处理成功后，如果业务处理本身耗时较长（比如调用外部API），会不会导致consume循环阻塞，进而影响心跳和rebalance？有没有配合max.poll.records或session.timeout.ms做调优？ 3. 异常隔离只打印日志跳过，如果出现大量异常消息，是否考虑过投递到死信topic做后续补偿？ 代码很干净，断线重连的休眠5秒加外层循环也很实用。感谢！

热心网友4

感谢楼主的详细分享！这套生产者和消费者的设计思路非常实用，尤其是配置与代码解耦、手动提交offset结合异常隔离，以及外层死循环重连消费者，都是生产环境中容易踩坑的点。我自己之前也遇到过SASL认证配置不对导致连接失败，但没像你这样把集群地址和认证参数统一放到config里，后续切环境确实方便很多。 想请教一下：在消费者重连之后，消费组的分区分配会自动触发rebalance吗？你目前这种死循环重建消费者的方式，是否有考虑过避免频繁rebalance对集群的影响？还是说因为休眠了5秒，实际影响可控？

热心网友4

很详实的分享，特别是配置与代码分离的思路和生产环境常见的断线重连、异常隔离处理，都是实际运维中容易踩坑的地方。手动提交 offset 和单条 try-except 的设计能有效避免业务失败导致消息丢失，同时不影响后续消费，非常实用。不过想请教一下，在外层死循环重建消费者时，会不会因为旧消费者未正确关闭导致资源泄漏或 group rebalance 频繁？另外同步发送用 future.get(timeout) 阻塞，如果集群短暂抖动超时，重试机制是否会因为超时抛异常而整体失败？期待进一步交流。