devbox
weixin_40725706
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

分布式- Mysql + Mongo多数据源的分布式事务回滚_transactionsynchronizationmanager.getresource(mong

作者：weixin_40725706 | 2024-03-20 16:01:41

赞

踩

transactionsynchronizationmanager.getresource(mongotransactionmanager.getdbf

1. XA两阶段提交模型

如上图，XA规范实现的两阶段提交流程：（下面全部翻译自XA规范原文）

阶段1：

　　TM要求所有RMs准备提交(或准备)事务分支。这询问RM是否能够保证提交事务分支的能力。RM可能会查询该RM内部的其他实例。CRM被要求准备它们创建的事务分支,将prepare请求发送到远程站点并接收结果。在返回失败并回滚其工作之后，RM可以丢弃事务分支的信息。

阶段2：

　　TM根据实际情况向所有RMs发出提交或回滚事务分支的请求。CRM被要求提交或回滚它们创建的事务分支,向远程站点发送提交或回滚请求并接收结果。所有RMs提交或回滚对共享资源的更改，然后将状态返回给TM。然后TM可以丢弃全局事务的信息。

1.1 XA对2PC的优化

  • 只读断言

　　当事务分支没有更新共享资源时，这个RM会断言并响应给TM的prepare请求。也就免去了阶段2。但是，如果一个RM在全局事务的所有RMs返回prepared之前返回了只读优化，该RM释放事务上下文，例如read locks。这时候其他事务就有机会去改变这些数据（可能是写锁），显然全局序列化被破坏。同样CRM也可以断言，当TM挂起或终止线程与事务分支的关联时，它不是某个特定线程中活动的事务分支的参与者。

  • 一阶段提交

　　如果一个TM知道DTP系统中只有一个RM在修改共享资源，那么它可以使用单阶段提交。即TM免去了阶段1的prepare，直接执行了阶段2的commit。

1.2 2PC的缺点

  • 资源阻塞

由于协调者的重要性，一旦协调者TM发生故障。参与者RM会一直阻塞下去。尤其在第二阶段，协调者发生故障，那么所有的参与者还都处于锁定事务资源的状态中，而无法继续完成事务操作。（如果是协调者挂掉，可以重新选举一个协调者，但是无法解决因为协调者宕机导致的参与者处于阻塞状态的问题）

  • 数据不一致

在阶段二，当协调者向参与者发送commit请求之后，发生了局部网络异常或者在发送commit请求过程中协调者发生了故障，导致只有一部分参与者接受到了commit请求。而在这部分参与者接到commit请求之后就会执行commit操作。但是其他部分未接到commit请求的机器则无法执行事务提交。于是整个分布式系统便出现了数据不一致性的现象。

由于二阶段提交存在着这些缺陷，所以，研究者们在二阶段提交的基础上做了改进，提出了三阶段提交(该篇文章暂不涉及)。

2. Spring集成jta-atomikos实现Mysql + Mongo多数据源事务回滚

（1）项目环境为SpringBoot2.1.6

（2）依赖版本管理

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>std-boot-starter-transaction</artifactId>
  7.         <groupId>priv.whh.std</groupId>
  8.         <version>1.0.0-SNAPSHOT</version>
  9.         <relativePath>../pom.xml</relativePath>
  10.     </parent>
  11.     <modelVersion>4.0.0</modelVersion>
  12.  
  13.     <artifactId>std-boot-jta</artifactId>
  14.     <properties>
  15.         <druid.version>1.1.23</druid.version>
  16.         <mybatis-spring-boot-starter.version>1.3.2</mybatis-spring-boot-starter.version>
  17.         <mybatis.version>3.4.6</mybatis.version>
  18.         <mybatis-spring.version>1.3.2</mybatis-spring.version>
  19.         <mysql-connector-java.version>8.0.11</mysql-connector-java.version>
  20.     </properties>
  21.  
  22.     <dependencies>
  23.         <dependency>
  24.             <groupId>org.springframework.boot</groupId>
  25.             <artifactId>spring-boot-starter-data-mongodb</artifactId>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.springframework.boot</groupId>
  29.             <artifactId>spring-boot-starter-jta-atomikos</artifactId>
  30.         </dependency>
  31.         <dependency>
  32.             <groupId>com.alibaba</groupId>
  33.             <artifactId>druid</artifactId>
  34.         </dependency>
  35.         <dependency>
  36.             <groupId>org.projectlombok</groupId>
  37.             <artifactId>lombok</artifactId>
  38.         </dependency>
  39.         <dependency>
  40.             <groupId>org.mybatis.spring.boot</groupId>
  41.             <artifactId>mybatis-spring-boot-starter</artifactId>
  42.             <scope>test</scope>
  43.         </dependency>
  44.         <dependency>
  45.             <groupId>org.mybatis</groupId>
  46.             <artifactId>mybatis-spring</artifactId>
  47.             <scope>test</scope>
  48.         </dependency>
  49.         <dependency>
  50.             <groupId>org.mybatis</groupId>
  51.             <artifactId>mybatis</artifactId>
  52.             <scope>test</scope>
  53.         </dependency>
  54.         <dependency>
  55.             <groupId>org.springframework.boot</groupId>
  56.             <artifactId>spring-boot-starter-web</artifactId>
  57.             <scope>test</scope>
  58.         </dependency>
  59.         <dependency>
  60.             <groupId>mysql</groupId>
  61.             <artifactId>mysql-connector-java</artifactId>
  62.             <scope>test</scope>
  63.         </dependency>
  64.         <dependency>
  65.             <groupId>org.springframework</groupId>
  66.             <artifactId>spring-jdbc</artifactId>
  67.             <scope>test</scope>
  68.         </dependency>
  69.     </dependencies>
  70.  
  71.     <dependencyManagement>
  72.         <dependencies>
  73.             <dependency>
  74.                 <groupId>com.alibaba</groupId>
  75.                 <artifactId>druid</artifactId>
  76.                 <version>${druid.version}</version>
  77.             </dependency>
  78.             <dependency>
  79.                 <groupId>org.mybatis.spring.boot</groupId>
  80.                 <artifactId>mybatis-spring-boot-starter</artifactId>
  81.                 <version>${mybatis-spring-boot-starter.version}</version>
  82.             </dependency>
  83.             <dependency>
  84.                 <groupId>org.mybatis</groupId>
  85.                 <artifactId>mybatis-spring</artifactId>
  86.                 <version>${mybatis-spring.version}</version>
  87.             </dependency>
  88.             <dependency>
  89.                 <groupId>org.mybatis</groupId>
  90.                 <artifactId>mybatis</artifactId>
  91.                 <version>${mybatis.version}</version>
  92.             </dependency>
  93.             <dependency>
  94.                 <groupId>mysql</groupId>
  95.                 <artifactId>mysql-connector-java</artifactId>
  96.                 <version>${mysql-connector-java.version}</version>
  97.             </dependency>
  98.         </dependencies>
  99.     </dependencyManagement>
  100. </project>

（3）配置类JtaAutoConfiguration

  1. package priv.whh.std.boot.jta.config;
  2.  
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.springframework.context.annotation.Primary;
  6. import org.springframework.data.mongodb.MongoDbFactory;
  7. import org.springframework.data.mongodb.MongoTransactionManager;
  8. import priv.whh.std.boot.jta.manager.JtaUserTransaction;
  9. import priv.whh.std.boot.jta.util.MongoUtils;
  10.  
  11. /**
  12.  * @author whh
  13.  * @date 2020/7/22
  14.  */
  15. @Configuration
  16. public class JtaAutoConfiguration {
  17.     @Bean
  18.     @Primary
  19.     public JtaUserTransaction jtaUserTransaction(MongoDbFactory factory, MongoUtils mongoUtils) {
  20.         return new JtaUserTransaction(new MongoTransactionManager(factory), mongoUtils);
  21.     }
  22. }

（4）自定义事务JtaUserTransaction

  1. package priv.whh.std.boot.jta.manager;
  2.  
  3. import com.atomikos.icatch.config.UserTransactionService;
  4. import com.atomikos.icatch.config.UserTransactionServiceImp;
  5. import com.atomikos.icatch.jta.TransactionManagerImp;
  6. import com.atomikos.util.SerializableObjectFactory;
  7. import com.mongodb.MongoException;
  8. import com.mongodb.TransactionOptions;
  9. import com.mongodb.client.ClientSession;
  10. import lombok.extern.slf4j.Slf4j;
  11. import org.springframework.data.mongodb.MongoDatabaseUtils;
  12. import org.springframework.data.mongodb.MongoDbFactory;
  13. import org.springframework.data.mongodb.MongoTransactionManager;
  14. import org.springframework.data.mongodb.SessionSynchronization;
  15. import org.springframework.lang.Nullable;
  16. import org.springframework.transaction.TransactionDefinition;
  17. import org.springframework.transaction.TransactionSystemException;
  18. import org.springframework.transaction.support.*;
  19. import org.springframework.util.Assert;
  20. import org.springframework.util.ClassUtils;
  21. import priv.whh.std.boot.jta.util.MongoUtils;
  22.  
  23. import javax.naming.NamingException;
  24. import javax.naming.Reference;
  25. import javax.naming.Referenceable;
  26. import javax.transaction.NotSupportedException;
  27. import javax.transaction.SystemException;
  28. import javax.transaction.TransactionManager;
  29. import javax.transaction.UserTransaction;
  30. import java.io.Serializable;
  31. import java.lang.reflect.Method;
  32. import java.util.Objects;
  33.  
  34. /**
  35.  * @author whh
  36.  * @date 2020/7/23
  37.  */
  38. @Slf4j
  39. public class JtaUserTransaction implements UserTransaction, Serializable, Referenceable {
  40.     private static final long serialVersionUID = -865418426269785202L;
  41.     private transient TransactionManager transactionManager;
  42.     private transient MongoTransactionManager mongoTransactionManager;
  43.     private final transient MongoUtils mongoUtils;
  44.  
  45.     public JtaUserTransaction(MongoTransactionManager mongoTransactionManager, MongoUtils mongoUtils) {
  46.         this.mongoTransactionManager = mongoTransactionManager;
  47.         this.mongoUtils = mongoUtils;
  48.     }
  49.  
  50.     /**
  51.      * @see javax.transaction.UserTransaction
  52.      */
  53.     @Override
  54.     public void begin() throws NotSupportedException, SystemException {
  55.         checkSetup();
  56.         transactionManager.begin();
  57.         mongoUtils.setSessionSynchronizationForTransactionBegin();
  58.     }
  59.  
  60.     /**
  61.      * @see javax.transaction.UserTransaction
  62.      */
  63.     @Override
  64.     public void commit() throws javax.transaction.RollbackException, javax.transaction.HeuristicMixedException,
  65.         javax.transaction.HeuristicRollbackException, javax.transaction.SystemException {
  66.         Assert.notNull(mongoTransactionManager.getDbFactory(), "DbFactory must not be null!");
  67.         if (Objects.nonNull(TransactionSynchronizationManager.getResource(mongoTransactionManager.getDbFactory()))) {
  68.             MongoTransactionObject mongoTransactionObject = extractMongoTransaction(getMongoTransaction());
  69.             MongoResourceHolder resourceHolder = newResourceHolder(new DefaultTransactionDefinition());
  70.             mongoTransactionObject.setResourceHolder(resourceHolder);
  71.             try {
  72.                 mongoTransactionObject.commitTransaction();
  73.                 TransactionSynchronizationManager.unbindResource(mongoTransactionManager.getDbFactory());
  74.                 mongoTransactionObject.getRequiredResourceHolder().clear();
  75.                 mongoTransactionObject.closeSession();
  76.                 mongoUtils.setSessionSynchronizationForTransactionCompletion();
  77.             } catch (Exception ex) {
  78.                 throw new TransactionSystemException(String.format("Could not commit Mongo transaction for session %s.",
  79.                     debugString(mongoTransactionObject.getSession())), ex);
  80.             }
  81.         }
  82.         checkSetup();
  83.         transactionManager.commit();
  84.     }
  85.  
  86.     /**
  87.      * @see javax.transaction.UserTransaction
  88.      */
  89.     @Override
  90.     public void rollback() throws SystemException {
  91.         Assert.notNull(mongoTransactionManager.getDbFactory(), "Db factory must not be null");
  92.         if (Objects.nonNull(TransactionSynchronizationManager.getResource(mongoTransactionManager.getDbFactory()))) {
  93.             MongoTransactionObject mongoTransactionObject = extractMongoTransaction(getMongoTransaction());
  94.             MongoResourceHolder resourceHolder = newResourceHolder(new DefaultTransactionDefinition());
  95.             mongoTransactionObject.setResourceHolder(resourceHolder);
  96.             try {
  97.                 mongoTransactionObject.abortTransaction();
  98.                 TransactionSynchronizationManager.unbindResource(mongoTransactionManager.getDbFactory());
  99.                 mongoTransactionObject.getRequiredResourceHolder().clear();
  100.                 mongoTransactionObject.closeSession();
  101.                 mongoUtils.setSessionSynchronizationForTransactionCompletion();
  102.             } catch (MongoException ex) {
  103.                 throw new TransactionSystemException(String.format("Could not abort Mongo transaction for session %s.",
  104.                     debugString(mongoTransactionObject.getSession())), ex);
  105.             }
  106.         }
  107.         checkSetup();
  108.         transactionManager.rollback();
  109.     }
  110.  
  111.     /**
  112.      * 不抛出异常进行回滚
  113.      *
  114.      * @see javax.transaction.UserTransaction
  115.      */
  116.     @Override
  117.     public void setRollbackOnly() throws SystemException {
  118.         checkSetup();
  119.         transactionManager.setRollbackOnly();
  120.     }
  121.  
  122.     /**
  123.      * @see javax.transaction.UserTransaction
  124.      */
  125.     @Override
  126.     public int getStatus() throws SystemException {
  127.         checkSetup();
  128.         return transactionManager.getStatus();
  129.     }
  130.  
  131.     /**
  132.      * @see javax.transaction.UserTransaction
  133.      */
  134.     @Override
  135.     public void setTransactionTimeout(int seconds) throws SystemException {
  136.         checkSetup();
  137.         transactionManager.setTransactionTimeout(seconds);
  138.     }
  139.  
  140.     /**
  141.      * IMPLEMENTATION OF REFERENCEABLE
  142.      */
  143.     @Override
  144.     public Reference getReference() throws NamingException {
  145.         return SerializableObjectFactory.createReference(this);
  146.     }
  147.  
  148.     protected int determineTimeout(TransactionDefinition definition) {
  149.         if (definition.getTimeout() != TransactionDefinition.TIMEOUT_DEFAULT) {
  150.             return definition.getTimeout();
  151.         }
  152.         return TransactionDefinition.TIMEOUT_DEFAULT;
  153.     }
  154.  
  155.     /**
  156.      * Referenceable mechanism requires later setup of transactionManager, otherwise binding
  157.      * <p>
  158.      * into JNDI already requires that TM is running.
  159.      */
  160.     private void checkSetup() {
  161.         synchronized (TransactionManagerImp.class) {
  162.             transactionManager = TransactionManagerImp.getTransactionManager();
  163.             if (Objects.isNull(transactionManager)) {
  164.                 UserTransactionService uts = new UserTransactionServiceImp();
  165.                 uts.init();
  166.                 transactionManager = TransactionManagerImp.getTransactionManager();
  167.             }
  168.         }
  169.     }
  170.  
  171.     private Object getMongoTransaction() {
  172.         MongoDbFactory mongoDbFactory = mongoTransactionManager.getDbFactory();
  173.         Assert.notNull(mongoDbFactory, "Db factory must not be null");
  174.         MongoResourceHolder resourceHolder = (MongoResourceHolder) TransactionSynchronizationManager
  175.             .getResource(mongoDbFactory);
  176.         return new MongoTransactionObject(resourceHolder);
  177.     }
  178.  
  179.     private static MongoTransactionObject extractMongoTransaction(Object transaction) {
  180.         Assert.isInstanceOf(MongoTransactionObject.class, transaction,
  181.             () -> String.format("Expected to find a %s but it turned out to be %s.", MongoTransactionObject.class,
  182.                 transaction.getClass()));
  183.         return (MongoTransactionObject) transaction;
  184.     }
  185.  
  186.     private MongoResourceHolder newResourceHolder(TransactionDefinition definition) {
  187.         MongoDbFactory dbFactory = mongoTransactionManager.getDbFactory();
  188.         Class<MongoDatabaseUtils> mongoDatabaseUtilsClazz = MongoDatabaseUtils.class;
  189.         ClientSession session = null;
  190.         try {
  191.             Method doGetSession = mongoDatabaseUtilsClazz.getDeclaredMethod(
  192.                 "doGetSession", MongoDbFactory.class, SessionSynchronization.class);
  193.             doGetSession.setAccessible(true);
  194.             session = (ClientSession) doGetSession.invoke(
  195.                 mongoDatabaseUtilsClazz.newInstance(), dbFactory, SessionSynchronization.ALWAYS);
  196.         } catch (Exception e) {
  197.             log.error("getSession err;", e.getCause());
  198.         }
  199.         if (Objects.nonNull(session)) {
  200.             MongoResourceHolder resourceHolder = new MongoResourceHolder(session, dbFactory);
  201.             resourceHolder.setTimeoutIfNotDefaulted(determineTimeout(definition));
  202.             return resourceHolder;
  203.         }
  204.         return null;
  205.     }
  206.  
  207.     private static String debugString(@Nullable ClientSession session) {
  208.         if (session == null) {
  209.             return "null";
  210.         }
  211.         String debugString = String.format("[%s@%s ", ClassUtils.getShortName(session.getClass()),
  212.             Integer.toHexString(session.hashCode()));
  213.         try {
  214.             if (session.getServerSession() != null) {
  215.                 debugString += String.format("id = %s, ", session.getServerSession().getIdentifier());
  216.                 debugString += String.format("causallyConsistent = %s, ", session.isCausallyConsistent());
  217.                 debugString += String.format("txActive = %s, ", session.hasActiveTransaction());
  218.                 debugString += String.format("txNumber = %d, ", session.getServerSession().getTransactionNumber());
  219.                 debugString += String.format("closed = %s, ", session.getServerSession().isClosed());
  220.                 debugString += String.format("clusterTime = %s", session.getClusterTime());
  221.             } else {
  222.                 debugString += "id = n/a";
  223.                 debugString += String.format("causallyConsistent = %s, ", session.isCausallyConsistent());
  224.                 debugString += String.format("txActive = %s, ", session.hasActiveTransaction());
  225.                 debugString += String.format("clusterTime = %s", session.getClusterTime());
  226.             }
  227.         } catch (RuntimeException e) {
  228.             debugString += String.format("error = %s", e.getMessage());
  229.         }
  230.         debugString += "]";
  231.         return debugString;
  232.     }
  233.  
  234.     /**
  235.      * @see org.springframework.data.mongodb.MongoResourceHolder
  236.      */
  237.     protected static class MongoTransactionObject implements SmartTransactionObject {
  238.         @Nullable
  239.         private MongoResourceHolder resourceHolder;
  240.  
  241.         MongoTransactionObject(@Nullable MongoResourceHolder resourceHolder) {
  242.             this.resourceHolder = resourceHolder;
  243.         }
  244.  
  245.         /**
  246.          * Set the {@link MongoResourceHolder}.
  247.          *
  248.          * @param resourceHolder can be {@literal null}.
  249.          */
  250.         void setResourceHolder(@Nullable MongoResourceHolder resourceHolder) {
  251.             this.resourceHolder = resourceHolder;
  252.         }
  253.  
  254.         /**
  255.          * @return {@literal true} if a {@link MongoResourceHolder} is set.
  256.          */
  257.         final boolean hasResourceHolder() {
  258.             return resourceHolder != null;
  259.         }
  260.  
  261.         /**
  262.          * Start a MongoDB transaction optionally given {@link TransactionOptions}.
  263.          *
  264.          * @param options can be {@literal null}
  265.          */
  266.  
  267.         void startTransaction(@Nullable TransactionOptions options) {
  268.             ClientSession session = getRequiredSession();
  269.             if (options != null) {
  270.                 session.startTransaction(options);
  271.             } else {
  272.                 session.startTransaction();
  273.             }
  274.         }
  275.  
  276.         /**
  277.          * Commit the transaction.
  278.          */
  279.         public void commitTransaction() {
  280.             getRequiredSession().commitTransaction();
  281.         }
  282.  
  283.         /**
  284.          * Rollback (abort) the transaction.
  285.          */
  286.         public void abortTransaction() {
  287.             getRequiredSession().abortTransaction();
  288.         }
  289.  
  290.         /**
  291.          * Close a {@link ClientSession} without regard to its transactional state.
  292.          */
  293.         void closeSession() {
  294.             ClientSession session = getRequiredSession();
  295.             if (session.getServerSession() != null && !session.getServerSession().isClosed()) {
  296.                 session.close();
  297.             }
  298.         }
  299.  
  300.         @Nullable
  301.         public ClientSession getSession() {
  302.             return resourceHolder != null ? resourceHolder.getSession() : null;
  303.         }
  304.  
  305.         private MongoResourceHolder getRequiredResourceHolder() {
  306.             Assert.state(resourceHolder != null, "MongoResourceHolder is required but not present. o_O");
  307.             return resourceHolder;
  308.         }
  309.  
  310.         private ClientSession getRequiredSession() {
  311.             ClientSession session = getSession();
  312.             Assert.state(session != null, "A Session is required but it turned out to be null.");
  313.             return session;
  314.         }
  315.  
  316.         /**
  317.          * (non-Javadoc)
  318.          *
  319.          * @see org.springframework.transaction.support.SmartTransactionObject#isRollbackOnly()
  320.          */
  321.         @Override
  322.         public boolean isRollbackOnly() {
  323.             return this.resourceHolder != null && this.resourceHolder.isRollbackOnly();
  324.         }
  325.  
  326.         /**
  327.          * (non-Javadoc)
  328.          *
  329.          * @see org.springframework.transaction.support.SmartTransactionObject#flush()
  330.          */
  331.         @Override
  332.         public void flush() {
  333.             TransactionSynchronizationUtils.triggerFlush();
  334.         }
  335.  
  336.     }
  337.  
  338.     /**
  339.      * @see org.springframework.data.mongodb.MongoResourceHolder
  340.      */
  341.     class MongoResourceHolder extends ResourceHolderSupport {
  342.         private @Nullable
  343.         ClientSession session;
  344.         private MongoDbFactory dbFactory;
  345.  
  346.         /**
  347.          * Create a new {@link com.shero.sport.web.conf.JtaTransactionImp.MongoResourceHolder} for a given {@link ClientSession session}.
  348.          *
  349.          * @param session   the associated {@link ClientSession}. Can be {@literal null}.
  350.          * @param dbFactory the associated {@link MongoDbFactory}. must not be {@literal null}.
  351.          */
  352.         MongoResourceHolder(@Nullable ClientSession session, MongoDbFactory dbFactory) {
  353.             this.session = session;
  354.             this.dbFactory = dbFactory;
  355.         }
  356.  
  357.         /**
  358.          * @return the associated {@link ClientSession}. Can be {@literal null}.
  359.          */
  360.         @Nullable
  361.         ClientSession getSession() {
  362.             return session;
  363.         }
  364.  
  365.         /**
  366.          * @return the required associated {@link ClientSession}.
  367.          * @throws IllegalStateException if no {@link ClientSession} is associated with this {@link com.shero.sport.web.conf.JtaTransactionImp.MongoResourceHolder}.
  368.          * @since 2.1.3
  369.          */
  370.         ClientSession getRequiredSession() {
  371.             ClientSession session = getSession();
  372.             if (session == null) {
  373.                 throw new IllegalStateException("No session available!");
  374.             }
  375.             return session;
  376.         }
  377.  
  378.         /**
  379.          * @return the associated {@link MongoDbFactory}.
  380.          */
  381.         public MongoDbFactory getDbFactory() {
  382.             return dbFactory;
  383.         }
  384.  
  385.         /**
  386.          * Set the {@link ClientSession} to guard.
  387.          *
  388.          * @param session can be {@literal null}.
  389.          */
  390.         public void setSession(@Nullable ClientSession session) {
  391.             this.session = session;
  392.         }
  393.  
  394.         /**
  395.          * Only set the timeout if it does not match the {@link TransactionDefinition#TIMEOUT_DEFAULT default timeout}.
  396.          *
  397.          * @param seconds
  398.          */
  399.         void setTimeoutIfNotDefaulted(int seconds) {
  400.             if (seconds != TransactionDefinition.TIMEOUT_DEFAULT) {
  401.                 setTimeoutInSeconds(seconds);
  402.             }
  403.         }
  404.  
  405.         /**
  406.          * @return {@literal true} if session is not {@literal null}.
  407.          */
  408.         boolean hasSession() {
  409.             return session != null;
  410.         }
  411.  
  412.         /**
  413.          * @return {@literal true} if the session is active and has not been closed.
  414.          */
  415.         boolean hasActiveSession() {
  416.             if (!hasSession()) {
  417.                 return false;
  418.             }
  419.             return hasServerSession() && !getRequiredSession().getServerSession().isClosed();
  420.         }
  421.  
  422.         /**
  423.          * @return {@literal true} if the session has an active transaction.
  424.          * @see #hasActiveSession()
  425.          * @since 2.1.3
  426.          */
  427.         boolean hasActiveTransaction() {
  428.             if (!hasActiveSession()) {
  429.                 return false;
  430.             }
  431.             return getRequiredSession().hasActiveTransaction();
  432.         }
  433.  
  434.         /**
  435.          * @return {@literal true} if the {@link ClientSession} has a {@link com.mongodb.session.ServerSession} associated
  436.          * <p>
  437.          * that is accessible via {@link ClientSession#getServerSession()}.
  438.          */
  439.         boolean hasServerSession() {
  440.             try {
  441.                 return getRequiredSession().getServerSession() != null;
  442.             } catch (IllegalStateException serverSessionClosed) {
  443.                 // ignore
  444.             }
  445.             return false;
  446.         }
  447.     }
  448. }

（5）工具类MongoUtils

  1. package priv.whh.std.boot.jta.util;
  2.  
  3. import lombok.RequiredArgsConstructor;
  4. import org.springframework.data.mongodb.SessionSynchronization;
  5. import org.springframework.data.mongodb.core.MongoTemplate;
  6. import org.springframework.stereotype.Component;
  7.  
  8. /**
  9.  * @author whh
  10.  * @date 2020/7/22
  11.  */
  12. @Component
  13. @RequiredArgsConstructor
  14. public class MongoUtils {
  15.     private final MongoTemplate mongoTemplate;
  16.  
  17.     public void setSessionSynchronizationForTransactionBegin() {
  18.         // 同步任何事务（即使是空事务）并在执行此操作时启动MongoDB事务
  19.         mongoTemplate.setSessionSynchronization(SessionSynchronization.ALWAYS);
  20.     }
  21.  
  22.     public void setSessionSynchronizationForTransactionCompletion() {
  23.         mongoTemplate.setSessionSynchronization(SessionSynchronization.ON_ACTUAL_TRANSACTION);
  24.     }
  25. }

（6）配置

  1. mybatis:
  2.   configuration:
  3.     map-underscore-to-camel-case: true
  4.   mapper-locations: classpath*:mapper/*.xml
  5.   type-aliases-package: com.tzc.whh.model
  6. spring:
  7.   datasource:
  8.     url: jdbc:mysql://127.0.0.1:3306/local?useUnicode=true&characterEncoding=UTF8&useSSL=false&serverTimezone=Asia/Shanghai
  9.     username: root
  10.     password: 123456
  11.     #    driver-class-name: com.mysql.jdbc.Driver
  12.     driver-class-name: com.mysql.cj.jdbc.Driver
  13.     xa:
  14.       data-source-class-name: com.alibaba.druid.pool.xa.DruidXADataSource
  15.   data:
  16.     mongodb:
  17.       uri: mongodb://127.0.0.1:27018,127.0.0.1:27019/coupon?authSource=coupon&slaveOk=true&replicaSet=rs0&write=1&readPreference=secondaryPreferred&connectTimeoutMS=300000
  18. server:
  19.   port: 9090

（7）事务类Manager

  1. package priv.whh.std.boot.jta.manager;
  2.  
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.data.mongodb.core.MongoTemplate;
  5. import org.springframework.stereotype.Component;
  6. import org.springframework.transaction.annotation.Transactional;
  7. import priv.whh.std.boot.jta.dao.UserMapper;
  8. import priv.whh.std.boot.jta.po.UserPo;
  9.  
  10. import java.util.Objects;
  11. import java.util.Random;
  12.  
  13. /**
  14.  * @author whh
  15.  * @date 2020/7/22
  16.  */
  17. @Component
  18. public class Manager {
  19.     @Autowired
  20.     private UserMapper userMapper;
  21.     @Autowired
  22.     private MongoTemplate mongoTemplate;
  23.  
  24.     @Transactional(rollbackFor = Exception.class)
  25.     public void test(Integer test) throws Exception {
  26.         userMapper.insert(new UserPo(1L, "test"));
  27.         mongoTemplate.save(new UserPo(new Random().nextLong(), "testA"), "t_account");
  28.         userMapper.insert(new UserPo(1L, "testB"));
  29.         mongoTemplate.save(new UserPo(new Random().nextLong(), "testB"), "t_account");
  30.         if (Objects.equals(1, test)) {
  31.             throw new Exception();
  32.         }
  33.     }
  34. }

当入参test为1时，抛出异常， 事务回滚。其它情况时，正常执行。

 

参考资料：

【分布式事务（一）原理概览】

【springboot + jta + mysql + mongo 分布式（多种数据源）事务】

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/274871
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号