如何重试来自Kafka的失败消息？

Question

我的Spring Boot应用程序（消费者）处理来自Apache Kafka的消息。周期性地，按摩无法处理，消费者抛出异常。不管怎样，消费者还是会做出补偿。在Kafka中，我能区分成功消息和失败消息吗？我想，我不能。这是真的吗？如果这是真的，我有一个主要问题：

如何重试失败消息？我知道一些方法，但我不确定它们是否正确。

1） 将“偏移”更改为“提前”。但通过这种方式，成功消息也会重试。

2） 当我捕捉到异常时，我会将此消息发送到另一个主题（例如错误主题）。但这看起来很难。

3）其他东西（你的变体）

Answer 1

您可以在您的消费者配置中进行以下更改：

>

启用。汽车提交=错误

RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();
    FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
    fixedBackOffPolicy.setBackOffPeriod(retryInterval);
    retryTemplate.setBackOffPolicy(fixedBackOffPolicy);
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(retryMaxCount));

    return retryTemplate;
}

在您的kafkaListenerContainerFactory：

setretryTemplate(retryTemplate);
factory.getContainerProperties().setAckOnError(true);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);

现在，在代码中，只要在异常发生时在使用者中抛出异常即可。这不会在发生异常时更新偏移量。它将在retryInterval时间后重试，最长时间为maxRetryCount。

如果您想忽略某些类型的异常，而不是重试，请创建一个异常映射，如下图所示，并将其传递到SimpleRetryPolicy（）中，如下图所示：

Map<Class<? extends Throwable>, Boolean> exceptionMap = new HashMap<>();
        exceptionMap.put(IllegalArgumentException.class, false);
        exceptionMap.put(TimeoutException.class, true);

有关更多详细信息，请访问此链接：Kafka错误处理

Answer 2

使用SeekToCurInterrorHandler。它将重新定位偏移量以重播消息（默认情况下为10次，但可配置）。

重试用尽后，它调用可以执行某些操作的“恢复者”，例如DeadLetterPublishingRecoverer。

Answer 3

如果您希望至少保证一次，一般模式如下：

>

禁用自动提交（将enable.auto.commit设置为false）

消费信息

对于每条消息：

• 如果没有错误，则提交偏移量
• 如果出现错误，请重试任意次数
• 如果成功，请提交
• 如果要放弃，请登录或发布到错误队列（用于分析或稍后重试）

重复