通过Spring Kafka列出Kafka主题

Question

我们想通过spring-kafka列出所有Kafka主题，以获得类似于kafka命令的结果：

bin/kafka-topics.sh --list --zookeeper localhost:2181

在下面的服务中运行 getTopics（） 方法时，我们会得到

配置：

@EnableKafka
@Configuration
public class KafkaConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:2181");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
            StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
}

服务：

@Service
public class TopicServiceKafkaImpl implements TopicService {
    @Autowired
    private ConsumerFactory<String, String> consumerFactory;

    @Override
    public Set<String> getTopics() {
        try (Consumer<String, String> consumer = 
            consumerFactory.createConsumer()) {
            Map<String, List<PartitionInfo>> map = consumer.listTopics();
            return map.keySet();
    }
}

Kafka已经启动并运行，我们可以成功地从应用程序向主题发送消息。

Answer 1

kafka-tops--list是一个外壳脚本，它只是kafka.admin.TopicCommand类的包装器，您可以在其中找到您正在寻找的方法

或者，您也可以使用AdminClient#listTopics方法

Answer 2

您正在连接到动物园管理员 （2181） 而不是 Kafka（默认为 9092）。

爪哇Kafka客户端不再直接与 ZK 通信。

Answer 3

您可以使用管理客户端列出此类主题

    Properties properties = new Properties();
    properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

    AdminClient adminClient = AdminClient.create(properties);

    ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
    listTopicsOptions.listInternal(true);

    System.out.println("topics:" + adminClient.listTopics(listTopicsOptions).names().get());