跳到主要内容
版本：5.1.1

变更流

Doubao 中英对照 Change Streams

自 MongoDB 3.6 起，变更流 让应用无需尾随 oplog 即可收到变更通知。

备注

变更流支持仅适用于副本集或分片集群。

变更流既可使用命令式 MongoDB Java 驱动程序消费，也可使用响应式 MongoDB Java 驱动程序消费。强烈建议使用响应式变体，因为它资源占用更少。不过，如果您无法使用响应式 API，仍可通过使用 Spring 生态系统中已广泛采用的消息传递概念来获取变更事件。

既可在集合级别、也可在数据库级别进行监听，其中数据库级别的变体会发布该数据库内所有集合的变更。订阅数据库变更流时，请确保为事件类型选用合适的类型，因为跨不同实体类型的转换可能无法正确应用。如有疑问，请使用 Document。

使用 MessageListener 的变更流​

使用同步驱动程序监听变更流会创建一个长时间运行的阻塞任务，该任务需要委托给单独的组件。在这种情况下，我们需要首先创建一个 MessageListenerContainer，它将作为运行特定 SubscriptionRequest 任务的主要入口点。Spring Data MongoDB 已提供一个默认实现，它基于 MongoTemplate 运行，并能够为 ChangeStreamRequest 创建和运行 Task 实例。

以下示例展示了如何将变更流与 MessageListener 实例配合使用：

示例 1. 使用 MessageListener 实例的变更流

MessageListenerContainer container = new DefaultMessageListenerContainer(template);
container.start();                                                                                              1

MessageListener<ChangeStreamDocument<Document>, User> listener = System.out::println;                           2
ChangeStreamRequestOptions options = new ChangeStreamRequestOptions("db", "user", ChangeStreamOptions.empty()); 3

Subscription subscription = container.register(new ChangeStreamRequest<>(listener, options), User.class);       4

// ...

container.stop();                                                                                               5

  • 启动容器会初始化资源，并为已注册的 SubscriptionRequest 实例启动 Task 实例。启动后添加的请求会立即运行。

  • 定义在收到 Message 时调用的监听器。Message#getBody() 会被转换为所请求的领域类型。使用 Document 可接收未经转换的原始结果。

  • 设置要监听的集合，并通过 ChangeStreamOptions 提供额外选项。

  • 注册请求。返回的 Subscription 可用于检查当前 Task 状态，并取消它以释放资源。

  • 当您确定不再需要容器时，不要忘记停止它。这样做会停止容器内所有正在运行的 Task 实例。

DefaultMessageListenerContainer 实现了 SmartLifecycle，默认情况下，当它作为 bean 注册到 Application Context 时会自动启动。

备注

处理过程中出现的错误会传递给 org.springframework.util.ErrorHandler。除非另行说明，默认会应用一个追加日志的 ErrorHandler。
请使用 register(request, body, errorHandler) 来提供额外功能。

响应式变更流​

使用响应式 API 订阅变更流是处理流更自然的方式。尽管如此，其基本构建块（例如 ChangeStreamOptions）仍然相同。以下示例展示了如何使用发出 ChangeStreamEvent 的变更流：

示例 2. 发出 ChangeStreamEvent 的变更流

Flux<ChangeStreamEvent<User>> flux = reactiveTemplate.changeStream(User.class) 1
    .watchCollection("people")
    .filter(where("age").gte(38))                                              2
    .listen();                                                                 3

  • 底层文档应转换为的事件目标类型。省略此项可接收未经转换的原始结果。

  • 使用聚合管道或仅使用查询 Criteria 来过滤事件。

  • 获取变更流事件的 Flux。ChangeStreamEvent#getBody() 会按 (2) 中所请求的领域类型进行转换。

恢复变更流​

变更流可以从您上次离开的位置恢复并继续发出事件。要恢复流，您需要提供恢复令牌或最后已知的服务器时间（UTC）。请使用 ChangeStreamOptions 相应地设置该值。

以下示例展示了如何使用服务器时间设置恢复偏移量：

示例 3. 恢复变更流

Flux<ChangeStreamEvent<User>> resumed = template.changeStream(User.class)
    .watchCollection("people")
    .resumeAt(Instant.now().minusSeconds(1)) 1
    .listen();

  • 您可以通过 getTimestamp 方法获取 ChangeStreamEvent 的服务器时间，或使用通过 getResumeToken 暴露的 resumeToken。

提示

在某些情况下，恢复变更流时 Instant 可能不够精确。为此请使用 MongoDB 原生的 BsonTimestamp。