<think>我们正在处理用户关于Python中Kafka消费者订阅多个topic的问题。 根据引用[1]和引用[2]，我们得知可以使用kafka-python库来实现，并且引用[2]提供了一个具体的代码示例。 用户明确要求订阅多个topic，因此我们需要展示如何创建消费者并订阅多个topic。 步骤： 1. 导入必要的模块（KafkaConsumer） 2. 创建KafkaConsumer实例，设置bootstrap_servers参数（Kafka集群地址） 3. 使用subscribe()方法订阅多个topic（传入一个topic列表） 4. 循环消费消息，处理每条消息 注意：根据引用[2]的代码，我们可以直接使用consumer.subscribe(['topic1','topic2'])来订阅多个topic。 另外，引用[4]也展示了消费者的基本使用，但只订阅了一个topic。不过，其创建消费者的方式类似。 因此，我们将编写一个示例代码，展示如何订阅多个topic。 同时，我们还需要注意其他配置参数，比如： - group_id：消费组ID - auto_offset_reset：当没有初始偏移量或偏移量失效时，从何处开始消费（'earliest'或'latest'） - enable_auto_commit：是否自动提交偏移量 - auto_commit_interval_ms：自动提交偏移量的时间间隔 我们将参考引用[2]和引用[4]中的配置，给出一个完整的示例。 由于用户要求使用中文回答，我们将用中文进行说明。 </think> 在Python中，使用`kafka-python`库实现消费者订阅多个Kafka topic的方法如下： ### 实现步骤 1. **创建消费者实例** 通过`KafkaConsumer`类初始化消费者，配置Kafka集群地址： ```python from kafka import KafkaConsumer consumer = KafkaConsumer( bootstrap_servers='kafka-server:9092', # Kafka集群地址 group_id='my_consumer_group', # 消费组ID（可选） auto_offset_reset='earliest', # 从最早消息开始消费 enable_auto_commit=True, # 自动提交偏移量 value_deserializer=lambda x: x.decode('utf-8') # 消息反序列化 ) ``` 2. **订阅多个Topic** 使用`subscribe()`方法传入**topic名称列表**： ```python consumer.subscribe(['topic1', 'topic2', 'topic3']) # 订阅多个topic ``` 3. **消费消息** 循环读取消息并处理，通过`message.topic`区分消息来源： ```python for message in consumer: print(f"收到消息 [Topic:{message.topic}]") print(f"内容: {message.value}") # 根据topic执行不同业务逻辑 if message.topic == 'topic1': process_topic1(message.value) elif message.topic == 'topic2': process_topic2(message.value) ``` ### 关键配置说明 | 参数 | 说明 | |--------------------|----------------------------------------------------------------------| | `bootstrap_servers` | Kafka集群地址（逗号分隔多个节点） | | `group_id` | 消费组ID，实现消费者负载均衡（相同组内共享topic）[^3] | | `auto_offset_reset` | `'earliest'`(从最早消息)/`'latest'`(从最新消息) | | `enable_auto_commit`| 自动提交偏移量，避免重复消费 | ### 完整示例 ```python from kafka import KafkaConsumer consumer = KafkaConsumer( bootstrap_servers=['kafka1:9092', 'kafka2:9092'], group_id='multi-topic-group', auto_offset_reset='earliest', value_deserializer=lambda x: x.decode('utf-8') ) consumer.subscribe(['user_events', 'system_logs', 'payment_records']) try: for msg in consumer: print(f"[{msg.topic}] 分区:{msg.partition} 偏移量:{msg.offset}") print(f"消息内容: {msg.value[:100]}...") # 截取前100字符 except KeyboardInterrupt: print("停止消费") finally: consumer.close() ``` ### 注意事项 1. **分区分配** 同一消费组内的消费者自动分配topic分区，无需手动管理[^3] 2. **偏移量提交** 建议启用`enable_auto_commit=True`避免消息重复处理 3. **异常处理** 添加`try/except`块捕获`KafkaTimeoutError`等网络异常 4. **性能优化** 批量消费消息时可配置`max_poll_records`参数