Python Kafka Consumer实现无消息接收问题求助

以下是你可能遗漏的关键配置和排查点：

• 必须指定消费者组ID
  未配置group_id时，Kafka会分配临时组ID，重启后偏移会丢失，且默认只会消费启动后的新消息。添加group_id配置：

  self.consumer = KafkaConsumer(
      # 原有配置...
      group_id="your_consumer_group_name",
  )
  

• 调整起始偏移策略
  默认auto_offset_reset为latest，仅消费消费者启动后产生的消息。如果要读取主题中已存在的历史消息，需设置为earliest：

  self.consumer = KafkaConsumer(
      # 原有配置...
      auto_offset_reset="earliest",
  )
  

• 补全json模块导入
  代码中使用了json.loads但未导入json模块，会导致消费环节抛出异常卡住。在顶部添加：

  import json
  

• 触发元数据同步
  订阅主题后，手动调用poll(0)强制同步元数据，确保消费者获取主题分区信息：

  def receive_messages(self, topic):
      self.consumer.subscribe(topics=[topic])
      self.consumer.poll(0)  # 同步元数据
      print(f"Subscribed to topics: {self.consumer.subscription()}")
      # 后续消费逻辑...
  

内容的提问来源于stack exchange，提问作者J. Donič