Spring/RabbitMQ: управление транзакциями

Question

Чтобы упростить мою проблему, я

App1 с @Transactionnal методом CreateUser():

• Вставьте новый пользователь в базе данных
• Добавить асинхронной сообщение в RabbitMQ так что пользователь получает письмо-уведомление
• (возможно, дополнительный код, но не очень)

App2 с RabbitMQ сообщение consummer

• Consummes сообщения на очереди рассылки в режиме реального времени
• Чтение почты данные в базе данных
• Отправить письмо

Проблема заключается в том, что иногда, App2 пытается использовать сообщение RabbitMQ до того, как транзакция будет совершена даже в App1. Это означает, что App2 не может читать почтовые данные в базе данных, так как пользователь еще не создан.

Некоторые решения могут быть:

• Используйте READ_UNCOMMITED уровень изоляции на App2
• Добавить некоторую задержку в доставке RabbitMQ сообщений (или некоторые RetryTemplate на consummer)
• изменить способ отправки сообщений электронной почты ...

Я видел, что весной есть RabbitTransactionManager, но я не могу понять, как это предполагается орк. Внутренние элементы обработки транзакций всегда кажутся немного трудными для понимания, и документация также не помогает.

---

Есть ли способ сделать что-то подобное?

• Добавить сообщение в очередь RabbitMQ в @Transactionnal метода
• При завершении транзакции, сообщение поручен в очередь, и изменения поручены к базе данных
• Так что сообщение не могут быть использованы до завершения транзакции db

Как? И что ожидать, например, если я отправляю синхронные сообщения RabbitMQ вместо асинхронных сообщений? Будет ли он блокировать поток, ожидающий ответа или что-то еще? Поскольку мы отправляем сообщения синхронизации и асинхронные сообщения для разных случаев использования.

Answer 1

Я не очень-то знаком с @Transactionnal и весной, но в AMQP стандартная очередь сообщений не является транзакционной операцией, поэтому вам нужно хранить данные в db (если соединение db транзактивно - транзакция фиксации) и только после этого отправьте сообщение для брокера.

Правильный рабочий процесс выглядит для меня как App1: createUser -> notifyUser; App2: listenForNotifications

Answer 2

Я знаю, что это поздно, но у меня был эта же проблема, из-за мое ограниченное понимание @Transactional в то время. Так что это больше для тех, кого вы случайно натыкаетесь на это.

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

Так что, если у вас есть метод, как

@Transactional(readOnly=false) 
public void save(Object object) { //Object should be one of your entities 
    entityManager.persist(object); //or however you have it set up 
    rabbitTemplate.convertAndSend(message); //again - however yours is 
} 

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

Возможно, что вложенные методы @Transactional (не прямолинейные) могут помещать сообщение в очередь после возврата метода save(). Однако вы не можете поместить сообщение в очередь и ожидать, что он не будет потреблен. Как только его там нет. Поэтому задержите его в очереди, если вам нужно.

Если вы хотите получить ответ от очереди синхронно. В моем примере с функцией - вы можете это сделать, однако это приведет к тому, что для фактического сохранения данных потребуется больше времени, так как он будет ожидать ответа от рабочего, прежде чем метод сможет вернуться и фактически сохранить данные. (также помните, что получение ответа от сообщения в очереди имеет тайм-аут).

Так что мой совет, чтобы не поставить эти 2 операции в том же @Transactional