/ /

Java Reactive Streams 驱动程序

Docs 主页

客户端库

Java

Java Reactive Streams 驱动程序

Docs 主页

客户端库

Java

Java Reactive Streams 驱动程序

教程：与 Spring Boot 集成

Overview

Spring Boot 是一个开源Java框架，可简化生产就绪应用程序的开发。当与Spring Data MongoDB结合使用时，该框架提供了一种与MongoDB数据库交互的有效方法。

本教程演示如何构建一个响应式 Spring Boot应用程序，该应用程序使用 Spring Data MongoDB对MongoDB 数据库执行操作。您可以使用此示例现金余额应用程序调用执行以下操作的 REST API：

创建或获取帐户
在一个账户或两个账户之间执行事务

Spring Data MongoDB

Spring Data MongoDB为MongoDB提供存储库支持，提供两种与数据库交互的主节点 (primary node in the replica set)方式：

ReactiveMongoRepository：更高级别的抽象，提供对常见数据库操作的访问权限
ReactiveMongoTemplate：较低级别的抽象，可更好地控制MongoDB操作和查询构造

多文档 ACID 事务

MongoDB支持多文档ACID事务，允许您执行多个满足原子性、一致性、隔离性和持久性ACID 一致性保证的操作。Spring Data MongoDB通过 @Transactional 注解或 TransactionalOperator 类提供对事务的内置支持。本教程中的示例应用程序配置ACID 事务支持以确保数据库操作的原子性。

要学习；了解有关MongoDB事务的更多信息，请参阅MongoDB Server手册中的事务。

Tutorial

本教程介绍如何下载实施简化现金余额系统的示例应用程序。该应用程序允许您创建、检索和更新银行账户余额。要查看完整的应用程序，请参阅 mdb-spring-boot-reactive GitHub存储库。

完成本教程，学习；了解如何执行以下操作：

下载示例Spring Boot应用程序
配置应用程序以连接到MongoDB
了解应用程序结构和关键组件
运行应用程序。
发送 POST 请求以创建银行帐户并执行事务

下载并配置连接

验证先决条件

在开始本教程之前，请确保您已准备好以下组件：

已配置集群的MongoDB Atlas帐户。要学习；了解如何创建集群，请参阅MongoDB入门指南。
卷曲。
Java21 或更高版本。
Maven.3 5或更高版本。

下载示例应用程序

通过运行以下命令，从MongoDB 开发者 GitHub 存储库克隆示例应用程序：

git clone git@github.com:mongodb-developer/mdb-spring-boot-reactive.git

此存储库包含一个响应式 Spring Boot应用程序，该应用程序使用 Spring Data MongoDB与MongoDB 数据库进行交互。

配置MongoDB连接

导航到示例应用程序目录中的 src/main/resources/application.properties文件。在此文件中，将 spring.data.mongodb.uri属性设立为您的Atlas连接 URI，如以下代码所示：

spring.data.mongodb.uri=<connection URI>

将 <connection URI> 占位符替换为您的Atlas连接 URI。

启用模式验证

从根目录运行以下命令，为应用程序设立模式验证：

mongosh "<connection URI>" --file setup.js

此命令创建一个约束条件，确保银行帐户余额永远不会低于 0。将 <connection URI> 占位符替换为您的Atlas连接 URI。

了解应用程序结构

查看帐户管理终结点

AccountController.java文件包含用于创建和获取帐户的REST API终结点。该文件定义了一个将文档插入 txn-demo.accounts集合的POST 方法终结点和一个根据其 accountNum 值查找文档的GET 方法终结点。以下代码展示了这些方法：

@RestController
public class AccountController {
   //...
   @PostMapping("/account")
   public Mono<Account> createAccount(@RequestBody Account account) {
       return accountRepository.save(account);
   }
   @GetMapping("/account/{accountNum}")
   public Mono<Account> getAccount(@PathVariable String accountNum) {
       return accountRepository.findByAccountNum(accountNum)
           .switchIfEmpty(Mono.error(new AccountNotFoundException()));
   }
   //...
}

