带rabbitmq绑定器的Spring Cloud流,如何应用@transactional?

我有一个 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 使用SpringCloudstream(或者至少是我的印象)是因为如果在写入MongoDB时出现异常,那么事件似乎已经被确认:ed到rabbitmq,并且操作不会被重试。

假设我希望实现与使用 @事务性 围绕一个函数 spring-amqp :

1. 使用Spring时,是否必须手动将消息打包到rabbitmq 云流与兔子的活页夹?
2. 如果是这样,我如何才能做到这一点?