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服務(wù)器地址
group_id = 'multi-topic-consumer-group'
topics = ['topic1', 'topic2', 'topic3']  # 替換為你要消費(fèi)的topic
 
# 消費(fèi)者配置
consumer_config = {
    'bootstrap_servers': bootstrap_servers,
    'group_id': group_id,
    'auto_offset_reset': 'earliest',  # 從最早的offset開始消費(fèi)
    'enable_auto_commit': True,
    'auto_commit_interval_ms': 5000,
    'value_deserializer': lambda x: json.loads(x.decode('utf-8'))  # 假設(shè)消息是JSON格式
}
 
# 創(chuàng)建Kafka消費(fèi)者
consumer = KafkaConsumer(**consumer_config)
 
# 訂閱多個(gè)topic
consumer.subscribe(topics)
 
try:
    # 無(wú)限循環(huán)，持續(xù)消費(fèi)消息
    while True:
        for message in consumer:
            topic = message.topic
            partition = message.partition
            offset = message.offset
            key = message.key
            value = message.value
 
            # 打印消費(fèi)到的消息
            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，優(yōu)雅關(guān)閉消費(fèi)者
    logger.info("Caught KeyboardInterrupt, closing consumer.")
    consumer.close()
 
except Exception as e:
    # 捕獲其他異常，記錄日志并關(guān)閉消費(fèi)者
    logger.error(f"An error occurred: {e}", exc_info=True)
    consumer.close()

python kafka_multi_topic_consumer.py