Spring 数据和 mongodb - 在 @Transactional 中使用 spring 进行简单回滚答案

【问题标题】：Spring data and mongodb - simple roll back with spring within @TransactionalSpring 数据和 mongodb - 在 @Transactional 中使用 spring 进行简单回滚
【发布时间】：2014-02-18 14:54:58
【问题描述】：

我有 2 个存储库，一个用于 mongodb (DocumentRepository)，另一个用于休眠实体 (EntityRepository)

我有一个简单的服务：

 @Transactional
 public doSomePersisting() {
     try {
           this.entityRepository.save(entity);
           this.documentRepository.save(document);
     }
     catch(...) {
         //Rollback mongoDB here
     }
 }

是否可以在“//Rollback mongoDB here”行上回滚mongoDB？我已经从实体部分得到回滚（事务注释）

【问题讨论】：

标签： java spring hibernate mongodb rollback

【解决方案1】：

MongoDB 不支持事务（至少不超出单个文档的范围）。如果您想回滚更改，则需要自己手工制作。如果您在某些情况下确实需要它们，有一些资源描述了在 Mongo 中实现您自己的事务的方法。你可以看看..

http://docs.mongodb.org/manual/tutorial/perform-two-phase-commits/

这只是对您可以使用的模式的解释。如果您发现您的应用程序中绝对需要事务，您应该考虑 MongoDB 是否适合您的需求。

【讨论】：

您可以将 mongo 操作留到最后（就像您在示例中所做的那样）。因此，当它失败时，它将回滚之前的弹簧。您通常需要 mongo 的原子操作，因此这将使您的两个数据源保持一致。除非您使用嵌套事务、分布式事务等，否则这些都不起作用，您必须通过后续的 mongo 更新来弥补。
迟到 ;-) 2018 年 2 月，MongoDB 4.0 发布。它确实支持 ACID 事务。从 Spring Data 2.1 (Lovelace) 开始，您可以将它与 @Transactional 注释一起使用。所以现在应该可以执行真正的两阶段提交了。您还可以看看更简单的解决方案-ChainedTransactionManager，其中可以将Mongodb TransactionManager和关系数据库TransactionManager结合在一起

【解决方案2】：

很抱歉重新发布我的答案。

早期的代码允许向 MongoDB 中插入数据，甚至在向 PostgreSQL 中插入数据时抛出查询异常（使用 myBatis）。

我已经解决了 MongoDB 和关系数据库之间的数据事务问题，@Transactional 通过在上述代码中进行这些更改完美地工作。

@Transactional 管理的解决方案。

Mongo 配置类

@Configuration
public class MongoConfig extends AbstractMongoConfiguration{
    private static final Logger LOG = LoggerFactory.getLogger(MongoConfig.class);

    @Value("${spring.data.mongodb.database}")
    private String dbName;

    @Value("${spring.data.mongodb.host}")
    private String dbHost;

    @Value("${spring.data.mongodb.port}")
    private int dbPort;

    @Override
    public String getDatabaseName() {
        return dbName;
    }

    @Bean
    public MongoClient mongoClient(){
        return new MongoClient(dbHost, dbPort);
    }

    @Bean
    public MongoDbFactory mongoDbFactory(){
        return new SimpleMongoDbFactory(mongoClient(),dbName);
    }

    @Bean
    public MongoTemplate mongoTemplate() {
        DbRefResolver dbRefResolver = new DefaultDbRefResolver(mongoDbFactory());
        MappingMongoConverter mappingMongoConverter = new MappingMongoConverter(dbRefResolver, new MongoMappingContext());
        // Don't save _class to mongo
        mappingMongoConverter.setTypeMapper(new DefaultMongoTypeMapper(null));
        MongoTemplate mongoTemplate = new MongoTemplate(mongoDbFactory(),mappingMongoConverter);
        mongoTemplate.setSessionSynchronization(SessionSynchronization.ON_ACTUAL_TRANSACTION);
        return mongoTemplate;
    }

    public MongoTemplate fetchMongoTemplate(int projectId) {
        DbRefResolver dbRefResolver = new DefaultDbRefResolver(mongoDbFactory());
        MappingMongoConverter mappingMongoConverter = new MappingMongoConverter(dbRefResolver, new MongoMappingContext());
        // Don't save _class to mongo
        mappingMongoConverter.setTypeMapper(new DefaultMongoTypeMapper(null));
        MongoDbFactory customizedDBFactory = new SimpleMongoDbFactory(mongoClient(), dbName+"_"+projectId);
        MongoTemplate mongoTemplate = new MongoTemplate(customizedDBFactory,mappingMongoConverter);
        MongoTransactionManager mongoTransactionManager = new MongoTransactionManager(customizedDBFactory);
        return mongoTemplate;
    }

    @Bean
    public MongoTransactionManager mongoTransactionManager() {
        return new MongoTransactionManager(mongoDbFactory());
    }

}

