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都提供了灵活的方式来满足不同的需求。