java - 为什么我需要创建 Kafka Consumer 来连接到 Schema Registry？

Question

上一篇注释:我对 Kafka 还很陌生。

我正在尝试从架构注册表获取所有架构，但我无法仅使用架构注册表客户端来执行此操作。仅当我在此之前实例化 KafkaConsumer 时它才有效。

不明白为什么。这是代码(消费者就位)。

ConsumerConfig 只是一个包含所需所有配置的类。包括架构注册表 URL。

Consumer<String, String>  consumer = new KafkaConsumer<String, String>(ConsumerConfig.get());
CachedSchemaRegistryClient client = new CachedSchemaRegistryClient(ConsumerConfig.getSchemaRegistryURL(), 30);
Collection<String> listOfSubjects = client.getAllSubjects();
consumer.close();

没有消费者，我得到:

io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: No content to map due to end-of-input

对于消费者来说，一切都运转良好。我希望有人能解释一下为什么会发生这种情况，我是否认为我没有理由需要通过消费者连接到实际的 Kafka 集群才能访问另一个端点上的架构注册表。

Answer 1

您根本不必创建 KafkaConsumer 实例。两者都是完全独立的。

如果您只想从 SchemaRegistry 获取所有主题和架构，只需创建 CachedSchemaRegistryClient 实例并调用相关操作即可。

这是一个工作示例:

 private final static Map<String, Schema> schemas = new ConcurrentHashMap<>();
 protected static SchemaRegistryClient schemaRegistryClient;

 public static void main(String[] args) {
       String registryUrl = "http://localhost:8081";
        try {
            schemaRegistryClient = new CachedSchemaRegistryClient(registryUrl, 30);
            System.out.println(schemaRegistryClient);
            Collection<String> subjects = schemaRegistryClient.getAllSubjects();
            System.out.println(subjects);
        } catch (Exception e){
            throw new RuntimeException(e);
        }
    }