如何消费Kafka Topic？现有Python消费代码无输出问题求助

核心问题分析 #

你的代码存在几个关键配置错误，导致无法正常连接目标Kafka集群，因此没有消息输出：

• 错误使用本地Kafka地址localhost:9092，而非同事提供的集群地址server1,server2,server3
• 未配置集群要求的用户名密码认证信息
• 缺少必要的安全协议配置，且未设置消费者组ID，可能引发消费逻辑异常

修正后的代码 #

import json
from kafka import KafkaConsumer
from kafka.errors import KafkaError

def consume_data():
    try:
        consumer = KafkaConsumer(
            'topic1',
            # 替换为实际集群地址
            bootstrap_servers=['server1', 'server2', 'server3'],
            # 必须设置消费者组ID，Kafka依赖其管理偏移量
            group_id='my_consumer_group',
            # 认证配置，匹配集群安全策略
            security_protocol='SASL_PLAINTEXT',  # 若集群用SSL则改为SASL_SSL
            sasl_mechanism='PLAIN',
            sasl_plain_username='xyz',
            sasl_plain_password='xyz',
            key_deserializer=lambda k: k.decode('utf-8') if k else None,
            value_deserializer=lambda v: json.loads(v.decode('utf-8')) if v else None,
            # 首次消费时读取历史消息，避免因默认latest策略错过内容
            auto_offset_reset='earliest'
        )
        
        print("成功连接Kafka集群，开始消费消息...")
        for message in consumer:
            print(f"主题: {message.topic}, 分区: {message.partition}, 偏移量: {message.offset}")
            print(f"Key: {message.key}, Value: {message.value}")
            
    except KafkaError as e:
        print(f"Kafka连接/消费失败: {str(e)}")
    finally:
        consumer.close()

if __name__ == '__main__':
    consume_data()

关键配置说明 #

• bootstrap_servers: 严格使用同事提供的集群节点地址，若节点有自定义端口需补充（如server1:9093）
• 认证信息: 必须加入security_protocol、sasl_mechanism及账号密码，确保与集群安全配置一致
• group_id: 强制设置，无组ID会导致Kafka无法跟踪消费偏移量，可能出现重复消费或无消费的情况
• auto_offset_reset: 设置为earliest可读取主题所有历史消息，若仅需新消息可改为latest
• 异常捕获: 加入错误处理，方便快速定位连接失败、认证错误等问题

额外排查点 #

1. 测试网络连通性：用telnet server1 9092或nc -zv server1 9092确认能访问集群节点
2. 验证主题状态：用Kafka命令行工具kafka-console-consumer.sh测试topic1是否有消息
3. 适配SSL场景：若集群用SASL_SSL，需添加ssl_cafile等证书配置（根据集群要求）

内容的提问来源于stack exchange，提问作者random23