Spring Cloud Stream с привязкой RabbitMQ, как применять @Transactional?

Question

У меня есть Spring Cloud Stream приложение, которое получает события от RabbitMQ с использованием Rabbit Binder .Мое приложение может быть кратко изложено следующим образом:

@Transactional
@StreamListener(MySink.SINK_NAME)
public void processEvents(Flux<Event> events) {
       // Transform events and store them in MongoDB using 
       // spring-boot-data-mongodb-reactive
       ...
}

Проблема в том, что не похоже, что @Transactional работает с Spring Cloud Stream (или, по крайней мере, это мое впечатление), так как если при записи есть исключениедля MongoDB событие, кажется, уже подтверждено: отправлено в RabbitMQ, и операция не повторяется.

Учитывая, что я хочу добиться в основном той же функциональности, что и при использовании @Transactional вокруг функции с spring-amqp :

1. Должен ли яВручную ACK сообщения в RabbitMQ при использовании Spring Cloud Stream с Rabbit Binder?
2. Если это так, как мне этого добиться?

Answer 1

Здесь есть несколько проблем.

1. Транзакции не требуются для подтверждения сообщений
2. Методы @StreamListener на основе реактора вызываются ровно один раз, просто для установки Flux поэтому @Transactional для этого метода не имеет смысла - сообщения затем проходят через поток, поэтому все, что касается отдельных сообщений, должно выполняться в контексте потока.
3. Spring Транзакции связаны с потоком - Reactorбез блокировки;сообщение будет подтверждено при первой передаче.

Да, вам нужно будет использовать ручные подтверждения;предположительно, в результате работы магазина mongodb.Вам, вероятно, потребуется использовать Flux<Message<Event>>, чтобы иметь доступ к каналу и заголовкам тегов доставки.