在 Python 中消费多个 Kafka topic，可以使用kafka-python库，这是一个流行的 Kafka 客户端库。以下是一个详细的代码示例，展示如何创建一个 Kafka 消费者，并同时消费多个 Kafka topic。
1.环境准备
（1）安装 Kafka 和 Zookeeper：确保 Kafka 和 Zookeeper 已经安装并运行。
（2）安装 kafka-python 库：通过 pip 安装kafka-python库。
bash复制代码
pip install kafka-python
复制代码
2.示例代码
以下是一个完整的 Python 脚本，展示了如何创建一个 Kafka 消费者并消费多个 topic。
from kafka import KafkaConsumerimport jsonimport 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) # 订阅多个topicconsumer.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()
复制代码
3.代码解释
（1）日志配置：使用 Python 的logging模块配置日志，方便调试和记录消费过程中的信息。
（2）Kafka 配置：设置 Kafka 服务器的地址、消费者组 ID 和要消费的 topic 列表。
（3）消费者配置：配置消费者参数，包括自动重置 offset、自动提交 offset 的时间间隔和消息反序列化方式（这里假设消息是 JSON 格式）。
（4）创建消费者：使用配置创建 Kafka 消费者实例。
（5）订阅 topic：通过consumer.subscribe方法订阅多个 topic。
（6）消费消息：在无限循环中消费消息，并打印消息的详细信息（topic、partition、offset、key 和 value）。
（7）异常处理：捕获KeyboardInterrupt（Ctrl+C）以优雅地关闭消费者，并捕获其他异常并记录日志。
4.运行脚本
确保 Kafka 和 Zookeeper 正在运行，并且你已经在 Kafka 中创建了相应的 topic（topic1、topic2、topic3）。然后运行脚本：
bash复制代码
python kafka_multi_topic_consumer.py
复制代码
这个脚本将开始消费指定的 topic，并在控制台上打印出每条消息的详细信息。你可以根据需要修改脚本中的处理逻辑，比如将消息存储到数据库或发送到其他服务。
5.参考价值和实际意义
这个示例代码展示了如何在 Python 中使用kafka-python库消费多个 Kafka topic，适用于需要处理来自不同 topic 的数据流的场景。例如，在实时数据处理系统中，不同的 topic 可能代表不同类型的数据流，通过消费多个 topic，可以实现数据的整合和处理。此外，该示例还展示了基本的异常处理和日志记录，有助于在生产环境中进行调试和监控。
文章转载自：TechSynapse
原文链接：https://www.cnblogs.com/TS86/p/18559602
体验地址：http://www.jnpfsoft.com/?from=infoq
评论