跳到主要内容
版本：4.1.1

功能

Doubao 中英对照 Features

大多数功能同时适用于 @RetryableTopic 注解和 RetryTopicConfiguration bean。

BackOff 配置​

BackOff 配置依赖于 Spring Retry 项目的 BackOffPolicy 接口。

它包括：

  • 固定退避（Fixed Back Off）
  • 指数退避（Exponential Back Off）
  • 随机指数退避（Random Exponential Back Off）
  • 均匀随机退避（Uniform Random Back Off）
  • 无退避（No Back Off）
  • 自定义退避（Custom Back Off）
@RetryableTopic(attempts = 5,
    backOff = @BackOff(delay = 1000, multiplier = 2, maxDelay = 5000))
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .fixedBackOff(3_000)
            .maxAttempts(4)
            .create(template);
}

你也可以提供 Spring Retry 的 SleepingBackOffPolicy 接口的自定义实现：

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .customBackoff(new MyCustomBackOffPolicy())
            .maxAttempts(5)
            .create(template);
}
备注

默认退避策略是 FixedBackOffPolicy，最多 3 次尝试、间隔 1000ms。

备注

ExponentialBackOffPolicy 有一个 30 秒的默认最大延迟。如果你的退避策略需要更大的延迟值，请相应调整 maxDelay 属性。

important

第一次尝试计入 maxAttempts，因此如果你提供的 maxAttempts 值为 4，那么将有最初的一次尝试加 3 次重试。

全局超时​

你可以为重试过程设置全局超时。如果达到该时间，下次消费者抛出异常时，消息将直接进入 DLT；如果没有 DLT，则直接结束处理。

@RetryableTopic(backOff = @BackOff(2_000), timeout = 5_000)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .fixedBackOff(2_000)
            .timeoutAfter(5_000)
            .create(template);
}
备注

默认是不设置超时，也可以通过将超时值设为 -1 来实现。

异常分类器​

你可以指定要重试哪些异常、不重试哪些异常。你还可以设置遍历原因以查找嵌套异常。

@RetryableTopic(include = {MyRetryException.class, MyOtherRetryException.class}, traversingCauses = "true")
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    throw new RuntimeException(new MyRetryException()); // will retry
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyOtherPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .notRetryOn(MyDontRetryException.class)
            .create(template);
}
备注

默认行为是对所有异常都重试，且不遍历原因。

自 2.8.3 起，有一个全局致命异常列表，这些异常会导致记录不经任何重试直接发送到 DLT。默认致命异常列表请参阅 DefaultErrorHandler。你可以通过在一个扩展了 RetryTopicConfigurationSupport 的 @Configuration 类中覆盖 configureNonBlockingRetries 方法，向此列表添加或移除异常。更多信息请参阅配置全局设置与功能。

@Override
protected void manageNonBlockingFatalExceptions(List<Class<? extends Throwable>> nonBlockingFatalExceptions) {
    nonBlockingFatalExceptions.add(MyNonBlockingException.class);
}
备注

要禁用致命异常分类，只需清空所提供的列表。

包含与排除主题​

你可以通过 .includeTopic(String topic)、.includeTopics(Collection<String> topics)、.excludeTopic(String topic) 和 .excludeTopics(Collection<String> topics) 方法来决定哪些主题由某个 RetryTopicConfiguration bean 处理、哪些不处理。

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .includeTopics(List.of("my-included-topic", "my-other-included-topic"))
            .create(template);
}

@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .excludeTopic("my-excluded-topic")
            .create(template);
}
备注

默认行为是包含所有主题。

主题自动创建​

除非另行指定，框架将使用 KafkaAdmin bean 所消费的 NewTopic bean 自动创建所需的主题。你可以指定创建主题时使用的分区数和副本因子，也可以关闭此功能。从 3.0 版本开始，默认副本因子为 -1，表示使用 broker 默认值。如果你的 broker 版本早于 2.4，则需要设置一个显式值。

important

请注意，如果你没有使用 Spring Boot，则必须提供一个 KafkaAdmin bean 才能使用此功能。

@RetryableTopic(numPartitions = "2", replicationFactor = "3")
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}

