跳到主要内容
版本：5.1.1

可尾游标

Doubao 中英对照 Tailable Cursors

默认情况下，当客户端耗尽游标所提供的全部结果时，MongoDB 会自动关闭游标。在耗尽时关闭游标会将流变为有限流。对于固定大小集合，您可以使用可尾游标，使其在客户端消费完最初返回的数据后保持打开。

提示

可使用 MongoOperations.createCollection 创建固定大小集合。为此，请提供所需的 CollectionOptions.empty().capped()…。

可尾游标既可使用命令式 MongoDB API 消费，也可使用响应式 MongoDB API 消费。强烈建议使用响应式变体，因为它资源占用更少。不过，如果您无法使用响应式 API，仍可使用 Spring 生态系统中已广泛采用的消息传递概念。

使用 MessageListener 的可尾游标​

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

以下示例展示了如何将可尾游标与 MessageListener 实例配合使用：

示例 1. 使用 MessageListener 实例的可尾游标

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

MessageListener<Document, User> listener = System.out::println;                     2

TailableCursorRequest request = TailableCursorRequest.builder()
  .collection("orders")                                                             3
  .filter(query(where("value").lt(100)))                                            4
  .publishTo(listener)                                                              5
  .build();

container.register(request, User.class);                                            6

// ...

container.stop();                                                                   7

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

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

  • 设置要监听的集合。

  • 为要接收的文档提供可选过滤器。

  • 设置用于将传入 Message 发布到的消息监听器。

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

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

响应式可尾游标​

将可尾游标与响应式数据类型配合使用，可以构建无限流。可尾游标会保持打开，直到被外部关闭。当新文档到达固定大小集合时，它会发出数据。

如果查询未返回任何匹配结果，或者游标返回了集合“末尾”的文档而应用随后删除了该文档，可尾游标可能会失效或变为无效。以下示例展示了如何创建并使用无限流查询：

示例 2. 使用 ReactiveMongoOperations 的无限流查询

Flux<Person> stream = template.tail(query(where("name").is("Joe")), Person.class);

Disposable subscription = stream.doOnNext(person -> System.out.println(person)).subscribe();

// …

// Later: Dispose the subscription to close the stream
subscription.dispose();

Spring Data MongoDB 响应式仓库通过使用 @Tailable 注解标注查询方法来支持无限流。如下例所示，这适用于返回 Flux 以及其他能够发出多个元素的响应式类型的方法：

示例 3. 使用 ReactiveMongoRepository 的无限流查询

public interface PersonRepository extends ReactiveMongoRepository<Person, String> {

  @Tailable
  Flux<Person> findByFirstname(String firstname);

}

Flux<Person> stream = repository.findByFirstname("Joe");

Disposable subscription = stream.doOnNext(System.out::println).subscribe();

// …

// Later: Dispose the subscription to close the stream
subscription.dispose();