每个终结点从 AccountRepository 返回一个 Mono<Account> 类型，是一个 ReactiveMongoRepository 接口，根本的Java Reactive Streams驱动程序的抽象。

查看余额事务终结点

AccountController 类还包括以下事务终结点：

添加帐户余额的 .../debit 终结点
从帐户余额中减去的 .../credit 终结点
执行从一个帐户到另一个帐户的转账的 .../transfer 终结点

这些终结点具有以下定义：

@RestController
public class AccountController {
   //...
   @PostMapping("/account/{accountNum}/debit")
   public Mono<Txn> debitAccount(@PathVariable String accountNum,
                                 @RequestBody Map<String, Object> requestBody) {
       //...
       txn.addEntry(new TxnEntry(accountNum, amount));
       return txnService.saveTransaction(txn).flatMap(txnService::executeTxn);
   }
   @PostMapping("/account/{accountNum}/credit")
   public Mono<Txn> creditAccount(@PathVariable String accountNum,
                                  @RequestBody Map<String, Object> requestBody) {
       //...
       txn.addEntry(new TxnEntry(accountNum, -amount));
       return txnService.saveTransaction(txn).flatMap(txnService::executeTxn);
   }
   @PostMapping("/account/{from}/transfer")
   public Mono<Txn> transfer(@PathVariable String from,
                             @RequestBody TransferRequest transferRequest) {
       //...
       txn.addEntry(new TxnEntry(from, -amount));
       txn.addEntry(new TxnEntry(to, amount));
       return txnService.saveTransaction(txn).flatMap(txnService::executeTxn);
   }
   //...
}

一个 TxnEntry对象代表对单个帐户的更改，而 Txn 可以由一个或多个 TxnEntry 对象组成。在示例代码中，借方和信用终结点创建一个新的 TxnEntry对象，转账终结点创建两个 TxnEntry 对象。

查看自定义查询方法

AccountRepository 接口扩展了 ReactiveMongoRepository 并定义了以下查询方法：

public interface AccountRepository extends ReactiveMongoRepository<Account, String> {
   @Query("{accountNum:'?0'}")
   Mono<Account> findByAccountNum(String accountNum);
   @Update("{'$inc':{'balance': ?1}}")
   Mono<Long> findAndIncrementBalanceByAccountNum(String accountNum, double increment);
}

这些代码使用以下注解，允许您在查询MongoDB时使用占位符，并动态替换为方法参数：

@Query：注释 findByAccountNum() 方法并指定查询条件以查找文档。?0 占位符将替换为 accountNum 参数值。
@Update：注解 findAndIncrementBalanceByAccountNum() 方法并指定要对匹配文档执行的更新操作。将 ?1 占位符替换为 increment 参数值，以使用 MongoDB 的 $inc操作符增加余额。

注意

Although the POST endpoint described in a previous step uses the accountRepository.save() method, you do not need to define this method. The save() method and many other base methods are already declared by interfaces in the inheritance chain of ReactiveMongoRepository.

了解ACID 事务服务

TxnService 类处理ACID 事务执行。该类包括以下代码：

@Service
public class TxnService {
   //...
   public Mono<Txn> saveTransaction(Txn txn) {
       return txnTemplate.save(txn);
   }
   public Mono<Txn> executeTxn(Txn txn) {
       return updateBalances(txn)
           .onErrorResume(DataIntegrityViolationException.class
               /*lambda expression to handle error*/)
           .onErrorResume(AccountNotFoundException.class
               /*lambda expression to handle error*/)
           .then(txnTemplate.findAndUpdateStatusById(txn.getId(), TxnStatus.SUCCESS));
   }
   public Flux<Long> updateBalances(Txn txn) {
       Flux<Long> updatedCounts = Flux.fromIterable(txn.getEntries()).concatMap(
           entry -> accountRepository.findAndIncrementBalanceByAccountNum(
               entry.getAccountNum(), entry.getAmount())
       );
       return updatedCounts.handle(/*...*/);
   }
}

