主题命名

Doubao 中英对照 Topic Naming

重试主题和 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"

备注

从 4.1 版本开始，默认行为是对相同的延迟间隔复用同一个重试主题。要为每次尝试创建独立的重试主题，请将 sameIntervalTopicReuseStrategy 设置为 MULTIPLE_TOPICS。

你可以配置后缀、选择是追加尝试索引还是延迟、在使用固定退避时使用单个重试主题，以及在使用指数退避时对具有 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"。

追加主题索引或延迟

你可以在后缀之后追加主题索引值或延迟值。

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

备注

默认行为是用延迟值作为后缀，但多主题的固定延迟配置除外，这种情况下主题用主题索引作为后缀。

固定延迟重试的单主题

如果你使用的是固定延迟策略，例如 FixedBackOffPolicy 或 NoBackOffPolicy，你可以使用单个主题来完成非阻塞重试。该主题将附加提供的或默认的后缀，而不会追加索引或延迟值。

备注

原有的 FixedDelayStrategy 现已弃用，可以用 SameIntervalTopicReuseStrategy 替代。

@RetryableTopic(backOff = @BackOff(2_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()
            .fixedBackOff(3_000)
            .maxAttempts(5)
            .useSingleTopicForSameIntervals()
            .create(template);
}

备注

从 4.1 版本开始，默认行为是对固定延迟重试使用单个主题。要使用多个主题，请将 sameIntervalTopicReuseStrategy 设置为 MULTIPLE_TOPICS。

maxInterval 指数延迟的单主题

如果你使用的是指数退避策略（ExponentialBackOffPolicy），你可以使用单个重试主题来完成那些延迟为已配置 maxInterval 的尝试的非阻塞重试。

这个"最终"重试主题将附加提供的或默认的后缀，并追加索引或 maxInterval 值。

备注

通过对具有 maxInterval 延迟的重试选择使用单个主题，配置一个长时间持续重试的指数重试策略可能会变得更可行，因为这种方式不需要大量主题。

从 3.2 开始，默认行为是对相同的间隔复用重试主题；使用指数退避时，重试主题用延迟值作为后缀，最后一个重试主题对相同的间隔复用（对应于 maxInterval 延迟）。

例如，在配置指数退避为 initialInterval=1_000、multiplier=2、maxInterval=16_000 时，为了持续尝试一小时，需要将 maxAttempts 配置为 229，默认情况下所需的重试主题为：

• -retry-1000
• -retry-2000
• -retry-4000
• -retry-8000
• -retry-16000

当使用重试主题数量等于已配置的 maxAttempts 减 1 的策略时，最后一个重试主题（对应于 maxInterval 延迟）会附加一个额外的索引，如下所示：

• -retry-1000
• -retry-2000
• -retry-4000
• -retry-8000
• -retry-16000-0
• -retry-16000-1
• -retry-16000-2
• …​
• -retry-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 的 bean 来实现。默认实现是 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);
                }

            };
        }
    }

}