配置

Doubao 中英对照 Configuration

对于默认设置，通过在 @KafkaListener 方法上添加 @RetryableTopic 注解来启用非阻塞重试。这是推荐且最简单的方式，因为它会自动配置所需的重试基础设施，并使用默认设置创建重试主题和 DLT 主题。

要导入非阻塞重试基础设施并将其组件暴露为 bean，请在一个 @Configuration 类上标注 @EnableKafkaRetryTopic。这将启用该功能组件的注入和运行时查找，并作为高级和全局配置的基础。

备注

如果添加了此注解，就不必再添加 @EnableKafka，因为 @EnableKafkaRetryTopic 已用 @EnableKafka 进行了元注解。

要进行高级和全局定制，请在单个 @Configuration 类中扩展 RetryTopicConfigurationSupport 并覆盖相关方法。更多细节请参阅配置全局设置与功能。

默认情况下，重试主题的容器与主容器具有相同的并发度。从 3.0 版本开始，你可以为重试容器设置不同的 concurrency（在注解上或在 RetryTopicConfigurationBuilder 中设置）。

important

上述两种全局配置方式（@EnableKafkaRetryTopic 或扩展 RetryTopicConfigurationSupport）只使用其一。此外，应该只有一个 @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 起，@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。如果找不到该 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(List.of("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(List.of("my-topic", "my-other-topic"))
            .retryOn(MyException.class)
            .create(template);
}

备注

重试主题和 dlt 的消费者将被分配到一个消费者组，该组的 group id 由你在 @KafkaListener 注解的 groupId 参数中提供的值与主题后缀组合而成。如果你未提供任何值，它们都将属于同一个组，这样重试主题上的重平衡会导致主主题上不必要的重平衡。

important

如果消费者配置了 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 JacksonJsonSerializer<Object>())));
}

@Bean
public KafkaTemplate<String, Object> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
}

important

多个 @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);
        });
    }

}

important

使用此配置方式时，不应使用 @EnableKafkaRetryTopic 注解，以避免因 bean 重复而导致上下文启动失败。请改用简单的 @EnableKafka 注解。

当 autoCreateTopics 为 true 时，主主题和重试主题将按指定的分区数和副本因子创建。从 3.0 版本开始，默认副本因子为 -1，表示使用 broker 默认值。如果你的 broker 版本早于 2.4，则需要设置一个显式值。要为特定主题（例如主主题或 DLT）覆盖这些值，只需添加一个具有所需属性的 NewTopic @Bean；这将覆盖自动创建属性。

important

默认情况下，记录使用接收记录的原始分区发布到重试主题。如果重试主题的分区数少于主主题，你应相应地配置框架；示例如下。

@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

尝试提供 RetryTopicConfiguration 的实例：要么从 @RetryableTopic 注解创建一个，要么在没有注解时从 bean 容器中获取。

如果在容器中找到 bean，会进行检查以确定所提供的主题是否应由这些实例中的某一个处理。

如果提供了 @RetryableTopic 注解，则查找一个带 DltHandler 注解的方法。

自 3.2 起，提供了新 API，用于在 @RetryableTopic 注解于类上时创建 RetryTopicConfiguration：

@Bean
public RetryTopicConfiguration myRetryTopic() {
    RetryTopicConfigurationProvider provider = new RetryTopicConfigurationProvider(beanFactory);
    return provider.findRetryConfigurationFor(topics, null, AnnotatedClass.class, bean);
}

@RetryableTopic
public static class AnnotatedClass {
    // NoOps
}