跳到主要内容
版本：4.1.1

连接到 Kafka

Doubao 中英对照 Connecting to Kafka

从 2.5 版本开始，上述每个工厂都继承 KafkaResourceFactory。这使得可以在运行时通过向其配置中添加一个 Supplier<String> 来更改引导服务器：setBootstrapServersSupplier(() -> …​)。该方法会为所有新建连接调用，以获取服务器列表。消费者和生产者通常是长生命周期的。要关闭已存在的生产者，请在 DefaultKafkaProducerFactory 上调用 reset()。要关闭已存在的消费者，请在 KafkaListenerEndpointRegistry 上调用 stop()（然后调用 start()），以及/或者在其他任意监听容器 bean 上调用 stop() 和 start()。

为方便起见，该框架还提供了一个 ABSwitchCluster，它支持两组引导服务器；任意时刻其中一组处于活动状态。配置 ABSwitchCluster 并通过调用 setBootstrapServersSupplier() 将其添加到生产者和消费者工厂以及 KafkaAdmin 中。当你想要切换时，调用 primary() 或 secondary()，并在生产者工厂上调用 reset() 以建立新连接；对于消费者，对所有监听容器调用 stop() 和 start()。使用 @KafkaListener 时，对 KafkaListenerEndpointRegistry bean 调用 stop() 和 start()。

更多信息请参阅 Javadoc。

Factory 监听器​

从 2.5 版本开始，DefaultKafkaProducerFactory 和 DefaultKafkaConsumerFactory 可以配置一个 Listener，以便在生产者或消费者被创建或关闭时接收通知。

interface Listener<K, V> {

    default void producerAdded(String id, Producer<K, V> producer) {
    }

    default void producerRemoved(String id, Producer<K, V> producer) {
    }

}
interface Listener<K, V> {

    default void consumerAdded(String id, Consumer<K, V> consumer) {
    }

    default void consumerRemoved(String id, Consumer<K, V> consumer) {
    }

}

在每种情况下，id 都是这样创建的：将创建后从 metrics() 获取的 client-id 属性追加到工厂的 beanName 属性之后，两者以 . 分隔。

这些监听器可用于——例如——在创建新客户端时创建并绑定一个 Micrometer KafkaClientMetrics 实例（并在客户端关闭时将其关闭）。

该框架提供了恰好实现此功能的监听器；请参阅 Micrometer 原生指标。

默认客户端 ID 前缀​

从 3.2 版本开始，对于使用 spring.application.name 属性定义了应用程序名称的 Spring Boot 应用程序，该名称现在被用作以下客户端类型自动生成客户端 ID 的默认前缀：

  • 不使用消费者组的消费者客户端
  • 生产者客户端
  • 管理客户端

这使得在服务端识别这些客户端以进行故障排查或应用配额变得更加容易。

表 1. 对 spring.application.name=myapp 的 Spring Boot 应用生成的客户端 ID 示例

客户端类型无应用程序名称有应用程序名称
无消费者组的消费者consumer-null-1myapp-consumer-1
使用消费者组 "mygroup" 的消费者consumer-mygroup-1consumer-mygroup-1
生产者producer-1myapp-producer-1
管理员adminclient-1myapp-admin-1