处理消息时重试次数有限的Kafka使用者

Question

我很难在Kafka主题的消费者中找到处理异常的简单模式。场景如下：在消费者中，我调用一个外部服务。如果服务不可用，我想重试几次，然后停止消费。

最简单的模式似乎是一种处理它的阻塞同步方式，在Java中如下所示：

ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
    boolean processed=false;
    int count=0;
    while (!processed) {
        try {
            callService(..); 
        } catch (Exception e) {
            if (count++ < 3) {
                Thread.sleep(5000);
                continue;
            } else throw new RuntimeException();
        }
    }
}

但是，我觉得必须有一种更简单的方法（不使用第三方库），并且避免阻塞线程。

这似乎是我们想要的一种常见的东西，但我找不到一个简单的例子来说明这种模式。

Answer 1

Kafaka提供的这种开箱即用的再审机制并不存在。使用RabbitMQ的经验，其中MQ提供重试交换。这些交换在RabbitMQ中称为死信交换。

https://www.rabbitmq.com/dlx.html

你可以在Kafaka的例子中应用同样的模式。

void consumeMainTopicWithPostponedRetry() {
    while (true) {
        Message message = takeNextMessage("main_topic");
        try {
            process(message);
        } catch (Exception ex) {
            publishTo("retry_topic");
            LOGGER.warn("Message processing failure. Will try once again in the future.", ex);
        }
    }
}

void consumeRetryTopic() {
    while (true) {
        Message message = takeNextMessage("retry_topic");
        try {
            process(message);
            waitSomeLongerTime();
        } catch (Exception ex) {
            publishTo("failed_topic");
            LOGGER.warn("Message processing failure. Will skip it.", ex);
        }
    }
}

以上的策略和例子摘自下面的链接。所有的功劳都要归功于博文的主人。https://blog.pragmatists.com/retring-consumer-architecture-in-the-apache-kafka-939AC4CB851A

对于以上所做的无阻塞方式，可以通过阅读整篇博文来理解。希望这有帮助。