#rabbitmq #spring-cloud-stream
#rabbitmq #spring-cloud-stream
Вопрос:
У меня есть вариант использования, когда сообщения должны обрабатываться с определенной задержкой, а также я хочу ввести экспоненциальный откат между повторными попытками, приложение reroutedlq, упомянутое в документации, точно соответствует моей цели.
https://cloud.spring.io/spring-cloud-static/Finchley.SR1/multi/multi__rabbitmq_binder.html
Единственная дополнительная информация, которая мне нужна, это то, можно ли также настроить delayedExchange через application.properties и привязать то же самое к исходным очередям вместо того, чтобы делать то же самое через Beans, чтобы не было необходимости дублировать логику в нескольких обработчиках ошибок (т. Е. один обработчик ошибок на один DLQ)
Пожалуйста, дайте мне знать, если потребуется какая-либо дополнительная информация.
Редактировать:
Согласно полученным комментариям, я удалил все обмены и очереди, а затем попробовал заново локально, но все еще сталкиваюсь с приведенной ниже ошибкой Error: «получено «x-delayed-message», но текущим является «тема» «
Файл моих свойств:
spring.cloud.stream.bindings.packagingConsumerChannel.destination=fabric-exchange
spring.cloud.stream.bindings.packagingConsumerChannel.group=package-queue
spring.cloud.stream.bindings.packagingConsumerChannel.consumer.max-attempts=1
spring.cloud.stream.rabbit.bindings.packagingConsumerChannel.consumer.durableSubscription=true
spring.cloud.stream.rabbit.bindings.packagingConsumerChannel.consumer.bindingRoutingKey=package
spring.cloud.stream.rabbit.bindings.packagingConsumerChannel.consumer.autoBindDlq=true
spring.cloud.stream.rabbit.bindings.packagingConsumerChannel.consumer.dlqDeadLetterExchange=
spring.cloud.stream.rabbit.bindings.packagingConsumerChannel.consumer.delayedExchange=true
Ответ №1:
Установите для свойства привязки потребителя delayedExchange значение true .
Фиксация была добавлена в версии 1.2.
Редактировать
@SpringBootApplication
@EnableBinding(Sink.class)
public class So54986375Application {
public static void main(String[] args) {
SpringApplication.run(So54986375Application.class, args);
}
@StreamListener(Sink.INPUT)
public void listen(String in) {
}
}
и
spring.cloud.stream.rabbit.bindings.input.consumer.delayedExchange=true
и
Комментарии:
1. Я пробовал это, но безуспешно, я всегда сталкиваюсь с этим исключением «получено «x-delayed-message», но текущее значение ‘direct’ «, и если я не задаю тип обмена как прямое, то получаю исключение «получено»x-delayed-message», но текущее значение ‘topic’ » , у меня установлен плагин отложенного сообщения, и я попробовал это после сброса всей настройки rabbitmq с помощью команды сброса
2. Сообщение об ошибке довольно четкое. Вы не можете изменить существующий тип exchange. Сначала вы должны удалить его. Связующее средство создает обмен темами по умолчанию. Если это прямое, то его создает что-то другое.
3. Я удалил все очереди и обмены, и именно тогда, когда я попытался, я также прокомментировал настройку exchangeType на direct, чтобы по умолчанию был создан обмен темами, но затем я также столкнулся с «полученным» x-delayed-сообщением » но текущим является «тема» » . Я обновлю вопрос с моими настройками, пожалуйста, просмотрите его один раз
4. Я не знаю, что еще я могу сказать, кроме как добавить пример; смотрите правку к моему ответу; это работает точно так, как я описал. У вас должен быть какой-то другой код, который объявляет exchange.
5. Ах да, у меня есть другое свойство на стороне производителя, которое объявляет тип exchange, я думаю, что это может быть причиной проблемы, отладим его, спасибо за поддержку. spring.cloud.stream.rabbit.bindings.packagingProducerChannel.producer.exchangeType=true