|
对于最新的稳定版本，请使用 Spring Data MongoDB 4.5.2！ |
会话和交易
从 3.6 版开始，MongoDB 支持会话的概念。
会话的使用启用了 MongoDB 的因果一致性模型，该模型保证以尊重其因果关系的顺序运行作。
这些被分成ServerSessioninstances 和ClientSession实例。
在本节中，当我们谈到会话时，我们指的是ClientSession.
| 客户端会话中的作与会话外部的作不隔离。 |
双MongoOperations和ReactiveMongoOperations提供用于绑定ClientSession到作。MongoCollection和MongoDatabase使用实现 MongoDB 的集合和数据库接口的会话代理对象，因此您无需在每次调用时添加会话。
这意味着对MongoCollection#find()被委托给MongoCollection#find(ClientSession).
诸如(Reactive)MongoOperations#getCollection返回原生 MongoDB Java Driver 网关对象（例如MongoCollection）本身提供了专门的方法ClientSession.
这些方法不是会话代理的。
您应该提供ClientSession直接与MongoCollection或MongoDatabase而不是通过其中一个#execute回调MongoOperations. |
ClientSession 支持
以下示例显示了会话的用法：
-
Imperative
-
Reactive
ClientSessionOptions sessionOptions = ClientSessionOptions.builder()
.causallyConsistent(true)
.build();
ClientSession session = client.startSession(sessionOptions); (1)
template.withSession(() -> session)
.execute(action -> {
Query query = query(where("name").is("Durzo Blint"));
Person durzo = action.findOne(query, Person.class); (2)
Person azoth = new Person("Kylar Stern");
azoth.setMaster(durzo);
action.insert(azoth); (3)
return azoth;
});
session.close() (4)| 1 | 从服务器获取新会话。 |
| 2 | 用MongoOperation方法和以前一样。
这ClientSession会自动应用。 |
| 3 | 确保关闭ClientSession. |
| 4 | 关闭会话。 |
在处理DBRef实例，尤其是延迟加载的实例，必须不要关闭ClientSession在加载所有数据之前。
否则，延迟获取失败。 |
ClientSessionOptions sessionOptions = ClientSessionOptions.builder()
.causallyConsistent(true)
.build();
Publisher<ClientSession> session = client.startSession(sessionOptions); (1)
template.withSession(session)
.execute(action -> {
Query query = query(where("name").is("Durzo Blint"));
return action.findOne(query, Person.class)
.flatMap(durzo -> {
Person azoth = new Person("Kylar Stern");
azoth.setMaster(durzo);
return action.insert(azoth); (2)
});
}, ClientSession::close) (3)
.subscribe(); (4)| 1 | 获取一个Publisher用于新会话检索。 |
| 2 | 用ReactiveMongoOperation方法和以前一样。
这ClientSession自动获取并应用。 |
| 3 | 确保关闭ClientSession. |
| 4 | 在您订阅之前，什么都不会发生。 有关详细信息，请参阅 Project Reactor 参考指南。 |
通过使用Publisher提供实际会话时，您可以将会话获取推迟到实际订阅点。
不过，您仍然需要在完成后关闭会话，以免过时的会话污染服务器。
使用doFinally挂钩execute调用ClientSession#close()当您不再需要会话时。
如果您希望对会话本身有更多的控制权，您可以获取ClientSession通过驱动程序，并通过Supplier.
反应性使用ClientSession仅限于模板 API 的使用。
目前没有与响应式存储库的会话集成。 |
MongoDB 事务
除非您指定MongoTransactionManager在应用程序上下文中，事务支持为 DISABLED。
您可以使用setSessionSynchronization(ALWAYS)参与正在进行的非原生 MongoDB 事务。 |
要以编程方式完全控制事务，您可能需要在MongoOperations.
以下示例显示了编程事务控制：
-
Imperative
-
Reactive
ClientSession session = client.startSession(options); (1)
template.withSession(session)
.execute(action -> {
session.startTransaction(); (2)
try {
Step step = // ...;
action.insert(step);
process(step);
action.update(Step.class).apply(Update.set("state", // ...
session.commitTransaction(); (3)
} catch (RuntimeException e) {
session.abortTransaction(); (4)
}
}, ClientSession::close) (5)| 1 | 获取新的ClientSession. |
| 2 | 开始交易。 |
| 3 | 如果一切按预期进行，请提交更改。 |
| 4 | 有东西坏了，所以把所有东西都回滚。 |
| 5 | 完成后不要忘记关闭会话。 |
前面的示例允许您在使用会话范围的事务行为时完全控制事务行为MongoOperations实例，以确保会话传递到每个服务器调用。
为了避免这种方法带来的一些开销，您可以使用TransactionTemplate以消除手动交易流的一些噪音。
Mono<DeleteResult> result = Mono
.from(client.startSession()) (1)
.flatMap(session -> {
session.startTransaction(); (2)
return Mono.from(collection.deleteMany(session, ...)) (3)
.onErrorResume(e -> Mono.from(session.abortTransaction()).then(Mono.error(e))) (4)
.flatMap(val -> Mono.from(session.commitTransaction()).then(Mono.just(val))) (5)
.doFinally(signal -> session.close()); (6)
});| 1 | 首先，我们显然需要启动会话。 |
| 2 | 一旦我们有了ClientSession在手头，开始交易。 |
| 3 | 通过传递ClientSession到作。 |
| 4 | 如果作异常完成，我们需要停止事务并保留错误。 |
| 5 | 或者当然，在成功的情况下提交更改。 仍然保留作结果。 |
| 6 | 最后，我们需要确保关闭会话。 |
上述作的罪魁祸首在于保持主流DeleteResult而不是通过commitTransaction()或abortTransaction()，这导致设置相当复杂。
除非您指定ReactiveMongoTransactionManager在应用程序上下文中，事务支持为 DISABLED。
您可以使用setSessionSynchronization(ALWAYS)参与正在进行的非原生 MongoDB 事务。 |
使用 TransactionTemplate / TransactionalOperator 进行事务
Spring Data MongoDB 事务同时支持TransactionTemplate和TransactionalOperator.
TransactionTemplate / TransactionalOperator-
Imperative
-
Reactive
template.setSessionSynchronization(ALWAYS); (1)
// ...
TransactionTemplate txTemplate = new TransactionTemplate(anyTxManager); (2)
txTemplate.execute(new TransactionCallbackWithoutResult() {
@Override
protected void doInTransactionWithoutResult(TransactionStatus status) { (3)
Step step = // ...;
template.insert(step);
process(step);
template.update(Step.class).apply(Update.set("state", // ...
}
});| 1 | 在模板 API 配置期间启用事务同步。 |
| 2 | 创建TransactionTemplate使用提供的PlatformTransactionManager. |
| 3 | 在回调中，ClientSession和交易已经注册。 |
状态的变化MongoTemplate在运行时（您可能认为在前面列表中的第 1 项中可能）可能会导致线程和可见性问题。 |
template.setSessionSynchronization(ALWAYS); (1)
// ...
TransactionalOperator rxtx = TransactionalOperator.create(anyTxManager,
new DefaultTransactionDefinition()); (2)
Step step = // ...;
template.insert(step);
Mono<Void> process(step)
.then(template.update(Step.class).apply(Update.set("state", …))
.as(rxtx::transactional) (3)
.then();| 1 | 为事务参与启用事务同步。 |
| 2 | 创建TransactionalOperator使用提供的ReactiveTransactionManager. |
| 3 | TransactionalOperator.transactional(…)为所有上游作提供事务管理。 |
使用 MongoTransactionManager 和 ReactiveMongoTransactionManager 进行事务
MongoTransactionManager / ReactiveMongoTransactionManager是通往众所周知的 Spring 事务支持的网关。
它允许应用程序使用 Spring 的托管事务功能。
这MongoTransactionManager绑定一个ClientSession到线程，而ReactiveMongoTransactionManager正在使用ReactorContext为此。MongoTemplate检测会话并相应地对与事务关联的这些资源进行作。MongoTemplate还可以参与其他正在进行的交易。
以下示例演示如何创建和使用具有MongoTransactionManager:
MongoTransactionManager / ReactiveMongoTransactionManager-
Imperative
-
Reactive
@Configuration
static class Config extends AbstractMongoClientConfiguration {
@Bean
MongoTransactionManager transactionManager(MongoDatabaseFactory dbFactory) { (1)
return new MongoTransactionManager(dbFactory);
}
@Bean
MongoTemplate mongoTemplate(MongoDatabaseFactory dbFactory) { (1)
return new MongoTemplate(dbFactory);
}
// ...
}
@Component
public class StateService {
@Transactional
void someBusinessFunction(Step step) { (2)
template.insert(step);
process(step);
template.update(Step.class).apply(Update.set("state", // ...
};
});| 1 | 注册MongoTransactionManager在应用程序上下文中。
另外，请确保使用相同的MongoDatabaseFactory创建时MongoTemplate参与同一范围内的交易MongoDatabaseFactory. |
| 2 | 将方法标记为事务性。 |
@Transactional(readOnly = true)建议MongoTransactionManager以启动一个将ClientSession到传出请求。 |
@Configuration
public class Config extends AbstractReactiveMongoConfiguration {
@Bean
ReactiveMongoTransactionManager transactionManager(ReactiveMongoDatabaseFactory factory) { (1)
return new ReactiveMongoTransactionManager(factory);
}
@Bean
ReactiveMongoTemplate reactiveMongoTemplate(ReactiveMongoDatabaseFactory dbFactory) { (1)
return new ReactiveMongoTemplate(dbFactory);
}
// ...
}
@Service
public class StateService {
@Transactional
Mono<UpdateResult> someBusinessFunction(Step step) { (2)
return template.insert(step)
.then(process(step))
.then(template.update(Step.class).apply(Update.set("state", …));
};
});| 1 | 注册ReactiveMongoTransactionManager在应用程序上下文中。
另外，请确保使用相同的ReactiveMongoDatabaseFactory创建时ReactiveMongoTemplate参与同一范围内的交易ReactiveMongoDatabaseFactory. |
| 2 | 将方法标记为事务性。 |
@Transactional(readOnly = true)建议ReactiveMongoTransactionManager以启动一个将ClientSession到传出请求。 |
控制特定于 MongoDB 的事务选项
事务服务方法可能需要特定的事务选项来运行事务。
Spring Data MongoDB 的事务管理器支持评估事务标签，例如@Transactional(label = { "mongo:readConcern=available" }).
默认情况下，使用mongo:prefix 由MongoTransactionOptionsResolver默认配置。
交易标签由TransactionAttribute并可用于编程事务控制TransactionTemplate和TransactionalOperator.
由于它们的声明性，@Transactional(label = …)提供了一个很好的起点，也可以作为文档。
目前，支持以下选项：
- 最大提交时间（Max Commit Time）
-
控制 commitTransaction作在服务器上的最大执行时间。 该值的格式对应于 ISO-8601 持续时间格式，如与
Duration.parse(…).用法：
mongo:maxCommitTime=PT1S - 阅读关注
-
设置事务的读取关注点。
用法：
mongo:readConcern=LOCAL|MAJORITY|LINEARIZABLE|SNAPSHOT|AVAILABLE - 读取首选项
-
设置事务的读取首选项。
用法：
mongo:readPreference=PRIMARY|SECONDARY|SECONDARY_PREFERRED|PRIMARY_PREFERRED|NEAREST - 写关注
-
设置事务的写入关注点。
用法：
mongo:writeConcern=ACKNOWLEDGED|W1|W2|W3|UNACKNOWLEDGED|JOURNALED|MAJORITY
| 加入外部事务的嵌套事务不会影响初始事务选项，因为事务已经启动。 事务选项仅在启动新事务时应用。 |
事务中的特殊行为
在事务中，MongoDB 服务器的行为略有不同。
连接设置
MongoDB 驱动程序提供了一个专用的副本集名称配置选项，将驱动程序调入自动检测模式。 此选项有助于在事务期间识别主副本集节点和命令路由。
确保添加replicaSet到 MongoDB URI。
有关更多详细信息，请参阅连接字符串选项。 |
收集作
MongoDB 不支持事务中的集合作，例如集合创建。 这也会影响首次使用时发生的动态集合创建。 因此，请确保拥有所有必需的结构。
瞬态错误
MongoDB可以为事务作期间引发的错误添加特殊标签。
这些可能表示暂时性故障，这些故障可能仅通过重试作而消失。
我们强烈建议将 Spring Retry 用于这些目的。
然而，可以覆盖MongoTransactionManager#doCommit(MongoTransactionObject)实现 Retry Commit Operation 行为，如 MongoDB 参考手册中所述。
计数
Mongo数据库count根据可能无法反映交易中实际情况的收款统计数据进行作。
服务器在发出count命令。
一次MongoTemplate检测活动事务，全部公开count()方法被转换并委托给聚合框架，使用$match和$count运算符， 保留Query设置，例如collation.
在聚合计数帮助程序中使用地理命令时，会应用限制。 不能使用以下运算符，必须替换为其他运算符：
-
$where→$expr -
$near→$geoWithin跟$center -
$nearSphere→$geoWithin跟$centerSphere
使用Criteria.near(…)和Criteria.nearSphere(…)必须重写为Criteria.within(…)各自Criteria.withinSphere(…).
这同样适用于nearquery 关键字，必须更改为within.
另请参阅 MongoDB JIRA 票证 DRIVERS-518 以获取进一步参考。
以下代码段显示count会话绑定闭包内的用法：
session.startTransaction();
template.withSession(session)
.execute(ops -> {
return ops.count(query(where("state").is("active")), Step.class)
});上面的代码片段在以下命令中实现：
db.collection.aggregate(
[
{ $match: { state: "active" } },
{ $count: "totalEntityCount" }
]
)而不是：
db.collection.find( { state: "active" } ).count()