会话与事务

从版本 3.6 开始，MongoDB 支持会话的概念。会话的使用使 MongoDB 的 因果一致性 模型成为可能，该模型保证以尊重其因果关系的方式运行操作。这些操作分为 ServerSession 实例和 ClientSession 实例。在本节中，当我们谈论会话时，我们指的是 ClientSession。

客户端会话中的操作与会话外的操作不是隔离的。

MongoOperations 和 ReactiveMongoOperations 都提供了将 ClientSession 绑定到操作的网关方法。MongoCollection 和 MongoDatabase 使用实现 MongoDB 集合和数据库接口的会话代理对象，因此您无需在每次调用时都添加会话。这意味着对 MongoCollection#find() 的潜在调用将委托给 MongoCollection#find(ClientSession)。

诸如 (Reactive)MongoOperations#getCollection 之类的方法返回原生的 MongoDB Java 驱动程序网关对象（例如 MongoCollection），这些对象本身提供了 ClientSession 的专用方法。这些方法未进行会话代理。当您直接与 MongoCollection 或 MongoDatabase 交互，而不是通过 MongoOperations 上的某个 #execute 回调交互时，您应该在需要的地方提供 ClientSession。

ClientSession 支持

以下示例展示了会话的使用

  • 命令式

  • 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，您可以将会话获取推迟到实际订阅点。但是，您仍需要在完成后关闭会话，以免用陈旧的会话污染服务器。当您不再需要会话时，使用 execute 上的 doFinally 钩子调用 ClientSession#close()。如果您希望对会话本身有更多的控制权，您可以通过驱动程序获取 ClientSession 并通过 Supplier 提供它。

ClientSession 的反应式使用仅限于模板 API 使用。目前没有与反应式存储库的会话集成。

MongoDB 事务

从版本 4 开始，MongoDB 支持 事务。事务建立在 会话 之上，因此需要一个活动的 ClientSession。

除非您在应用程序上下文中指定 MongoTransactionManager，否则事务支持将禁用。您可以使用 setSessionSynchronization(ALWAYS) 来参与正在进行的非原生 MongoDB 事务。

要完全以编程方式控制事务，您可能希望使用 MongoOperations 上的会话回调。

以下示例展示了编程事务控制

编程事务

  • 命令式

  • 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，否则事务支持将禁用。您可以使用 setSessionSynchronization(ALWAYS) 来参与正在进行的非原生 MongoDB 事务。

使用 TransactionTemplate / TransactionalOperator 进行事务

Spring Data MongoDB 事务支持 TransactionTemplate 和 TransactionalOperator。

使用 TransactionTemplate / TransactionalOperator 进行事务

  • 命令式

  • 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 使用提供的 PlatformTransactionManager 创建 TransactionTemplate。
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 使用提供的 ReactiveTransactionManager 创建 TransactionalOperator。
3 TransactionalOperator.transactional(…) 为所有上游操作提供事务管理。

使用 MongoTransactionManager 和 ReactiveMongoTransactionManager 进行事务

MongoTransactionManager / ReactiveMongoTransactionManager 是众所周知的 Spring 事务支持的网关。它允许应用程序使用 Spring 的托管事务功能。MongoTransactionManager 将 ClientSession 绑定到线程，而 ReactiveMongoTransactionManager 使用 ReactorContext 来实现此目的。MongoTemplate 检测会话并根据与事务关联的这些资源进行操作。MongoTemplate 还可以参与其他正在进行的事务。以下示例展示了如何使用 MongoTransactionManager 创建和使用事务

使用 MongoTransactionManager / ReactiveMongoTransactionManager 进行事务

  • 命令式

  • 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。此外，请确保在创建 MongoTemplate 时使用相同的 MongoDatabaseFactory，以便在同一 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。此外，请确保在创建 ReactiveMongoTemplate 时使用相同的 ReactiveMongoDatabaseFactory，以便在同一 ReactiveMongoDatabaseFactory 范围内参与事务。
2 将方法标记为事务性。
@Transactional(readOnly = true) 建议 ReactiveMongoTransactionManager 也启动一个事务，该事务将 ClientSession 添加到传出请求中。

控制 MongoDB 特定的事务选项

事务性服务方法可能需要特定的事务选项来运行事务。Spring Data MongoDB 的事务管理器支持评估事务标签，例如 @Transactional(label = { "mongo:readConcern=available" })。

默认情况下，使用 mongo: 前缀的标签命名空间由默认配置的 MongoTransactionOptionsResolver 进行评估。事务标签由 TransactionAttribute 提供，并通过 TransactionTemplate 和 TransactionalOperator 可用于编程事务控制。由于它们的声明性性质，@Transactional(label = …) 提供了一个很好的起点，也可以作为文档。

目前支持以下选项

最大提交时间

控制服务器上 commitTransaction 操作的最大执行时间。值的格式与 Duration.parse(…) 中使用的 ISO-8601 持续时间格式相对应。

用法：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) 以实现 MongoDB 参考手册中概述的 重试提交操作 行为。

计数

MongoDB 的 count 操作基于集合统计信息，这可能无法反映事务内的实际情况。在多文档事务中发出 count 命令时，服务器将响应错误 50851。一旦 MongoTemplate 检测到活动事务，所有公开的 count() 方法都将转换为使用 $match 和 $count 运算符的聚合框架，保留 Query 设置（例如 collation）。

在聚合计数助手中使用地理命令时，会受到限制。以下运算符不能使用，必须替换为不同的运算符

  • $where → $expr

  • $near → $geoWithin 与 $center

  • $nearSphere → $geoWithin 与 $centerSphere

使用 Criteria.near(…) 和 Criteria.nearSphere(…) 的查询必须重写为 Criteria.within(…) 和 Criteria.withinSphere(…)。对于存储库查询方法中的 near 查询关键字也适用，必须将其更改为 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()
© . This site is unofficial and not affiliated with VMware.