数据插入服务类

@Service
@Component
public class TestRepositoryImpl implements TestRepository{
    private static final Logger LOG = LoggerFactory.getLogger(TestRepositoryImpl.class);


@Autowired MongoConfig mongoConfig;
@Autowired MongoTemplate mongoTemplate;
@Autowired MongoTransactionManager mongoTransactionManager;

@Autowired UserService userService;

@Override
@Transactional
public void save(Test test){
    int projectId = 100;
    if (projectId != 0) {
        mongoTemplate = mongoConfig.fetchMongoTemplate(100);
        mongoTemplate.setSessionSynchronization(SessionSynchronization.ALWAYS);
    }
    mongoTemplate.insert(test);
    IdName idName = new IdName();
    idName.setName("test");
    mongoTemplate.insert(idName);
    User user = new User();
    user.setName("Demo");
    user.setEmail("srini@abspl.in");
    user.setPassword("sdfsdfsdf");
    userService.save(user);
    }
 }

POM.XML

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.abcplusd.sample.mongoapi</groupId>
  <artifactId>sample-mongo-api</artifactId>
  <version>1.0-SNAPSHOT</version>

  <name>Sample Spring Boot Mongo API</name>
  <description>Demo project for Spring Boot Mongo with Spring Data Mongo</description>

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.1.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
  </parent>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <java.version>1.8</java.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.data</groupId>
      <artifactId>spring-data-mongodb</artifactId>
      <version>2.1.0.RELEASE</version>
      <exclusions>
        <exclusion>
          <groupId>org.mongodb</groupId>
          <artifactId>mongo-java-driver</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>org.springframework.data</groupId>
      <artifactId>spring-data-commons</artifactId>
      <version>2.1.0.RELEASE</version>
    </dependency>
    <dependency>
      <groupId>org.mongodb</groupId>
      <artifactId>mongo-java-driver</artifactId>
      <version>3.8.2</version>
    </dependency>
    <dependency>
      <groupId>org.mybatis</groupId>
      <artifactId>mybatis-spring</artifactId>
      <version>1.3.1</version>
    </dependency>
    <dependency>
      <groupId>org.postgresql</groupId>
      <artifactId>postgresql</artifactId>
      <version>42.2.2</version>
    </dependency>
    <dependency>
      <groupId>org.mybatis.spring.boot</groupId>
      <artifactId>mybatis-spring-boot-starter</artifactId>
      <version>1.3.2</version>
    </dependency>
    <dependency>
      <groupId>org.mybatis</groupId>
      <artifactId>mybatis</artifactId>
      <version>3.4.5</version>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

【讨论】：

嘿，你能分享一下你的项目设置（build.gradle 或 pom.xml）吗？我很难在 spring 上设置这个
@Ben 请在上面的答案中找到 pom.xml。
@Srini 什么表示 projectId ？我每次都需要传递什么价值
@pudaykiran ProjectId 用于在 mongodb 中创建动态数据库名称。请参考上述代码 sn-p 中的 MonoConfig 类。

【解决方案3】：

使用 MongoDb 4.0.x，您可以使用事务。如果您使用以下版本，则必须实现两阶段提交。

注意：MongoDb 仅允许您在拥有 ReplicaSet 时使用事务。

要同时为 JPA 和 MongoDb 使用事务，您必须使用 ChainedTransactionManager。过程是：

创建 Jpa 事务管理器
创建 MongoDb 事务管理器
创建将使用上述两个的 ChainedTransactionManager

我的 conf 看起来像这样（我不使用 spring boot，但它应该是等效的）：

Jpa 配置

@Configuration
@EnableTransactionManagement
@EnableJpaRepositories("com....")
public class HibernateConfig {

    //define entity manager, data source and all the stuff needed for your DB

    @Bean("jpaTransactionManager")
    public JpaTransactionManager transactionManager() throws NamingException { 

        JpaTransactionManager transactionManager = new JpaTransactionManager();
        //transactionManager.setEntityManagerFactory(entityManagerFactory().getObject());

        return transactionManager;
    }
}

MongoDb 配置

@Configuration
@EnableMongoRepositories(basePackages = "com....")
public class MongoDbConf extends AbstractMongoClientConfiguration {

    private final Environment environment;

    @Autowired
    public MongoDbConf(Environment environment) {
        this.environment = environment;
    }

    @Override
    public MongoClient mongoClient() {
        String connectionString = environment.getProperty("mongodb.connectionString");

        if(StringUtils.isBlank(connectionString))
            throw new IllegalArgumentException("No connection string to initialize mongo client");

        return MongoClients.create(
                MongoClientSettings.builder()
                        .applyConnectionString(new ConnectionString(connectionString))
                        .applicationName("MY_APP")
                        .build());
    }

    @Override
    protected String getDatabaseName() {
        return environment.getProperty("mongodb.database", "myDB");
    }