TxnService 类包括以下方法：

saveTransaction()：将 Txn文档保存到 transactions集合中
executeTxn()：调用 updateBalances() 方法，然后更新 Txn文档中的ACID 事务状态
updateBalances()：遍历每个 TxnEntry 并对每个 account文档进行相应更新

查看ACID 事务模板

上一步中描述的 saveTransaction() 和 executeTxn() 方法均使用 TxnTemplate 类中定义的方法。这些方法具有以下定义：

@Service
public class TxnTemplate {
   //...
   public Mono<Txn> save(Txn txn) {
       return template.save(txn);
   }
   public Mono<Txn> findAndUpdateStatusById(String id, TxnStatus status) {
      Query query = query(where("_id").is(id));
      Update update = update("status", status);
      FindAndModifyOptions options = FindAndModifyOptions.options().returnNew(true);
      return template.findAndModify(query, update, options, Txn.class);
   }
   //...
}

这些方法使用 ReactiveMongoTemplate 与MongoDB交互。

配置多文档事务

在帐户之间转账时，可以使用多文档事务。跨两个帐户的更新必须是原子性的，事务可确保数据原子性。

为了访问权限Spring 的ACID 事务支持，示例应用程序将 ReactiveMongoTransactionManager bean 添加到 ReactiveMongoConfig.java文件：

@Configuration
public class ReactiveMongoConfig extends AbstractReactiveMongoConfiguration {
   //...
   @Bean
   ReactiveMongoTransactionManager transactionManager(ReactiveMongoDatabaseFactory dbFactory) {
       return new ReactiveMongoTransactionManager(dbFactory);
   }
}

您可以使用 TransactionalOperator对象或 @Transactional 注解来定义ACID 事务范围。TxnService 类显示了这两种方法。

提示

Spring Data 事务

要学习；了解有关 Spring Data MongoDB中事务和会话的更多信息，请参阅会话和事务 Spring Data 文档中的会话和事务。

测试应用程序

运行应用程序。

要启动应用程序，请从项目的根目录运行以下命令：

mvn spring-boot:run

启动应用程序后，您可以通过 http://localhost:8080访问权限它。然后，您可以使用REST API终结点创建账户并执行事务。

创建账户

要创建帐号为 12345 的新银行帐户，请在Shell中运行以下命令：

curl --location 'localhost:8080/account' \
--header 'Content-Type: application/json' \
--data '{
   "accountNum": "12345"
}'

此命令向 /account 终结点发送 POST请求，以在 accounts集合中创建新的银行帐户文档。

增加账户余额

要将钱存入在上一步中创建的帐户，请在Shell中运行以下命令：

curl --location 'localhost:8080/account/12345/debit' \
--header 'Content-Type: application/json' \
--data '{
   "amount": 1000
}'

此命令向 .../debit 终结点发送请求，将相应 account文档的 amount 值增加 1000。

在账户之间转账

要将资金从一个帐户转账到另一个帐户，请创建一个帐号为 67890 的新帐户：

curl --location 'localhost:8080/account' \
--header 'Content-Type: application/json' \
--data '{
   "accountNum": "67890"
}'

然后，通过运行以下命令，将 500 美元转入此新帐户：

curl --location 'localhost:8080/account/12345/transfer' \
--header 'Content-Type: application/json' \
--data '{
   "to": "67890",
   "amount": 500
}'

此命令向 .../transfer 终结点发送请求以完成余额转移。

其他资源

要学习；了解有关 Spring Data MongoDB 的更多信息，请参阅 Spring Data MongoDB参考文档。

要查看完整的示例应用程序，请参阅mdb-spring-boot-reactive GitHub存储库。