Всем привет. В проекте два модуля общаются между собой через очередь (RabbitMq). Использую spring amqp для работы с RabbitMq.

Собственно вопрос, как осуществить проверку что сообщение ушло именно туда куда нужно и что его вторая сторона сможет прочитать ?


Это сообщение отредактировал(а) 4epT - 18.12.2012, 13:50

Привет.

Это можно узнать с помощью RabbitTemplate#returnCallback & RabbitTemplate#mandatory = true.
Что означает, что твои сообщения являются важными и ты получишь отбой, если брокер не смог отроутить твое сообщение.

Это уже не есть проблема твоего сендера. С твоей стороны важно доставить сообщение в конкретную очередь(очереди) по указанному routingKey.
Если ты сообщение доставил, то уже никак не сможешь понять сможет ли его листененр прочитать.
В этом и есть принцип сообщений: доставили и забыли.
сендер и ресивер абсолютно не должны знать о друг друге и они развязаны через посредника брокера: такой себе loosely-coupled.

С другой стороны у RabbitTemplate есть метод sendAndReceive, с помощью которого ты можешь организовать шаблон gateway, но тогда и твой ресивер, тоже должен обеспечить возврат ответа.
Но это все-равно тебя не спасает от ошибки "no route", что успешно можно поймать returnCallback'ом.

http://static.springsource.org/spring-amqp...l#amqp-template

Спасибо большое за ответ. Я что то не пойму где установить mandatory в true и установить returnCallback.

Использую RabbitTemplate (org.springframework.amqp.rabbit.core), сообщения отправляю вот так:



Так же в CachingConnectionFactory, нету property publisherReturns.

Можно какой нибудь пример ?)

Все ж зависит от того, какую версию Spring-AMQP ты используешь. Эта фича появилась в версии 1.1: https://jira.springsource.org/browse/AMQP-213

Да, точно. Использовал 1.0.

Тогда на солько я понял:

1) делаем

2) делаем

3) делаем


И если сообщение не получилось положить, то свалиться эксепшен ?

Все ли эти три пункта обязательны ?

Когда сработает returnedMessage и для что нужно писать в реализации ?

Это зависит от реализации


Да

См. исходники RabbitTemplate. И вообще рекомендую читать исходники библиотек, которые используете: много чего полезного можно для себя найти

Все остальное правильно.
Здесь подобный нашему дискашн: http://forum.springsource.org/showthread.php?127065

Спасибо большое за помощь. Как только разберусь, опишу в этой теме кратко что и как... может кому нибудь будет полезно в будущем.

Поразбирался.. посмотрел исходники, даже получилось кое что набросать, но есть пару не понятных вещей.

returnedMessage - словит сообщение, которое не удалось отроутить.

У меня есть 1 Exchange, 1 Routing key и 2 очереди. В одну очередь пишу (Queue_for_send), а другую слушаю (Queue_for_listen). Вот так делаю биндинг:



1) Слушателя биндить никуда не нужно ?

2) exchange это и есть брокер? Или я не так себе представляю это ?)

3) Конфигурация rabbitmq происходит в момент создания бина RabbitAdmin ?

Это сообщение отредактировал(а) 4epT - 19.12.2012, 18:51

Привет, сорри, что поздно возвращаюсь: болел.
Давай с самого начала:

Это допустимо, если твой Exchange is Topic: http://www.rabbitmq.com/tutorials/tutorial-five-java.html

Тут вообще не понял: если есть очередь, то кто-то в нее пишет, а кто-то из нее читает.
Опиши, плз, свою задачу: чего ты хочешь достигнуть с точки зрения архитектуры?
Да, и в очередь никто не пишет, по AMQP: есть Exchange и есть RoutingKey, по которому данный Exchange и решит, куда передавать сообщение.

Тепрь по твоим вопросам:

Нет, тебе достаточно сконфигурить на клиенте (слушателе) листенер-контейнер, ну и все, что он попросит.

Брокер, это и есть сервер AMQP, ну или если идти более широко, то это тот сервер, где размещаются очереди и хранилище сообщений. Это из теории.

Примерно так, но это происходит только один раз: если на брокере все уже сконфигурировано, то твой RabbitAdmin тихонько эти все пропустит.
Но если ты как-то радикально поменял настройки очередей или обменников(e.g. добавил аргумент в очередь 'x-dead-letter-exchange'), то может даже отругаться.

Да, мы обсуждаем вопросы Spring-AMQP здесь: http://forum.springsource.org/forumdisplay.php?74-AMQP
Там получишь рекомендацию из первых рук. Это я тебе говорю, как один из разработчиков Spring.
Да-а-а, совсем себя не прорекламировал...

Это сообщение отредактировал(а) MisterCleric - 21.12.2012, 21:03

Да, мой Exchange - Topic.



Делаю асинхронную обработку некоторых событий. То есть есть сервис для приема сообщений (заявок), на этой стороне сообщения обрабатываются и кладутся в очередь (Queue_for_send) на обработку. Другое приложение обрабатывает поступившие сообщения и кладет ответ в другую очередь (Queue_for_listen), которую я слушаю.



Понятно, спасибо   



Понятно, спасибо   



Тут вот в чем дело, допустим приложение нормально работает и в какой то момент времени падает очередь. Нужно ее переподнять (или переключить на другую), но нужно сконфигурировать настройки очередей и обменников, то есть связать exchange с очередью и роутинг кей. Но в таком случае RabbitAdmin уже проинициализирован, как быть в такой ситуации ?



Я бы с радостью, но боюсь что с моим англ. вы меня слабо поймете )

p.s. Выздоравливай!)))
 

Так я не понял: у тебя все рабюотает, как ожидалось или что-то еще осталось?

Счас исходников нет под рукой, но кажись ты это спокойно можешь делать используя RabbitAdmin бин напрямую.
Это с одной стороны.
С другой всегда все эти конфигурационные вещи можно делать прямо на сервере RabbitMQ или через удобный web-интерфейс: http://www.rabbitmq.com/management.html

Сейчас уже все работает, это я просто интересуюсь в тех моментах которые не совсем понятны.

Да, можно и напрямую через веб интерфейс, но там каждая минута на счету ...

И такой еще вопрос. Я могу как то сделать "гарантированную обработку" ? То есть, слушатель получил сообщение из очереди, начал его обработку, если все ок, то он помечает сообщение что оно отоработано, если упал эксепшен, то сообщение остается в очереди на обработке.

Сделай свой листенер транзакционным: http://static.springsource.org/spring-amqp...lsingle/#d4e415

  

Класс!!! Спасибо )

И последний вопрос не по теме.

Я могу как то спрингом обернуть некотороый метод в транзакцию ? в методе происходит цепочка событий (работа с mongodb, redis, rabbitmq и т..д), и в случае эксепшена на одном из событий, откат всех изменений (подчистка redis, подчистка БД и т.п.)    или такое делается только в ручную ?