配置
从 2.9 版本开始，对于默认配置，应在 @Configuration 注释的类中使用 @EnableKafkaRetryTopic 注释。这使该功能能够正确启动，并允许注入一些功能组件以供在运行时查找。
如果添加了此注释，则不必再添加 @EnableKafka，因为 @EnableKafkaRetryTopic 使用 @EnableKafka 进行了元注释。
|
此外，从该版本开始，要对该功能的组件和全局功能进行更高级的配置，应在 @Configuration 类中扩展 RetryTopicConfigurationSupport 类，并重写适当的方法。有关更多详细信息，请参阅 配置全局设置和功能。
默认情况下，重试主题的容器将与主容器具有相同的并发性。从 3.0 版本开始，可以为重试容器设置不同的 concurrency（在注释中或在 RetryConfigurationBuilder 中）。
只能使用上述技术之一，并且只有一个 @Configuration 类可以扩展 RetryTopicConfigurationSupport。
|
使用 @RetryableTopic 注释
要为 @KafkaListener 注释的方法配置重试主题和 dlt，只需向其添加 @RetryableTopic 注释，Spring for Apache Kafka 就会使用默认配置引导所有必需的主题和使用者。
@RetryableTopic(kafkaTemplate = "myRetryableTopicKafkaTemplate")
@KafkaListener(topics = "my-annotated-topic", groupId = "myGroupId")
public void processMessage(MyPojo message) {
// ... message processing
}从 3.2 开始，类上 @KafkaListener 的 @RetryableTopic 支持将是
@RetryableTopic(listenerContainerFactory = "my-retry-topic-factory")
@KafkaListener(topics = "my-annotated-topic")
public class ClassLevelRetryListener {
@KafkaHandler
public void processMessage(MyPojo message) {
// ... message processing
}
}可以通过使用 @DltHandler 注释对同一类中的方法进行注释，来指定用于处理 dlt 消息的方法。如果没有提供 DltHandler 方法，则会创建一个默认使用者，该使用者只会记录使用情况。
@DltHandler
public void processMessage(MyPojo message) {
// ... message processing, persistence, etc
}
如果您未指定 kafkaTemplate 名称，则会查找名称为 defaultRetryTopicKafkaTemplate 的 bean。如果找不到 bean，则会引发异常。
|
从 3.0 版本开始，@RetryableTopic 注释可以用作自定义注释的元注释；例如
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@RetryableTopic
static @interface MetaAnnotatedRetryableTopic {
@AliasFor(attribute = "concurrency", annotation = RetryableTopic.class)
String parallelism() default "3";
}使用 RetryTopicConfiguration bean
您还可以在 @Configuration 注释的类中创建 RetryTopicConfiguration bean，以配置非阻塞重试支持。
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, Object> template) {
return RetryTopicConfigurationBuilder
.newInstance()
.create(template);
}这将为使用 @KafkaListener 注释的方法中的所有主题创建重试主题和 dlt，以及相应的使用者，并使用默认配置。消息转发需要 KafkaTemplate 实例。
为了更精细地控制如何针对每个主题处理非阻塞重试，可以提供多个 RetryTopicConfiguration 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 是您在 @KafkaListener 注释的 groupId 参数中提供的 ID 与主题后缀的组合。如果您没有提供任何信息，则它们都将属于同一组，并且在重试主题上重新平衡会导致在主主题上进行不必要的重新平衡。
|
如果消费者配置了 ErrorHandlingDeserializer 来处理反序列化异常，则配置 KafkaTemplate 及其生产者非常重要，它使用一个既能处理常规对象又能处理反序列化异常产生的原始 byte[] 值的序列化器。模板的泛型值类型应为 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());
}
对于同一主题，可以使用多个 @KafkaListener 注解，无论是否手动分配分区以及是否使用非阻塞重试，但给定主题只会使用一个配置。最好对这类主题使用一个 RetryTopicConfiguration bean 来进行配置；如果对同一主题使用多个 @RetryableTopic 注解，则所有注解都应具有相同的值，否则其中一个值将应用于该主题的所有侦听器，而其他注解的值将被忽略。
|
配置全局设置和功能
自 2.9 起，用于配置组件的先前 bean 覆盖方法已被移除（未弃用，因为上述 API 具有实验性质）。这不会更改 RetryTopicConfiguration bean 方法 - 仅更改基础设施组件的配置。现在应在（单个）@Configuration 类中扩展 RetryTopicConfigurationSupport 类，并覆盖适当的方法。示例如下
@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 注解，以防止由于重复 bean 而导致上下文启动失败。请改用简单的 @EnableKafka 注解。
|
当 autoCreateTopics 为 true 时，将使用指定数量的分区和复制因子创建主主题和重试主题。从版本 3.0 开始，默认复制因子为 -1，表示使用代理默认值。如果代理版本早于 2.4，则需要设置显式值。要覆盖特定主题（例如主主题或 DLT）的这些值，只需添加一个具有所需属性的 NewTopic @Bean；这将覆盖自动创建属性。
| 默认情况下，记录将使用接收记录的原始分区发布到重试主题。如果重试主题的分区少于主主题，则应适当地配置框架；示例如下。 |
@EnableKafka
@Configuration
public class Config extends RetryTopicConfigurationSupport {
@Override
protected Consumer<DeadLetterPublishingRecovererFactory> configureDeadLetterPublishingContainerFactory() {
return dlprf -> dlprf.setPartitionResolver((cr, nextTopic) -> null);
}
...
}该函数的参数是消费者记录和下一个主题的名称。您可以返回特定分区号或 null 以指示 KafkaProducer 应确定分区。
默认情况下，当记录通过重试主题时，重试标头的所有值（尝试次数、时间戳）都会保留。从 2.9.6 版本开始，如果你只想保留这些标头的最后一个值，请使用上面所示的 configureDeadLetterPublishingContainerFactory() 方法将工厂的 retainAllRetryHeaderValues 属性设置为 false。
查找 RetryTopicConfiguration
尝试通过 @RetryableTopic 注释或在没有注释可用时从 bean 容器中创建一个 RetryTopicConfiguration 实例。
如果在容器中找到 bean，则会检查提供的主题是否应由其中任何实例处理。
如果提供了 @RetryableTopic 注释，则会查找 DltHandler 注释的方法。
自 3.2 起，当 @RetryableTopic 注释在类上时，提供新的 API 来创建 RetryTopicConfiguration
@Bean
public RetryTopicConfiguration myRetryTopic() {
RetryTopicConfigurationProvider provider = new RetryTopicConfigurationProvider(beanFactory);
return provider.findRetryConfigurationFor(topics, null, AnnotatedClass.class, bean);
}
@RetryableTopic
public static class AnnotatedClass {
// NoOps
}
