对于最新的稳定版本,请使用 Spring for Apache Kafka 3.3.0! |
主题命名
重试主题和 DLT 的命名方式是在主主题后加上提供的或默认值,并附加该主题的延迟或索引。
例子:
“my-topic” → “my-topic-retry-0”, “my-topic-retry-1”, ..., “my-topic-dlt”
“my-other-topic” → “my-topic-myRetrySuffix-1000”, “my-topic-myRetrySuffix-2000”, ..., “my-topic-myDltSuffix”
默认行为是为每次尝试创建单独的重试主题,并附加索引值:retry-0、retry-1、...、retry-n。
因此,默认情况下,重试主题的数量为配置的减 1。maxAttempts |
您可以配置后缀,选择是附加尝试索引还是延迟,在使用固定回退时使用单个重试主题,以及在使用指数回退时对具有 maxInterval 的尝试使用单个重试主题。
重试主题和 DLT 后缀
您可以指定重试和 DLT 主题将使用的后缀。
@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。 |
附加主题的索引或延迟
您可以在后缀后附加主题的索引或 delay 值。
@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);
}
默认行为是使用延迟值作为后缀,但具有多个主题的固定延迟配置除外,在这种情况下,主题以主题的索引为后缀。 |
固定延迟重试的单个主题
如果您使用的是固定延迟策略,例如 or,则可以使用单个主题来完成非阻塞重试。
本主题将以 provided 或 default 后缀为后缀,并且不会附加 index 或 delay 值。FixedBackOffPolicy
NoBackOffPolicy
前者现已弃用,可替换为 。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 Exponential Delay 的单个主题
如果您使用的是指数回退策略 (),则可以使用单个重试主题来完成其延迟为配置的尝试的非阻塞重试。ExponentialBackOffPolicy
maxInterval
此“最终”重试主题将以 provided 或 default 后缀为后缀,并将附加索引或值。maxInterval
通过选择使用单个主题进行延迟重试,配置长时间持续重试的指数重试策略可能变得更加可行,因为在这种方法中,您不需要大量主题。maxInterval |
从 3.2 开始,默认行为是以相同的间隔重复使用重试主题,当使用指数退避时,重试主题以延迟值为后缀,最后一个重试主题以相同的间隔(对应于延迟)重复使用。maxInterval
例如,当使用 、 和 配置指数回退时,为了继续尝试一小时,需要配置为 229,默认情况下,所需的重试主题为:initialInterval=1_000
multiplier=2
maxInterval=16_000
maxAttempts
-
-重试 1000
-
-重试 2000
-
-重试 4000
-
-重试 8000
-
-重试 16000
当使用重试主题数等于配置的减 1 的策略时,最后一个以附加索引为后缀的重试主题(对应于延迟)将为:maxAttempts
maxInterval
-
-重试 1000
-
-重试 2000
-
-重试 4000
-
-重试 8000
-
-重试-16000-0
-
-重试 16000-1
-
-重试-16000-2
-
…
-
-重试-16000-224
如果需要多个主题,则可以使用以下配置来完成。
@RetryableTopic(attempts = 230,
backoff = @Backoff(delay = 1_000, multiplier = 2, maxDelay = 16_000),
sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS)
@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
@Override
protected RetryTopicComponentFactory createComponentFactory() {
return new RetryTopicComponentFactory() {
@Override
public RetryTopicNamesProviderFactory retryTopicNamesProviderFactory() {
return new CustomRetryTopicNamesProviderFactory();
}
};
}
例如,除了标准后缀之外,以下实现还为 retry/dlt 主题名称添加了前缀:
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);
}
};
}
}
}