Retry and Dead Letter Processing

当重试和延迟的总和超过消费者的 max.poll.interval.ms 属性，或者需要将死信发布到不同的 Kafka 集群，或者需要将重试监听器添加到错误处理程序时，就需要进行此移动。

文章提供了如何配置将这些功能从 Binder 移至容器的说明，包括如何定义 @Bean 和如何使用 BinderCustomizer 设置容器自定义器。

默认情况下，当你配置消费者绑定中的重试（例如，maxAttemts）和 enableDlq 时，这些函数将在 Binder 内执行，而侦听器容器或 Kafka 消费者不会参与。

在某些情况下，最好将此功能移至侦听器容器，例如：

  • 重试、延时的总和将超过消费者的 max.poll.interval.ms 属性，可能会导致分区重新平衡。

  • 您希望将死信发布到其他 Kafka 集群。

  • 您希望对错误处理程序中添加重试侦听器。

  • …​

要配置将此功能从 Binder 移至容器，请定义类型为 ListenerContainerWithDlqAndRetryCustomizer 的 @Bean。此接口具有以下方法：

/**
 * Configure the container.
 * @param container the container.
 * @param destinationName the destination name.
 * @param group the group.
 * @param dlqDestinationResolver a destination resolver for the dead letter topic (if
 * enableDlq).
 * @param backOff the backOff using retry properties (if configured).
 * @see #retryAndDlqInBinding(String, String)
 */
void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName, String group,
        @Nullable BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> dlqDestinationResolver,
        @Nullable BackOff backOff);

/**
 * Return false to move retries and DLQ from the binding to a customized error handler
 * using the retry metadata and/or a {@code DeadLetterPublishingRecoverer} when
 * configured via
 * {@link #configure(AbstractMessageListenerContainer, String, String, BiFunction, BackOff)}.
 * @param destinationName the destination name.
 * @param group the group.
 * @return false to disable retries and DLQ in the binding
 */
default boolean retryAndDlqInBinding(String destinationName, String group) {
    return true;
}

目标解析器和 BackOff 是通过绑定属性（如果已配置）创建的。KafkaTemplate 使用 spring.kafka…​. 属性中的配置。然后，你可以使用它们创建自定义错误处理程序和死信发布者；例如：

@Bean
ListenerContainerWithDlqAndRetryCustomizer cust(KafkaTemplate<?, ?> template) {
    return new ListenerContainerWithDlqAndRetryCustomizer() {

        @Override
        public void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName,
                String group,
                @Nullable BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> dlqDestinationResolver,
                @Nullable BackOff backOff) {

            if (destinationName.equals("topicWithLongTotalRetryConfig")) {
                ConsumerRecordRecoverer dlpr = new DeadLetterPublishingRecoverer(template,
                        dlqDestinationResolver);
                container.setCommonErrorHandler(new DefaultErrorHandler(dlpr, backOff));
            }
        }

        @Override
        public boolean retryAndDlqInBinding(String destinationName, String group) {
            return !destinationName.contains("topicWithLongTotalRetryConfig");
        }

    };
}

现在，只需一个重试延迟需要大于消费者的 max.poll.interval.ms 属性。

与多个绑定一起工作时，“DefaultBinderFactory” 会覆盖 ListenerContainerWithDlqAndRetryCustomizer bean。要应用 Bean，你需要使用 BinderCustomizer 设置容器自定义器（参见 [binder-customizer]）：

@Bean
public BinderCustomizer binderCustomizer(ListenerContainerWithDlqAndRetryCustomizer containerCustomizer) {
    return (binder, binderName) -> {
        if (binder instanceof KafkaMessageChannelBinder kafkaMessageChannelBinder) {
            kafkaMessageChannelBinder.setContainerCustomizer(containerCustomizer);
        }
        else if (binder instanceof KStreamBinder) {
            ...
        }
        else if (binder instanceof RabbitMessageChannelBinder) {
            ...
        }
    };
}