问题的可能预测
我认为这里的问题与save
和saveAndFlush
无关。问题似乎与Spring @Transactional
方法的性质有关,并且在涉及数据库和AMQP代理商的分布式环境中错误地使用了这些事务,或许还加入了一些对JPA上下文如何工作的基本误解。
在你的解释中,你似乎暗示你在@Transactional
方法内开始了JPA事务,并在事务期间(但在提交之前),向AMQP代理商发送消息。稍后,在队列的另一侧,消费者应用程序获取消息并进行REST服务调用。此时,您注意到发布方的事务更改尚未提交到数据库,因此对于消费方而言不可见。
问题似乎是您在JPA事务提交到磁盘之前传播了那些AMQP消息。当消费者读取并处理消息时,您发布方的事务可能尚未完成。因此,这些更改对于消费者应用程序不可见。
如果你的AMPQ实现是Rabbit,那么我之前见过这个问题。当你启动一个使用数据库事务管理器的
@Transactional
方法,并在该方法中使用
RabbitTemplate
发送相应消息时,如果你的
RabbitTemplate
没有使用事务通道(即
channelTransacted=true
),则你的消息会在数据库事务提交之前被传送。我认为通过在
RabbitTemplate
中启用事务通道(默认情况下禁用),你可以解决部分问题。
<rabbit:template id="rabbitTemplate"
connection-factory="connectionFactory"
channel-transacted="true"/>
当通道被交易时,
RabbitTemplate
会“加入”当前数据库事务(显然是JPA事务)。一旦你的JPA事务提交,它会运行一些结尾代码,也会提交Rabbit通道中的更改,这将强制实际地“发送”消息。
关于save和saveAndFlush
你可能认为刷新JPA上下文中的更改应该解决问题,但你是错误的。刷新JPA上下文只是强制在内存中更改实体并写入磁盘。然而,它们仍然在相应的数据库事务中写入磁盘,直到你的JPA事务提交才会提交。这发生在你的
@Transactional
方法结束时(不幸的是,在你已经发送AMQP消息之后的某个时间点——如果你不使用如上所述的交易通道)。
因此,即使你刷新了JPA上下文,你的消费者应用程序也不会看到这些更改(根据经典的数据库隔离级别规则),直到你的发布者应用程序中的
@Transactional
方法完成。
当您调用
save(entity)
时,
EntityManager
不需要立即同步任何更改。大多数JPA实现将实体标记为脏数据并等待在最后一刻将所有更改与数据库同步,并在数据库级别提交这些更改。
注意:有些情况下,您可能希望某些更改立即写入磁盘,而不是等待主观的
EntityManager
决定何时进行同步。经典案例是,在数据库表中存在一个触发器,您需要运行它以生成稍后在事务期间需要使用的其他记录。因此,您强制将更改刷新到磁盘,以便强制触发器运行。
通过刷新上下文,您只是将内存中的更改强制同步到磁盘,但这并不意味着立即对这些修改进行数据库提交。因此,您刷新的更改不一定会对其他事务可见。根据传统的数据库隔离级别,很可能不会。
二阶段提交问题
这里还有另一个经典问题,即您的数据库和AMQP代理是两个独立的系统。如果涉及Rabbit,则没有2PC(两阶段提交)。
您可能需要考虑一些有趣的情况,例如,您的数据库事务成功提交。然而,此时Rabbit未能提交您的消息,在这种情况下,您将不得不重复整个事务,可能跳过数据库副作用并尝试重新发送消息到Rabbit。
您应该阅读这篇关于
Spring中的分布式事务,带或不带XA的文章,特别是有关链式事务的部分,有助于解决此问题。
他们建议使用更复杂的事务管理器定义。例如:
<bean id="jdbcTransactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
<bean id="rabbitTransactionManager" class="org.springframework.amqp.rabbit.transaction.RabbitTransactionManager">
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
<bean id="chainedTransactionManager" class="org.springframework.data.transaction.ChainedTransactionManager">
<constructor-arg name="transactionManagers">
<array>
<ref bean="rabbitTransactionManager"/>
<ref bean="jdbcTransactionManager"/>
</array>
</constructor-arg>
</bean>
在你的代码中,你只需要使用这个链接的事务管理器来协调你的数据库事务部分和Rabbit事务部分。现在,仍然有可能你提交了数据库部分,但是你的Rabbit事务部分失败了。所以,想象一下这样的情况:
@Retry
@Transactional("chainedTransactionManager")
public void myServiceOperation() {
if(workNotDone()) {
doDatabaseTransactionWork();
}
sendMessagesToRabbit();
}
如果你的Rabbit事务部分因任何原因失败,并且你被迫重试整个链式事务,那么你将避免重复数据库副作用,只需确保将失败的消息发送到Rabbit。
同时,如果你的数据库部分失败,那么你就没有发送消息到Rabbit,也不会有问题。
或者,如果你的数据库副作用是幂等的,那么你可以跳过检查,只需重新应用数据库更改,并尝试重新发送消息到Rabbit。
事实上,最初你所尝试的似乎很容易,但一旦你深入了解不同的问题,你就会意识到这是一个棘手的业务,需要正确的方式来处理。