使用 Spring Data 进行 MongoDB 4.0 事务处理
在 MongoDB 4.0 中，ACID 事务已经用于 Document 的存储，强制维护全执行或全不执行的数据一致性状态。所以让我们直接在 synchronous 模型和 reactive 执行模型中验证该特性。
在撰写本文时，MongoDB 的多文档事务在单副本集中受支持，并且给用户的感受像是在使用关系型数据库的事务一样。看到驱动程序提供的 API 立刻会感觉到回到家里一样。
try (ClientSession session = client.startSession()) {
session.startTransaction();
try {
collection.insertOne(session, documentOne);
collection.insertOne(session, documentTwo);
session.commitTransaction();
} catch (Exception e) {
session.abortTransaction();
}
}
逻辑会话建立在 MongoDB的基础上，当然，事务，当然还有事务构建了基础。
逻辑会话通过帮助跨分布式节点协调操作来为MangoDB的因果一致性和事务建立基础。客户端从 client.startSession() 中获取会话，会话的生命周期不应过长，在不再使用的时候应该立刻关闭它。所以确保使用 close() 来关闭客户端会话。
在底层的协议层，上面的代码片段将会转变为如下一系列命令，你可以清楚的发现在每个命令中都包含会话(lsid)。startTransaction 标志位将会与第一个命令一起发送，表示事务的开始。在事务完成后，发送commitTransaction 表示事务的提交。
{ insert: "col", ordered: true, $db: "db",
$clusterTime: { … },
lsid: { id: { $binary: { base64 : "I3M7Nj…", … } } },
txnNumber: 1,
startTransaction: true,
documents: [ { … } ] }
{ insert: "col", ordered: true, $db: "db",
$clusterTime: { … },
lsid: { id: { $binary: { base64 : "I3M7Nj…", … } } },
txnNumber: 1,
autocommit: false,
documents: [ { …} ] }
{ commitTransaction: 1,
$db: "admin",
$clusterTime: { … },
lsid: { id: { $binary: { base64 : "I3M7Nj…", … } } },
txnNumber: 1 }
随着即将发布的 Spring Data Lovelace 版本，MongoDB 模块将提供对 synchronous 和 reactive 事务的支持。
我们从 synchronous 模式开始，你可以能已经非常熟悉 Spring 框架对事务的支持 (Spring Framework’s transaction support) 。因此，一个 MongoTransactionManager 的存在并不令人吃惊。该事务管理器是在命令式世界中基于注解的事务支持的入口。
现在，因为 MongoDB 在早期版本中不支持事务，你必须明确的在 ApplicationContext 中注册 MongoTransactionManager 。如果你这样做的话，MongoTemplate 将会开始参与管理事务。这是一个你需要记住的要点。下面的例子展示了你应该如何配置事务管理器。
@Configuration
class Config extends AbstractMongoConfiguration {
@Bean
MongoTransactionManager transactionManager(MongoDbFactory dbFactory) {
return new MongoTransactionManager(dbFactory);
}
}
@Service
class DocumentService {
private final MongoOperations operations;
DocumentService(MongoOperations operations) {
this.operations = operations;
}
@Transactional
void insertDocuments() {
operations.insert(documentOne);
operations.insert(documentTwo);
}
}
非常直播的操作是吧？但是，这里有一些隐含的缺点。 集群环境下的事务支持在下一个 MongDB 的 release 主要版本中才会支持，因此在您使用时会发生错误。此外，作为一个 MongoDB 的用户，你可能已经习惯了他提供的所有的便利，但一些特性在事务中无法使用了，包括了几乎所有的元命令，创建集合，索引以及受此使用集合时隐式创建集合。为了避免错误和折腾，请务必设置所需的结构。此外，某些命令可能还会有一些不同。例如使用集合集合统计信息的 count 命令可能在事务中并不准确。命令将会出错并且需要使用聚合计数文档，当前的驱动已经提供一个替代方法 countDocuments 来利用聚合策略解决这个问题。
考虑到这一点，让我们继续进行 reactive 使用的部分。
在 MongoDB的ReactiveStreams驱动程序 提供了一个反应切入点多文档交易。将本机驱动程序管道Publisher化为 Reactor 类型可让您表达事务用法，如下所示：
Mono.from(client.startSession()).flatMap(session -> {
session.startTransaction();
return Mono.from(collection.insertOne(session, documentOne))
.then(Mono.from(collection.insertOne(session, documentTwo)))
.onErrorResume(e -> Mono.from(session.abortTransaction())
.then(Mono.error(e)))
.flatMap(val -> Mono.from(session.commitTransaction())
.then(Mono.just(val)))
.doFinally(signal -> session.close());
});
不管事务的结果是成功还是回滚，我们都需要保证事务的终止。因此，onErrorResume(...) 保证了事务在失败的时候可以回滚，然后在 flatMap(...) 中提交，这两个阶段都保存了主流 (main flow) 的结果或错误。不同于 sync 部分，截止撰稿时还没有 reactive 模型可用的事务管理器能够让你通过注解 @Transactional 那样简单的完成事务工作。相反，你需要通过 ReactiveMongoTemplate.inTransaction(...) 获取一个 transaction 闭包。它在保持主流 (main flow) 结果的同事负责所有必需的会话，提交和终止操作。回调方法中的操作在MongoDB事务中执行，而外部的处理步骤将不会影响事务。这意味着闭包之外的处理错误不会导致事务终止，就像下面的例子描述的那样。
template.inTransaction().execute(action ->
// All code in here runs inside the transaction
action.insert(documentOne).then(action.insert(documentTwo)
).flatMap(val -> {
// An exception here does not affect the transaction
});
在这个例子中，你能够通过流访问到 ClientSession，它存放在 Reactor 的 Context 中，并且你可以通过 ReactiveMongoContext.getSession() 来获取它。
最后一件事情：我们非常高兴你能够尝试并且给我们提供一些反馈，所以请查看 Spring Data Examples，您可以在其中找到相关的 项目。
如果你想要学习更多有关 Spring Data 或者通用的 Spring eco-system，即将在华盛顿召开的 SpringOne Platform 会议对您来说是一个非常好的机会。查看会话并注册。