python消费kafka

在Python中消费Kafka数据的方法有多种。以下是其中一种方法的示例代码：

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    't2',  # 要消费的Kafka主题
    bootstrap_servers=['IP:9092'],  # Kafka的服务器地址
    auto_offset_reset='latest',  # 消费者的起始偏移量
    group_id='group1'  # 消费者所属的消费组
)

for msg in consumer:
    try:
        if msg.value is not None:
            data_json = msg.value.decode()
            print(data_json)
    except Exception as e:
        print(e)

print("finish")

该代码创建了一个KafkaConsumer对象，指定要消费的Kafka主题、Kafka服务器地址、起始偏移量和消费组。然后通过迭代消费者对象来获取消息，并进行相应的处理。在上述示例中，我们将收到的消息解码为字符串并打印出来。当没有更多的消息可供消费时，循环会结束并打印"finish"。

这是一种基本的消费Kafka数据的方法。你可以根据自己的需求进行扩展和修改。希望对你有所帮助！