kafka如何查詢topic列表和topic下的消息

Kafka如何查詢Topic列表和Topic下的消息

Apache Kafka是一個分布式流處理平臺，廣泛用于構建實時數據管道和流應用。在使用Kafka時，查詢Topic列表和Topic下的消息是常見的操作。本文將詳細介紹如何使用Kafka命令行工具和API來查詢Topic列表以及Topic下的消息。

1. 查詢Topic列表

1.1 使用Kafka命令行工具

Kafka提供了一個命令行工具kafka-topics.sh，可以用來查詢Kafka集群中的Topic列表。

1.1.1 查詢所有Topic

要查詢Kafka集群中的所有Topic，可以使用以下命令：

kafka-topics.sh --list --bootstrap-server <broker_address>

其中，<broker_address>是Kafka broker的地址，例如localhost:9092。

1.1.2 查詢特定Topic

如果你只想查詢某個特定的Topic，可以使用--topic參數：

kafka-topics.sh --describe --topic <topic_name> --bootstrap-server <broker_address>

這將顯示指定Topic的詳細信息，包括分區數、副本數等。

1.2 使用Kafka AdminClient API

除了命令行工具，你還可以使用Kafka的Java API來查詢Topic列表。Kafka提供了AdminClient類，可以用來管理Kafka集群。

1.2.1 創建AdminClient

首先，你需要創建一個AdminClient實例：

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
AdminClient adminClient = AdminClient.create(props);

1.2.2 查詢Topic列表

使用AdminClient的listTopics方法可以查詢Topic列表：

ListTopicsResult topics = adminClient.listTopics();
Set<String> topicNames = topics.names().get();
for (String topicName : topicNames) {
    System.out.println(topicName);
}

1.3 使用Kafka REST Proxy

如果你不想直接使用命令行工具或Java API，Kafka還提供了一個REST Proxy，可以通過HTTP請求來查詢Topic列表。

1.3.1 查詢所有Topic

發送一個GET請求到/topics端點：

curl -X GET http://localhost:8082/topics

這將返回一個JSON格式的Topic列表。

2. 查詢Topic下的消息

2.1 使用Kafka命令行工具

Kafka提供了一個命令行工具kafka-console-consumer.sh，可以用來消費指定Topic的消息。

2.1.1 消費指定Topic的消息

要消費指定Topic的消息，可以使用以下命令：

kafka-console-consumer.sh --topic <topic_name> --bootstrap-server <broker_address>

這將從指定Topic的最新消息開始消費，并實時輸出到控制臺。

2.1.2 從指定偏移量開始消費

如果你想從某個特定的偏移量開始消費消息，可以使用--offset參數：

kafka-console-consumer.sh --topic <topic_name> --bootstrap-server <broker_address> --offset <offset>

其中，<offset>是你想要開始消費的偏移量。

2.2 使用Kafka Consumer API

除了命令行工具，你還可以使用Kafka的Java API來消費指定Topic的消息。Kafka提供了KafkaConsumer類，可以用來消費消息。

2.2.1 創建KafkaConsumer

首先，你需要創建一個KafkaConsumer實例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

2.2.2 訂閱Topic

使用KafkaConsumer的subscribe方法可以訂閱指定的Topic：

consumer.subscribe(Arrays.asList("my-topic"));

2.2.3 消費消息

使用KafkaConsumer的poll方法可以消費消息：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

2.3 使用Kafka REST Proxy

Kafka REST Proxy也支持通過HTTP請求來消費指定Topic的消息。

2.3.1 消費指定Topic的消息

發送一個GET請求到/consumers/<group_id>/instances/<instance_id>/topics/<topic_name>端點：

curl -X GET http://localhost:8082/consumers/my-group/instances/my-instance/topics/my-topic

這將返回指定Topic的消息。

3. 總結

本文介紹了如何使用Kafka命令行工具、Java API和REST Proxy來查詢Topic列表和Topic下的消息。通過這些方法，你可以輕松地管理和監控Kafka集群中的Topic和消息。無論是通過命令行工具還是編程接口，Kafka都提供了靈活的方式來滿足不同的需求。