从另一个维护整个重试逻辑的侦听器内部调用重试侦听器

Question

我们正在使用 spring-kafka-2.2.8.RELEASE。我有需要帮助的特殊情况。我有 4 个主题 topic、retryTopic、successTopic 和 errorTopic。 如果 topic 失败，应重定向到 retryTopic 3 次重试尝试的位置制成。如果这些尝试失败，则必须重定向到 errorTopic。如果 topic 和 retryTopic 都成功，应重定向到 sucessTopic。这种情况已经基于问题How to retry with spring kafka version 2..2实现了。 但是现在，我遇到了一种新情况，我需要根据业务逻辑错误从 topic 侦听器内部调用 retryTopic 侦听器，而不会引发异常（它已经调用retryTopic 抛出异常并且它必须保持这种行为）。而且我还需要知道 retryTopic 在哪个重试尝试次数上被称为以下侦听器的参数。

 @KafkaListener(id = "so60172304.2", topics = "retryTopic")
 public void listen2(String in) {
            RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.execute(new RetryCallback<Void, RuntimeException>() {
        @Override
        public Void doWithRetry(RetryContext retryContext) throws RuntimeException {
             // Can I get the retry count here? It didn't work
             Integer count =RetrySynchronizationManager.getContext().getRetryCount());
            return this.doWithRetry(retryContext);
        }
    }); 
  }

Answer 1

您没有理由不能从另一个侦听器调用一个侦听器（但除非您在第一种方法中使用 RetryTemplate 调用它，否则您不会重试）。

如果您使用容器工厂上配置的RetryTemplate 进行重试（而不是在 2.3.x 及更高版本中将BackOff 添加到SeektoCurrentErrorHandler），您可以获得重试计数（从零）像这样......

@KafkaListener(id = "so60172304.2", topics = "retryTopic")
public void listen2(String in) {
    int retryCount = RetrySynchronizationManager.getContext().getRetryCount();
    ...
}

如果您直接从第一个方法调用它，getContext() 将返回 null（除非您将调用包装在 RetryTemplate.execute() 中）。

在 2.5.x 中，即使使用 SeektoCurrentErrorHandler 和 BackOff 而不是在容器工厂中使用 RetryTemplate，也会（可选）提供传递尝试标头。