对于最新的稳定版本,请使用 Spring for Apache Kafka 3.3.9spring-doc.cadn.net.cn

主题命名

重试主题和 DLT 的命名方法是在主主题后缀为提供的或默认值,并附加该主题的延迟或索引。spring-doc.cadn.net.cn

“我的主题” → “我的主题重试-0”、“我的主题重试-1”, ..., “我的主题-dlt”spring-doc.cadn.net.cn

“my-other-topic” → “my-topic-myRetrySuffix-1000”, “my-topic-myRetrySuffix-2000”, ..., “my-topic-myDltSuffix”spring-doc.cadn.net.cn

默认行为是为每次尝试创建单独的重试主题,并附加索引值:retry-0、retry-1、...、retry-n。因此,默认情况下,重试主题数为maxAttempts减 1。

重试主题和 DLT 后缀

可以指定重试和 DLT 主题将使用的后缀。spring-doc.cadn.net.cn

@RetryableTopic(retryTopicSuffix = "-my-retry-suffix", dltTopicSuffix = "-my-dlt-suffix")
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyOtherPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .retryTopicSuffix("-my-retry-suffix")
            .dltTopicSuffix("-my-dlt-suffix")
            .create(template);
}
默认后缀是“-retry”和“-dlt”,分别用于重试主题和 dlt。

附加主题的索引或延迟

可以在后缀后附加主题的索引或延迟值。spring-doc.cadn.net.cn

@RetryableTopic(topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .suffixTopicsWithIndexValues()
            .create(template);
    }
默认行为是以延迟值为后缀,但具有多个主题的固定延迟配置除外,在这种情况下,主题以主题的索引为后缀。

固定延迟重试的单个主题

如果您使用的是固定延迟策略,例如FixedBackOffPolicyNoBackOffPolicy可以使用单个主题来完成非阻塞重试。此主题将以提供的或默认后缀为后缀,并且不会附加索引或延迟值。spring-doc.cadn.net.cn

前一个FixedDelayStrategy现已弃用,可以替换为SameIntervalTopicReuseStrategy.
@RetryableTopic(backoff = @Backoff(2_000), fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC)
@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(5)
            .useSingleTopicForFixedDelays()
            .create(template);
}
默认行为是为每次尝试创建单独的重试主题,并附加其索引值:retry-0、retry-1、...

maxInterval 指数延迟的单个主题

如果您使用的是指数退避策略 (ExponentialBackOffPolicy),您可以使用单个重试主题来完成延迟配置为maxInterval.spring-doc.cadn.net.cn

此“最终”重试主题将以提供的或默认后缀为后缀,并且将具有索引或maxInterval值附加。spring-doc.cadn.net.cn

通过选择使用单个主题进行重试,并使用maxInterval延迟,配置一个长时间持续重试的指数重试策略可能会变得更加可行,因为在这种方法中,你不需要大量的主题。

默认行为是使用等于配置的maxAttempts减去 1,并且当使用指数退避时,重试主题以延迟值为后缀,最后一个重试主题(对应于maxIntervaldelay)后缀为附加索引。spring-doc.cadn.net.cn

例如,当使用initialInterval=1_000,multiplier=2maxInterval=16_000,为了继续尝试一小时,需要配置maxAttempts作为 229,默认情况下,所需的重试主题为:spring-doc.cadn.net.cn

当使用在相同时间间隔内重用重试主题的策略时,在上述相同配置中,所需的重试主题将是:spring-doc.cadn.net.cn

这将是未来版本中的默认设置。spring-doc.cadn.net.cn

@RetryableTopic(attempts = 230,
    backoff = @Backoff(delay = 1_000, multiplier = 2, maxDelay = 16_000),
    sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
    // ... message processing
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .exponentialBackoff(1_000, 2, 16_000)
            .maxAttempts(230)
            .useSingleTopicForSameIntervals()
            .create(template);
}

自定义命名策略

更复杂的命名策略可以通过注册实现RetryTopicNamesProviderFactory. 默认实现是SuffixingRetryTopicNamesProviderFactory可以通过以下方式注册不同的实现:spring-doc.cadn.net.cn

@Override
protected RetryTopicComponentFactory createComponentFactory() {
    return new RetryTopicComponentFactory() {
        @Override
        public RetryTopicNamesProviderFactory retryTopicNamesProviderFactory() {
            return new CustomRetryTopicNamesProviderFactory();
        }
    };
}

例如,除了标准后缀之外,以下实现还向重试/dlt 主题名称添加了前缀:spring-doc.cadn.net.cn

public class CustomRetryTopicNamesProviderFactory implements RetryTopicNamesProviderFactory {

    @Override
    public RetryTopicNamesProvider createRetryTopicNamesProvider(
                DestinationTopic.Properties properties) {

        if (properties.isMainEndpoint()) {
            return new SuffixingRetryTopicNamesProvider(properties);
        }
        else {
            return new SuffixingRetryTopicNamesProvider(properties) {

                @Override
                public String getTopicName(String topic) {
                    return "my-prefix-" + super.getTopicName(topic);
                }

            };
        }
    }

}