|
此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 Spring for Apache Kafka 3.3.0！ |
配置
从版本 2.9 开始，对于默认配置，@EnableKafkaRetryTopic注解应该在@Configurationannotated 类。
这使该功能能够正确引导，并允许注入一些功能的组件，以便在运行时进行查找。
没有必要同时添加@EnableKafka，如果添加此批注，则因为@EnableKafkaRetryTopic使用@EnableKafka. |
此外，从该版本开始，要对功能的组件和全局功能进行更高级的配置，RetryTopicConfigurationSupport类应该在@Configurationclass 的 API 和适当的方法。
有关更多详细信息，请参阅配置全局设置和功能。
默认情况下，重试主题的容器将具有与主容器相同的并发性。
从版本 3.0 开始，您可以设置不同的concurrency对于重试容器（在注解上，或在RetryConfigurationBuilder).
只能使用上述技术中的一种，并且只能使用一种@Configuration类可以扩展RetryTopicConfigurationSupport. |
使用@RetryableTopic注解
要为@KafkaListenerannotated 方法，您只需添加@RetryableTopic注解，Spring for Apache Kafka 将使用默认配置引导所有必要的主题和使用者。
@RetryableTopic(kafkaTemplate = "myRetryableTopicKafkaTemplate")
@KafkaListener(topics = "my-annotated-topic", groupId = "myGroupId")
public void processMessage(MyPojo message) {
// ... message processing
}从 3.2 开始，@RetryableTopic对类@KafkaListener的支持为：
@RetryableTopic(listenerContainerFactory = "my-retry-topic-factory")
@KafkaListener(topics = "my-annotated-topic")
public class ClassLevelRetryListener {
@KafkaHandler
public void processMessage(MyPojo message) {
// ... message processing
}
}您可以在同一类中指定一个方法来处理 DLT 消息，方法是使用@DltHandler注解。
如果未提供 DltHandler 方法，则会创建一个默认使用者，该使用者仅记录消耗。
@DltHandler
public void processMessage(MyPojo message) {
// ... message processing, persistence, etc
}如果未指定 kafkaTemplate 名称，则名称为defaultRetryTopicKafkaTemplate将被查找。
如果未找到 bean，则会引发异常。 |
从版本 3.0 开始，@RetryableTopicannotation 可以用作自定义 annotation 上的元 Annotation;例如：
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@RetryableTopic
static @interface MetaAnnotatedRetryableTopic {
@AliasFor(attribute = "concurrency", annotation = RetryableTopic.class)
String parallelism() default "3";
}用RetryTopicConfiguration豆
您还可以通过创建RetryTopicConfigurationbean 中的@Configurationannotated 类。
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, Object> template) {
return RetryTopicConfigurationBuilder
.newInstance()
.create(template);
}这将为带有@KafkaListener使用默认配置。这KafkaTemplateinstance 是消息转发所必需的。
为了实现对如何处理每个主题的非阻塞重试的更精细控制，需要多个RetryTopicConfigurationbean 的 bean 中。
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
return RetryTopicConfigurationBuilder
.newInstance()
.fixedBackOff(3000)
.maxAttempts(5)
.concurrency(1)
.includeTopics("my-topic", "my-other-topic")
.create(template);
}
@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<String, MyOtherPojo> template) {
return RetryTopicConfigurationBuilder
.newInstance()
.exponentialBackoff(1000, 2, 5000)
.maxAttempts(4)
.excludeTopics("my-topic", "my-other-topic")
.retryOn(MyException.class)
.create(template);
}重试主题和 dlt 的使用者将被分配给一个使用者组，该使用者组的组 ID 是您在groupId参数的@KafkaListener注释。
如果您不提供任何 URL，它们都将属于同一个组，并且对重试主题进行再平衡将导致对主主题进行不必要的再平衡。 |
如果使用者配置了ErrorHandlingDeserializer，要处理反序列化异常，请务必配置KafkaTemplate及其 producer 及其序列化器，该序列化器可以处理普通对象和原始对象byte[]值，这是由反序列化异常引起的。
模板的 generic value type 应为Object.
一种方法是使用DelegatingByTypeSerializer;示例如下： |
@Bean
public ProducerFactory<String, Object> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfiguration(), new StringSerializer(),
new DelegatingByTypeSerializer(Map.of(byte[].class, new ByteArraySerializer(),
MyNormalObject.class, new JsonSerializer<Object>())));
}
@Bean
public KafkaTemplate<String, Object> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}倍数@KafkaListenerAnnotations 可以用于同一主题，无论是否手动分配分区以及非阻塞重试，但给定主题只能使用一种配置。
最好使用单个RetryTopicConfigurationbean 用于配置此类主题;如果多个@RetryableTopic注释用于同一主题，则所有注释都应该具有相同的值，否则其中一个注释将应用于该主题的所有侦听器，而其他注释的值将被忽略。 |
配置全局设置和功能
从 2.9 开始，之前用于配置组件的 bean 覆盖方法已被删除（由于上述 API 的实验性质，没有弃用）。
这不会更改RetryTopicConfigurationBean 方法 - 仅基础结构组件的配置。
现在，RetryTopicConfigurationSupport类应该扩展为 （single）@Configurationclass 和正确的方法被覆盖。
示例如下：
@EnableKafka
@Configuration
public class MyRetryTopicConfiguration extends RetryTopicConfigurationSupport {
@Override
protected void configureBlockingRetries(BlockingRetriesConfigurer blockingRetries) {
blockingRetries
.retryOn(MyBlockingRetriesException.class, MyOtherBlockingRetriesException.class)
.backOff(new FixedBackOff(3000, 3));
}
@Override
protected void manageNonBlockingFatalExceptions(List<Class<? extends Throwable>> nonBlockingFatalExceptions) {
nonBlockingFatalExceptions.add(MyNonBlockingException.class);
}
@Override
protected void configureCustomizers(CustomizersConfigurer customizersConfigurer) {
// Use the new 2.9 mechanism to avoid re-fetching the same records after a pause
customizersConfigurer.customizeErrorHandler(eh -> {
eh.setSeekAfterError(false);
});
}
}使用此配置方法时，@EnableKafkaRetryTopic不应使用 annotation 来防止上下文由于重复的 bean 而无法启动。
使用简单的@EnableKafkaannotation 来代替。 |
什么时候autoCreateTopics为 true，则将使用指定数量的分区和复制因子创建 main 和 retry 主题。
从版本 3.0 开始，默认复制因子为-1，表示使用代理默认值。
如果您的代理版本低于 2.4，则需要设置一个显式值。
要覆盖特定主题（例如主题或 DLT）的这些值，只需添加NewTopic @Bean具有必需的属性;这将覆盖 Auto Creation 属性。
| 默认情况下，使用接收记录的原始分区将记录发布到重试主题。 如果重试主题的分区数少于主主题的分区数，则应适当配置框架;下面是一个示例。 |
@EnableKafka
@Configuration
public class Config extends RetryTopicConfigurationSupport {
@Override
protected Consumer<DeadLetterPublishingRecovererFactory> configureDeadLetterPublishingContainerFactory() {
return dlprf -> dlprf.setPartitionResolver((cr, nextTopic) -> null);
}
...
}该函数的参数是使用者记录和下一个主题的名称。
您可以返回特定的分区编号，或者null以指示KafkaProducer应确定分区。
默认情况下，当记录通过重试主题转换时，将保留重试标头的所有值（尝试次数、时间戳）。
从版本 2.9.6 开始，如果你只想保留这些 Headers 的最后一个值，请使用configureDeadLetterPublishingContainerFactory()方法将工厂的retainAllRetryHeaderValuesproperty 设置为false.
查找 RetryTopicConfiguration
尝试提供RetryTopicConfiguration方法是从@RetryableTopic注解，或者如果没有可用的注解，则从 Bean 容器中获取。
如果在容器中找到 bean，则会进行检查以确定所提供的主题是否应由任何此类实例处理。
如果@RetryableTopic注解时，会提供DltHandlerannotated 方法。
从 3.2 开始，提供新的 API 来创建RetryTopicConfiguration什么时候@RetryableTopic在 class 上注释：
@Bean
public RetryTopicConfiguration myRetryTopic() {
RetryTopicConfigurationProvider provider = new RetryTopicConfigurationProvider(beanFactory);
return provider.findRetryConfigurationFor(topics, null, AnnotatedClass.class, bean);
}
@RetryableTopic
public static class AnnotatedClass {
// NoOps
}