|
对于最新的稳定版本，请使用 Spring Data MongoDB 4.5.2！ |
可追踪光标
默认情况下，当客户端用尽游标提供的所有结果时，MongoDB会自动关闭游标。 在耗尽时关闭光标会将流变成有限流。对于有上限的集合， 您可以使用在客户端之后保持打开状态的 Tailable Cursor 消耗了所有最初返回的数据。
可以使用MongoOperations.createCollection.为此，请提供所需的CollectionOptions.empty().capped()…. |
可追踪游标可以与命令式和响应式 MongoDB API 一起使用。强烈建议使用 响应式变体，因为它的资源密集度较低。但是，如果您无法使用响应式 API，您仍然可以使用消息传递 概念已经在 Spring 生态系统中盛行。
可追踪游标MessageListener
使用同步驱动程序侦听有上限的集合会创建一个长时间运行的阻塞任务，需要将其委托给
一个单独的组件。在这种情况下，我们需要首先创建一个MessageListenerContainer，这将是主要切入点
用于运行特定的SubscriptionRequest.Spring Data MongoDB 已经附带了一个默认实现，该实现
作MongoTemplate并且能够创建和运行Task实例TailableCursorRequest.
以下示例显示了如何使用可追踪游标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)| 1 | 启动容器会初始化资源并启动Task已注册的实例SubscriptionRequest实例。启动后添加的请求会立即运行。 |
| 2 | 定义当Message收到。这Message#getBody()转换为请求的域类型。用Document接收原始结果而不进行转换。 |
| 3 | 设置要收听的集合。 |
| 4 | 为要接收的文档提供可选筛选器。 |
| 5 | 设置消息侦听器以发布传入Messages 到。 |
| 6 | 注册请求。返回的Subscription可用于检查电流Task状态并取消它以释放资源。 |
| 7 | 一旦确定不再需要容器，请不要忘记停止容器。这样做会停止所有运行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 Reactive 存储库通过使用@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();