Kafka Binder 侦听器容器定制器
Spring Cloud Stream 通过使用定制器为消息监听器容器提供了强大的自定义选项。
本节介绍可用于 Kafka 的定制器接口：ListenerContainerCustomizer，其特定于 Kafka 的扩展KafkaListenerContainerCustomizer，以及 specializedListenerContainerWithDlqAndRetryCustomizer.
侦听器容器定制器
这ListenerContainerCustomizer是 Spring Cloud Stream 中的一个通用接口，允许自定义消息侦听器容器。
用法
要使用ListenerContainerCustomizer，创建一个在配置中实现此接口的 bean：
@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> genericCustomizer() {
return (container, destinationName, group) -> {
// Customize the container here
};
}这ListenerContainerCustomizerinterface 定义了以下方法：
void configure(C container, String destinationName, String group);-
container：要自定义的消息侦听器容器。 -
destinationName：目标（主题）的名称。 -
group：消费者组 ID。
KafkaListenerContainer定制器
这KafkaListenerContainerCustomizer接口扩展ListenerContainerCustomizer修改侦听器容器的行为，并提供对特定于绑定的扩展 Kafka 使用者属性的访问。
用法
要使用KafkaListenerContainerCustomizer，创建一个在配置中实现此接口的 bean：
@Bean
public KafkaListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> kafkaCustomizer() {
return (container, destinationName, group, properties) -> {
// Customize the Kafka container here
};
}这KafkaListenerContainerCustomizerinterface 添加了以下方法：
default void configureKafkaListenerContainer(
C container,
String destinationName,
String group,
ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
configure(container, destinationName, group);
}这种方法扩展了底座configure方法，并带有附加参数：
-
extendedConsumerProperties：扩展的使用者属性，包括特定于 Kafka 的属性。
ListenerContainerWithDlqAndRetryCustomizer
这ListenerContainerWithDlqAndRetryCustomizer接口为涉及死信队列 （DLQ） 和重试机制的场景提供了额外的自定义选项。
用法
要使用ListenerContainerWithDlqAndRetryCustomizer，创建一个在配置中实现此接口的 bean：
@Bean
public ListenerContainerWithDlqAndRetryCustomizer dlqCustomizer() {
return (container, destinationName, group, dlqDestinationResolver, backOff, properties) -> {
// Access the container here with access to the extended consumer binding properties.
};
}这ListenerContainerWithDlqAndRetryCustomizerinterface 定义了以下方法：
void configure(
AbstractMessageListenerContainer<?, ?> container,
String destinationName,
String group,
BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> dlqDestinationResolver,
BackOff backOff,
ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties
);-
container：要自定义的 Kafka 侦听器容器。 -
destinationName：目标（主题）的名称。 -
group：消费者组 ID。 -
dlqDestinationResolver：用于解析失败记录的 DLQ 目标的函数。 -
backOff：重试的退避策略。 -
extendedConsumerProperties：扩展的使用者属性，包括特定于 Kafka 的属性。
总结
-
ListenerContainerWithDlqAndRetryCustomizer如果启用了 DLQ，则使用。 -
KafkaListenerContainerCustomizer用于没有 DLQ 的 Kafka 特定定制。 -
基地
ListenerContainerCustomizer用于通用定制。
这种分层方法允许在 Spring Cloud Stream 应用程序中灵活且具体地自定义 Kafka 侦听器容器。