    @Bean("mongoDbTransactionManager")
    public MongoTransactionManager transactionManager(MongoDbFactory dbFactory) {
        return new MongoTransactionManager(dbFactory);
    }
}

ChainedTransactionManager 配置

@Configuration
public class ChainedTransactionConf {

    private MongoTransactionManager mongoTransactionManager;
    private JpaTransactionManager jpaTransactionManager;

    @Autowired
    public ChainedTransactionConf(MongoTransactionManager mongoTransactionManager, JpaTransactionManager jpaTransactionManager) {
        this.mongoTransactionManager = mongoTransactionManager;
        this.jpaTransactionManager = jpaTransactionManager;
    }

    @Bean("chainedTransactionManager")
    public PlatformTransactionManager getTransactionManager() {
        ChainedTransactionManager transactionManager = new ChainedTransactionManager(jpaTransactionManager, mongoTransactionManager);
        return transactionManager;
    }

}

mongoDb 存储库示例

@Service
public class MongoDbRepositoryImpl implements MongoDbRepository {

    private static final Logger logger = Logger.getLogger(MongoDbRepositoryImpl.class);

    //MongoOperations will handle a mongo session
    private final MongoOperations operations;

    @Autowired
    public MongoDbRepositoryImpl(MongoOperations operations) {
        this.operations = operations;
    }

    @Override
    public void insertData(Document document) {
        MongoCollection<Document> collection = operations.getCollection("myCollection");
        collection.insertOne(document);
    }

在您的服务中使用事务

@Service
public class DocumentServiceImpl implements DocumentService {

    private final MongoDbRepository mongoDbRepository;
    private final JpaRepository jpaRepository;

    @Autowired
    public DocumentServiceImpl(MongoDbRepository mongoDbRepository,JpaRepository jpaRepository) {
        this.mongoDbRepository = mongoDbRepository;
        this.jpaRepository = jpaRepository;
    }

    @Override
    @Transactional("chainedTransactionManager")
    public void insertNewDoc(Map<String,Object> rawData) {
        //use org.springframework.transaction.annotation.Transactional so you can define used transactionManager
        //jpaRepository.insert...
        Document mongoDoc = new Document(rawData);
        mongoDbRepository.insertData(mongoDoc)

        //you can test like this : breakpoint and throw new IllegalStateException() 
        //to see that data is not commited 
    }

【讨论】：

【解决方案4】：

MongoDB v4.x.x 与 @Transactional 完美配合，它们通过使用以下依赖项和存储库明确支持这一点：-

<dependency>
    <groupId>org.springframework.data</groupId>
    <artifactId>spring-data-releasetrain</artifactId>
    <version>Lovelace-M3</version>
    <type>pom</type>
    <scope>import</scope>
</dependency>
<repositories>
    <repository>
        <id>spring-milestones</id>
        <name>Spring Milestones</name>
        <url>https://repo.spring.io/milestone</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
</repositories>

还有一个 MongoTransactionConfig 类：-

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.MongoTransactionManager;
import org.springframework.data.mongodb.config.AbstractMongoClientConfiguration;
import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;

import com.mongodb.ConnectionString;
import com.mongodb.MongoClientSettings;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;

@Configuration
@EnableMongoRepositories(basePackages = "com.airtel.africa.caxr.repository")
public class MongoTransactionConfig extends AbstractMongoClientConfiguration {

    @Value("${spring.data.mongodb.host}")
    private String host;
    @Value("${spring.data.mongodb.port}")
    private String port;
    @Value("${spring.data.mongodb.database}")
    private String database;

    @Bean
    MongoTransactionManager transactionManager(MongoDbFactory dbFactory) {
        return new MongoTransactionManager(dbFactory);
    }

    @Override
    protected String getDatabaseName() {
        return database;
    }
    @Override
    public MongoClient mongoClient() {
        String connectionString = "mongodb://"+host+":"+port;

        return MongoClients.create(MongoClientSettings.builder()
            .applyConnectionString(new 
        ConnectionString(connectionString)).build());
    }
}

在这里，我将 mongo 与 kafka 一起用作 1 事务，因此如果此处发生任何已检查或未检查的异常，则应回滚 mongo 事务，因此我使用了 @Transactional(rollbackFor = Exception.class):-

@Transactional(rollbackFor = Exception.class)
public void receiveInEventRequest(TransactionDto transactionDto) throws 
    InterruptedException, ExecutionException {
    // db insert    
    TransactionRequest transactionRequest = requestDbDumpService.dumpToDb(transactionDto);
    // kafka insert
    ListenableFuture<SendResult<String, TransactionDto>> kafkaResult = kafkaTemplate.send(kafkaProducerQueueName, “ID”, transactionDto);
    SendResult<String, TransactionDto> kafkaSendResult = kafkaResult.get();
}

【讨论】：

【解决方案5】：

如果有人需要 transactional 支持 reactive 风格 spring boot 和 MongoDb 集成，请查看答案 @987654321 @

【讨论】：