@RetryableTopic(autoCreateTopics = "false")
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .autoCreateTopicsWith(2, 3)
            .create(template);
}

@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .doNotAutoCreateRetryTopics()
            .create(template);
}
备注

默认情况下，主题自动创建为一个分区，副本因子为 -1（表示使用 broker 默认值）。如果你的 broker 版本早于 2.4，则需要设置一个显式值。

失败头管理​

在考虑如何管理失败头（原始头和异常头）时，框架委托给 DeadLetterPublishingRecoverer 来决定是追加还是替换这些头。

默认情况下，它显式地将 appendOriginalHeaders 设为 false，而 stripPreviousExceptionHeaders 则沿用 DeadLetterPublishingRecover 使用的默认值。

这意味着在默认配置下，只保留第一个"原始"头和最后一个异常头。这是为了在涉及多个重试步骤时避免创建过大的消息（例如由于堆栈跟踪头）。

更多信息请参阅管理死信记录头。

要重新配置框架为这些属性使用不同的设置，可通过在一个扩展了 RetryTopicConfigurationSupport 的 @Configuration 类中覆盖 configureCustomizers 方法来配置一个 DeadLetterPublishingRecoverer 定制器。更多信息请参阅配置全局设置与功能。

@Override
protected void configureCustomizers(CustomizersConfigurer customizersConfigurer) {
    customizersConfigurer.customizeDeadLetterPublishingRecoverer(dlpr -> {
        dlpr.setAppendOriginalHeaders(true);
        dlpr.setStripPreviousExceptionHeaders(false);
    });
}

从 2.8.4 版本开始，如果你希望添加自定义头（除了工厂添加的重试信息头之外），可以向工厂添加一个 headersFunction —— factory.setHeadersFunction((rec, ex) -> { ... })。

默认情况下，添加的任何头都是累积的——Kafka 头可以包含多个值。从 2.9.5 版本开始，如果函数返回的 Headers 包含一个 DeadLetterPublishingRecoverer.SingleRecordHeader 类型的头，则该头的所有现有值都将被移除，只保留新的单个值。

自定义 DeadLetterPublishingRecoverer​

如失败头管理中所示，可以对框架创建的默认 DeadLetterPublishingRecoverer 实例进行定制。不过，对于某些用例，需要继承 DeadLetterPublishingRecoverer，例如覆盖 createProducerRecord() 以修改发送到重试（或死信）主题的内容。从 3.0.9 版本开始，你可以覆盖 RetryTopicConfigurationSupport.configureDeadLetterPublishingContainerFactory() 方法来提供一个 DeadLetterPublisherCreator 实例，例如：

@Override
protected Consumer<DeadLetterPublishingRecovererFactory>
        configureDeadLetterPublishingContainerFactory() {

    return (factory) -> factory.setDeadLetterPublisherCreator(
            (templateResolver, destinationResolver) ->
                    new CustomDLPR(templateResolver, destinationResolver));
}

建议你在构建自定义实例时使用所提供的解析器。

基于抛出异常将消息路由到自定义 DLT​

从 3.2.0 版本开始，可以根据消息处理期间抛出的异常类型，将消息路由到自定义 DLT。为此，需要指定路由。路由定制由附加目标的定义组成。目标又由两个设置构成：suffix 和 exceptions。当 exceptions 中指定的异常类型被抛出时，包含该 suffix 的 DLT 将被视为消息的目标主题，之后才会考虑通用 DLT。以下是使用注解或 RetryTopicConfiguration bean 的配置示例：

@RetryableTopic(exceptionBasedDltRouting = {
    @ExceptionBasedDltDestination(
        suffix = "-deserialization", exceptions = {DeserializationException.class}
    )}
)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .dltRoutingRules(Map.of("-deserialization", Set.of(DeserializationException.class)))
            .create(template);
}

suffix 在自定义 DLT 名称中位于通用 dltTopicSuffix 之前。就所给示例而言，导致 DeserializationException 的消息将被路由到 my-annotated-topic-deserialization-dlt，而不是 my-annotated-topic-dlt。自定义 DLT 将按照主题自动创建中所述的相同规则创建。