kafka如何查詢topic數據

要查詢Kafka中的topic數據，您可以使用Kafka的命令行工具或編程API。以下是兩種常用方法的簡要說明：

1. 使用Kafka命令行工具 kafka-console-consumer.sh：

   首先，確保您已經安裝并啟動了Kafka。然后，運行以下命令來消費指定topic的數據：

   ./kafka-console-consumer.sh --bootstrap-server <kafka_broker_address> --topic <topic_name> --from-beginning
   

   其中，將 <kafka_broker_address> 替換為您的Kafka代理地址（例如：localhost:9092），將 <topic_name> 替換為您要查詢的topic名稱。--from-beginning 參數表示從topic的最早記錄開始消費。

   這將啟動一個交互式消費者，您可以查看并消費topic中的數據。要停止消費者，請按Ctrl+C。

2. 使用Kafka客戶端庫編程API：

   您可以使用Kafka客戶端庫（如Java、Python、Go等）編寫程序來查詢Kafka中的topic數據。以下是使用Python和confluent_kafka庫的示例：

   首先，安裝confluent_kafka庫：

   pip install confluent-kafka
   

   然后，編寫以下Python代碼來消費指定topic的數據：

   from confluent_kafka import Consumer, KafkaError
   
   def consume_messages(kafka_broker, topic):
       conf = {
           'bootstrap.servers': kafka_broker,
           'group.id': 'my_consumer_group',
           'auto.offset.reset': 'earliest'
       }
   
       consumer = Consumer(conf)
       consumer.subscribe([topic])
   
       try:
           while True:
               msg = consumer.poll(timeout=1.0)
   
               if msg is None:
                   continue
   
               if msg.error():
                   if msg.error().code() == KafkaError._PARTITION_EOF:
                       print(f"Reached end of partition {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")
                   else:
                       raise KafkaException(msg.error())
   
               print(f"Received message: {msg.value().decode('utf-8')}")
   
       except KeyboardInterrupt:
           pass
   
       finally:
           consumer.close()
   
   if __name__ == '__main__':
       kafka_broker = '<kafka_broker_address>'
       topic = '<topic_name>'
       consume_messages(kafka_broker, topic)
   

   將 <kafka_broker_address> 替換為您的Kafka代理地址，將 <topic_name> 替換為您要查詢的topic名稱。運行此程序后，您將看到從topic的最早記錄開始接收的消息。要停止程序，請按Ctrl+C。