分布式消息与数据库事务共享模式

Posted 2020-09-12 力奋

事务管理器org.springframework.jdbc.datasource.DataSourceTransactionManager(AbstractPlatformTransactionManager).getTransaction创建ConnectionHolder并用DriverManagerDataSource对象作键值保存在TransactionSynchronizationManager.resources．使TransactionSynchronizationManager.isSynchronizationActive=true表示可以执行TransactionSynchronizationManager.registerSynchronization保存事务对应的同步器到synchronizations如JmsResourceSynchronization。在事务使用JmsTemplate方法首先会创建JmsResourceHolder并用ActiveMQConnectionFactor对象作键值保存在TransactionSynchronizationManager.resources，同一个事务内共用session和connection。

ActiveMQSession.createConsumer()方法创建连接MQ服务器的消费者对象，MQ服务器会把消费者要操作的消息队列锁住，其它消费者对象不能操作这个消息队列，其它生产者对象可进行写操作，执行MessageConsumer.close()才释放锁。然后JmsTemplate.doReceive()方法是读取MQ服务器的消息，读取消息只是被MQ服务器隔离，执行session.commit()后MQ服务器才最终删除读取消息。不执行session.commit()消息可以回滚。代码片段下图，表示如果session由DataSourceTransactionManager管理可以忽略执行session.commit()，等到执行事务方法DataSourceTransactionManager.commit()再执行。同步器JmsResourceSynchronization已注册到TransactionSynchronizationManager，DataSourceTransactionManager.commit()会触发TransactionSynchronizationManager.synchronizations里所有同步器的afterCommit()方法，如JmsResourceSynchronization.afterCommit()，最终执行session.commit()。

            if (session.getTransacted()) {
                // Commit necessary - but avoid commit call within a JTA transaction.
                if (isSessionLocallyTransacted(session)) {
                    // Transacted session created by this template -> commit.
                    JmsUtils.commitIfNecessary(session);
                }
            }