整体方案设计思路本次封装的 Kafka 工具分为生产者和消费者两大模块全程采用配置文件解耦核心参数避免硬编码服务器地址、账号密码、超时时间等关键配置提升代码可维护性和通用性。核心设计原则配置与代码分离、异常容错、服务稳定常驻、消息不阻塞。所有 Kafka 集群地址、认证方式、消费分组、超时参数等均统一写入 ini 配置文件业务代码仅专注于消息发送和消息消费逻辑便于后续集群变更、参数调优无需修改业务代码。二、Kafka 生产者核心实现与优势SASL 安全认证初始化生产者基于官方 kafka-python 库初始化完整适配 SASL_PLAINTEXT 认证模式配置文件读取账号、密码、加密机制、安全协议等参数严格匹配线上 Kafka 集群认证规则保障消息传输的安全性杜绝匿名访问风险。同步可靠消息发送机制采用同步发送模式生产消息通过 future.get() 阻塞获取消息发送结果精准捕获消息的 Topic、分区、偏移量信息方便日志排查和问题追溯。相较于异步发送同步模式更适合对消息可靠性要求较高的业务场景确保每条消息都成功投递。同时配置了消息重试机制针对临时网络波动、集群短暂不可用等问题自动重试发送大幅提升消息投递成功率。并且自定义 JSON 序列化规则支持中文正常传输避免默认编码导致的中文乱码问题。异常容错处理对 Kafka 专属异常进行单独捕获发送失败时打印详细报错信息并返回失败标识方便上层业务做重试、告警等拓展逻辑。测试场景下通过循环调用验证接口稳定性搭配手动刷新缓冲区确保消息即时推送至集群。三、Kafka 消费者核心优化与亮点常驻式自动重连机制原生 Kafka 消费者存在断连后无法自动恢复的问题网络波动、集群重启都会导致消费服务终止。本次封装通过外层死循环异常捕获休眠重试的方案实现消费者断线自动重建一旦检测到 Kafka 连接异常、未知运行异常服务不会宕机休眠5秒后自动重启消费任务保障服务7×24小时常驻运行。规避会话超时掉线问题优化原生消费者配置删除多余的会话超时参数依托 Kafka 原生长轮询机制自动维持心跳会话持续与 Broker 保持连接有效避免常规配置不当导致的消费者被集群剔除、频繁掉线等问题极大提升消费稳定性。单条消息异常隔离消费循环中增加单条消息独立异常捕获逻辑当某一条消息格式错误、字段缺失、业务处理异常时不会中断整体消费流程仅跳过当前异常消息并打印日志保证后续正常消息可以持续消费避免单点异常导致整个消费服务瘫痪。手动提交偏移量精准可控采用手动提交 offset 的方式在单条消息完整处理完成后再提交偏移量。相较于自动提交手动提交可以精准保证消息处理成功才标记消费完成避免业务处理失败但偏移量已提交导致消息丢失的问题完美适配业务一致性要求较高的场景。四、配置文件解耦设计所有核心参数统一收纳在 config.ini 配置文件中分为基础集群配置、生产者配置、消费者配置三大模块。涵盖集群地址、主题名称、认证信息、超时时间、消费分组、初始偏移量策略等全部参数。这种设计极大提升了代码的通用性开发、测试、生产环境可通过切换配置文件快速适配后续集群扩容、认证信息更新、参数调优无需改动业务代码降低维护成本。五、实战落地总结这套 Kafka 生产消费方案完全适配线上 SASL 认证集群解决了 Python 原生 Kafka 封装常见的中文乱码、服务掉线、消息阻塞、消息丢失、配置混乱等痛点。整体代码轻量化、无冗余依赖容错机制完善、稳定性高可直接复用至日常数据同步、业务消息推送、日志采集等各类业务场景。同时代码结构清晰、拓展性强可基于现有逻辑快速新增消息重试、异常消息落盘、消费监控告警、批量消费等功能适配复杂的生产环境需求。依赖包1pip install kafka-python2.3.0配置文件 config.ini12345678910111213141516171819202122[KAFKA_BASE]bootstrap_servers 192.168.0.10:9000topic JavaToPython[KAFKA_PRODUCER]security_protocol SASL_PLAINTEXTsasl_mechanism PLAINsasl_plain_username adminsasl_plain_password admin-secretmax_request_size 10485760send_timeout 3000[KAFKA_CONSUMER]group_id JavaToPythonGroupsecurity_protocol SASL_PLAINTEXTsasl_mechanism PLAINsasl_plain_username adminsasl_plain_password admin-secretauto_offset_reset latestenable_auto_commit False#auto_commit_interval_ms 1000#consumer_timeout_ms 5000生产者 kafka_producer.py12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364import configparserimport jsonfrom kafka import KafkaProducerfrom kafka.errors import KafkaErrorimport os获取当前脚本文件所在文件夹script_dir os.path.dirname(os.path.abspath(file))拼接config.ini完整路径ini_path os.path.join(script_dir, “config.ini”)加载ini配置conf configparser.ConfigParser()read返回成功读取的文件列表可打印校验load_files conf.read(ini_path, encoding“utf-8”)print(“成功加载的配置文件”, load_files)base_conf conf[“KAFKA_BASE”]prod_conf conf[“KAFKA_PRODUCER”]初始化带SASL认证生产者producer KafkaProducer(bootstrap_serversbase_conf[“bootstrap_servers”].split(“,”),security_protocolprod_conf[“security_protocol”],sasl_mechanismprod_conf[“sasl_mechanism”],sasl_plain_usernameprod_conf[“sasl_plain_username”],sasl_plain_passwordprod_conf[“sasl_plain_password”],max_request_sizeint(prod_conf[“max_request_size”]),retries3,# json序列化value_serializerlambda v: json.dumps(v, ensure_asciiFalse).encode(“utf-8”))def send_sync_msg(msg_data: dict):“”“同步发送json业务消息”“”try:future producer.send(topicbase_conf[“topic”], valuemsg_data)meta future.get(timeoutint(prod_conf[“send_timeout”]) / 1000)print(f发送成功 | topic:{meta.topic} 分区:{meta.partition} offset:{meta.offset}“)return Trueexcept KafkaError as e:print(f发送失败{str(e)}”)return Falseifname “main”:test_msg { resourceId: 184501, type: 文档, resourceType: txt } for i in range(2): print(f 第{i1}次调用 ) send_sync_msg(test_msg) producer.flush()