详解Python脚本如何消费多个Kafka topic

pip install kafka-python

from kafka import KafkaConsumer
import json
import logging
 
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
 
# Kafka配置
bootstrap_servers = 'localhost:9092'  # 替换为你的Kafka服务器地址
group_id = 'multi-topic-consumer-group'
topics = ['topic1', 'topic2', 'topic3']  # 替换为你要消费的topic
 
# 消费者配置
consumer_config = {
    'bootstrap_servers': bootstrap_servers,
    'group_id': group_id,
    'auto_offset_reset': 'earliest',  # 从最早的offset开始消费
    'enable_auto_commit': True,
    'auto_commit_interval_ms': 5000,
    'value_deserializer': lambda x: json.loads(x.decode('utf-8'))  # 假设消息是JSON格式
}
 
# 创建Kafka消费者
consumer = KafkaConsumer(**consumer_config)
 
# 订阅多个topic
consumer.subscribe(topics)
 
try:
    # 无限循环，持续消费消息
    while True:
        for message in consumer:
            topic = message.topic
            partition = message.partition
            offset = message.offset
            key = message.key
            value = message.value
 
            # 打印消费到的消息
            logger.info(f"Consumed message from topic: {topic}, partition: {partition}, offset: {offset}, key: {key}, value: {value}")
 
            # 你可以在这里添加处理消息的逻辑
            # process_message(topic, partition, offset, key, value)
 
except KeyboardInterrupt:
    # 捕获Ctrl+C，优雅关闭消费者
    logger.info("Caught KeyboardInterrupt, closing consumer.")
    consumer.close()
 
except Exception as e:
    # 捕获其他异常，记录日志并关闭消费者
    logger.error(f"An error occurred: {e}", exc_info=True)
    consumer.close()

python kafka_multi_topic_consumer.py