Spring Cloud Stream: как повторно опубликовать в очередь недоставленных сообщений, а также выдать исключение
Я переношу проект, использующий Spring AMQP, в проект, использующий Spring Cloud Stream с RabbitMQ.
В моем старом проекте, когда возникало какое-то исключение при обработке сообщения с помощью @RabbitListener, выбрасывалось это исключение. Если была привязана очередь недоставленных сообщений, исключение все равно выбрасывалось (только один раз, если были повторные попытки, я думаю, последняя). Это было очень полезно для ведения журнала.
В Spring Cloud существует механизм очереди недоставленных сообщений для @StreamListener, если вы определяете свойства:
spring.cloud.stream.bindings.input1.destination=dest1
spring.cloud.stream.rabbit.bindings.input1.consumer.auto-bind-dlq=true
spring.cloud.stream.rabbit.bindings.input1.consumer.republishToDlq=true
Но если у вас есть такой метод (это просто пример):
@StreamListener("input1")
public void process(String message){
System.out.println("Trying...");
throw new RuntimeException();
}
Журналы:
Trying...
Trying...
Trying...
(end of log, no exception thrown)
Есть ли способ сгенерировать исключение (только при последней попытке)?
Спасибо!
Ответы 3
См. документацию о потребительских свойствах.
Установите ...consumer.max-attempts=1, чтобы отключить повторную попытку.
Что ж, у вас есть spring.cloud.stream.rabbit.bindings.input1.consumer.republishToDlq=true, что означает, что связыватель повторно публикует сообщение в DLQ (и добавляет информацию об исключении) вместо создания исключения, которое заставляет брокера отправлять его в DLQ, но без информации об исключении. Если это то, что вы хотите, удалите republishToDlq, чтобы он по умолчанию был false.
Но если я установлю для republishToDql значение false, сообщение (без исключения) не будет повторно опубликовано в очереди недоставленных сообщений. В документации (docs.spring.io/spring-cloud-stream/docs/current/reference/…) кажется достаточным использовать auto-bind-dlq=true, но если я удалю это свойство, сообщение никогда не поступит в очередь недоставленных сообщений.
Есть два варианта: генерировать исключение, и брокер отправляет исходное сообщение в DLQ или republishToDlq=true, что означает, что вместо этого связующее публикует сообщение в DLQ с дополнительной диагностической информацией (исключение и т. д. в заголовках). auto-bind-dlq просто устанавливает DLQ и настраивает исходную очередь для отправки туда отклонений (при возникновении исключения). Очередь не должна существовать без конфигурации DLQ; вы не можете изменить очередь после ее создания.
В этом была проблема: "Очередь не должна уже существовать без конфигурации DLQ; вы не можете изменить очередь после ее создания". Спасибо!!
Вы можете обработать исключение, зарегистрировать его, а затем бросить исключение AmqpRejectAndDontRequeueException. Это отправит сообщение в очередь недоставленных сообщений.
Вы находитесь под @StreamListener, где, по вашему мнению, должно быть исключение? кто ловит?
вы можете сделать это примерно так:
@StreamListener("input1")
public void process(String message){
try {
System.out.println("Trying...");
throw new RuntimeException();
// or the actual code that handle the message
} catch (RuntimeException re) {
// handle the exception, logging etc.
throw re
}
}
Я хочу: 1) иметь повторные попытки 2) письмо мертвой очереди, когда повторные попытки исчерпаны 3) выдать последнее исключение, у меня есть все 3 вещи при использовании Spring AMQP.