主页 / coding / 问题 / 79147403
Accepted
Rayyan
Asked: 2024-11-01 16:50:34 +0800 CST

Spring Integration Kafka 延迟重试

  • 772

用例是使用 Spring Integration 的 Kafka 延迟重新处理下游失败消息ConcurrentMessageListenerContainer。假设最大重试次数应为 2，固定延迟为 5 分钟。

Spring Integration 框架中是否有现成的解决方案？

已经过去了DeadLetterPublishingRecoverer，它只有助于转移到 DLT，除非我们用轮询器监听同一个 DLT，否则不会进行进一步处理。

spring-integration

1 个回答

  1. Best Answer

    如果您研究一下 Spring Integration 解决方案，那么KafkaMessageDrivenChannelAdapter使用上述内容ConcurrentMessageListenerContainer可以配置RetryTemplate：

    /**
 * Specify a {@link RetryTemplate} instance to use for retrying deliveries.
 * <p>
 * IMPORTANT: This form of retry is blocking and could cause a rebalance if the
 * aggregate retry delays across all polled records might exceed the
 * {@code max.poll.interval.ms}. Instead, consider adding a
 * {@code DefaultErrorHandler} to the listener container, configured with a
 * {@link KafkaErrorSendingMessageRecoverer}.
 * @param retryTemplate the {@link RetryTemplate} to use.
 */
public void setRetryTemplate(RetryTemplate retryTemplate) {

    Spring Integration 的另一种方法是在处理下游消息的位置进行重试。为此，可以使用以下方式配置特定端点RequestHandlerRetryAdvice：https ://docs.spring.io/spring-integration/reference/handler-advice/classes.html#retry-advice

    更新

    @EnableKafkaRetryTopicSpring Integration 中是否有等效项。

    不，没有办法@RetryableTopic在 Spring Integration 中应用KafkaMessageDrivenChannelAdapter。问题是这种非阻塞重试正是为声明式、基于注释的 POJO 设计的。其中通道适配器针对定期MessageListener注入目标侦听器容器进行了优化。

    我建议改为研究@KafkaListener一下@RetryableTopic，然后使用@MessagingGateway它从该侦听器方法调用 Spring Integration 流：https://docs.spring.io/spring-integration/reference/gateway.html。您甚至可能只是期望Message<?>在您的中有一个@KafkaListener并将其发送到注入的MessageChannel。

    • 2

相关问题