|
对于最新的稳定版本，请使用 Spring for Apache Kafka 3.3.0！ |
连接到 Kafka
从版本 2.5 开始，这些都扩展了KafkaResourceFactory.
这允许在运行时通过添加Supplier<String>添加到他们的配置中：setBootstrapServersSupplier(() -> …).
将对所有新连接调用此函数以获取服务器列表。
Consumer 和 Producer 通常存在很长时间。
要关闭现有 Producer，请调用reset()在DefaultKafkaProducerFactory.
要关闭现有 Consumer，请调用stop()（然后start()） 在KafkaListenerEndpointRegistry和/或stop()和start()在任何其他侦听器容器 bean 上。
为方便起见，该框架还提供了一个ABSwitchCluster它支持两组引导服务器;其中 1 个随时处于活动状态。
配置ABSwitchCluster并将其添加到 producer 和 consumer 工厂中，并且KafkaAdmin，通过调用setBootstrapServersSupplier().
当您想要切换时，调用primary()或secondary()并调用reset()在生产商工厂建立新的连接;对于消费者，stop()和start()all listener containers 的 Listener 容器。
使用@KafkaListeners,stop()和start()这KafkaListenerEndpointRegistry豆。
有关更多信息，请参阅 Javadocs。
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是通过附加client-id属性（从metrics()创建后）到工厂beanName属性，以..
例如，这些侦听器可用于创建和绑定 MicrometerKafkaClientMetrics实例（并在客户端关闭时关闭它）。
该框架提供的侦听器正是这样做的;请参阅 Micrometer 本机度量。
默认客户端 ID 前缀
从版本 3.2 开始，对于使用spring.application.nameproperty，则现在使用此名称
作为这些客户端类型的自动生成的客户端 ID 的默认前缀：
-
不使用 Consumer Group 的 Consumer 客户端
-
生产者客户端
-
管理员客户端
这样可以更轻松地在服务器端识别这些客户端，以便进行故障排除或应用配额。
| 客户端类型 | 无应用程序名称 | 使用应用程序名称 |
|---|---|---|
没有 Consumer Group 的 Consumer |
使用者 null-1 |
myapp-consumer-1 |
具有使用者组 “mygroup” 的使用者 |
消费者 mygroup-1 |
消费者 mygroup-1 |
制作人 |
生产者-1 |
myapp-producer-1 |
管理 |
adminclient-1 |
myapp-admin-1 |