Spring Data MongoDB Tailable Cursors – Spring Data MongoDB的Tailable Cursors

最后修改: 2019年 6月 29日

中文/混合/英文(键盘快捷键：t)

1. Introduction

1.绪论

In this tutorial, we’re going to discuss how to use MongoDB as an infinite data stream by utilizing tailable cursors with Spring Data MongoDB.

在本教程中，我们将讨论如何通过利用Spring Data MongoDB的可跟踪游标，将MongoDB作为一个无限的数据流。

2. Tailable Cursors

2.可跟踪的游标

When we execute a query, the database driver opens a cursor to supply the matching documents. By default, MongoDB automatically closes the cursor when the client reads all results. Therefore, turning results in a finite data stream.

当我们执行一个查询时，数据库驱动程序会打开一个游标来提供匹配的文档。默认情况下，MongoDB在客户端读取所有结果时自动关闭游标。因此，将结果变成一个有限的数据流。

However, we can use capped collections with a tailable cursor that remains open, even after the client consumed all initially returned data – making the infinite data stream. This approach is useful for applications dealing with event streams, like chat messages, or stock updates.

然而，我们可以使用带有尾部游标的封顶集合，即使在客户端消耗了所有最初返回的数据之后，该游标仍然保持开放状态–这就是无限的数据流。这种方法对于处理事件流的应用程序非常有用，例如聊天信息或股票更新。

Spring Data MongoDB project helps us utilizing reactive database capabilities, including tailable cursors.

Spring Data MongoDB项目帮助我们利用了反应式数据库的能力，包括可跟踪的游标。

3. Setup

3.设置

To demonstrate the mentioned features, we’ll implement a simple logs counter application. Let’s assume there is some log aggregator that collects and persists all logs into a central place – our MongoDB capped collection.

为了演示上述功能，我们将实现一个简单的日志计数器应用程序。让我们假设有一些日志聚合器，收集并持久化所有的日志到一个中心位置–我们的MongoDB上限集合。

Firstly, we’ll use the simple Log entity:

首先，我们将使用简单的Log实体。

@Document
public class Log {
    private @Id String id;
    private String service;
    private LogLevel level;
    private String message;
}

Secondly, we’ll store the logs in our MongoDB capped collection. Capped collections are fixed-size collections that insert and retrieve documents based on the insertion order. We can create them with the MongoOperations.createCollection:

其次，我们将在MongoDB的封顶集合中存储日志。封顶集合是固定大小的集合，可根据插入顺序插入和检索文档。我们可以使用MongoOperations.createCollection来创建它们。

db.createCollection(COLLECTION_NAME, new CreateCollectionOptions()
  .capped(true)
  .sizeInBytes(1024)
  .maxDocuments(5));

For capped collections, we must define the sizeInBytes property. Moreover, the maxDocuments specifies the maximum number of documents a collection can have. Once reached, the older documents will be removed from the collection.

对于有上限的集合，我们必须定义sizeInBytes属性。此外，maxDocuments规定了一个集合可以拥有的最大文件数。一旦达到，旧的文件将被从集合中删除。

Thirdly, we’ll use the appropriate Spring Boot starter dependency:

第三，我们将使用适当的Spring Boot启动器的依赖性。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
    <versionId>2.2.2.RELEASE</versionId>
</dependency>

4. Reactive Tailable Cursors

4.反应式可跟踪游标

We can consume tailable cursors with both the imperative and the reactive MongoDB API. It’s highly recommended to use the reactive variant.

我们可以使用imperative和reactive MongoDB API来消费可跟踪光标。我们强烈建议使用反应式变体。

Let’s implement WARN level logs counter using a reactive approach. We’re able to create infinite stream queries with ReactiveMongoOperations.tail method.

让我们使用反应式方法实现WARN级日志计数器。我们能够用ReactiveMongoOperations.tail方法创建无限的流查询。

A tailable cursor remains open and emits data – a Flux of entities – as new documents arrive in a capped collection and match the filter query:

一个可跟踪的游标保持开放，并在新的文档到达封顶的集合并与过滤器查询相匹配时发出数据–一个Flux的实体。

private Disposable subscription;

public WarnLogsCounter(ReactiveMongoOperations template) {
    Flux<Log> stream = template.tail(
      query(where("level").is(LogLevel.WARN)), 
      Log.class);
    subscription = stream.subscribe(logEntity -> 
      counter.incrementAndGet()
    );
}

Once the new document, having the WARN log level, is persisted in the collection, the subscriber (lambda expression) will increment the counter.

一旦具有WARN日志级别的新文件被持久化到集合中，订阅者（lambda表达式）将增加计数器。

Finally, we should dispose of the subscription to close the stream:

最后，我们应该处理掉这个订阅，以关闭流。

public void close() {
    this.subscription.dispose();
}

Also, please note that tailable cursors may become dead, or invalid if the query initially returns no match. In other words, even if new persisted documents match the filter query, the subscriber will not be able to receive them. This is a known limitation of MongoDB tailable cursors. We must ensure that there are matching documents in the capped collection, before creating a tailable cursor.

另外，请注意，如果查询最初没有返回匹配结果，可跟踪的游标可能会变成死的，或者是无效的。换句话说，即使新的持久化文档与过滤查询相匹配，订阅者也将无法接收它们。这是MongoDB可跟踪游标的一个已知限制。我们必须确保在创建一个可追踪游标之前，在封顶的集合中存在匹配的文档。

5. Tailable Cursors with a Reactive Repository

5.带有反应式存储库的可跟踪游标

Spring Data projects offer a repository abstraction for different data stores, including the reactive versions.

Spring Data项目为不同的数据存储提供了一个存储库抽象，包括反应式版本。

MongoDB is no exception. Please check the Spring Data Reactive Repositories with MongoDB article for more details.

MongoDB 也不例外。请查看Spring Data Reactive Repositories with MongoDB文章以了解更多细节。

Moreover, MongoDB reactive repositories support infinite streams by annotating a query method with @Tailable. We can annotate any repository method returning Flux or other reactive types capable of emitting multiple elements:

此外，MongoDB 反应式存储库通过用@Tailable注释查询方法来支持无限流。我们可以注释任何返回Flux或其他能够发出多个元素的反应式类型的存储库方法。

public interface LogsRepository extends ReactiveCrudRepository<Log, String> {
    @Tailable
    Flux<Log> findByLevel(LogLevel level);
}

Let’s count INFO logs using this tailable repository method:

让我们用这种可跟踪的存储库方法来统计INFO日志。

private Disposable subscription;

public InfoLogsCounter(LogsRepository repository) {
    Flux<Log> stream = repository.findByLevel(LogLevel.INFO);
    this.subscription = stream.subscribe(logEntity -> 
      counter.incrementAndGet()
    );
}

Similarly, as for WarnLogsCounter, we should dispose of the subscription to close the stream:

同样地，对于WarnLogsCounter，我们应该处理掉订阅，以关闭流。

public void close() {
    this.subscription.dispose();
}

6. Tailable Cursors with a MessageListener

6.带有MessageListener的可跟踪游标

Nevertheless, if we can’t use the reactive API, we can leverage Spring’s messaging concept.

尽管如此，如果我们不能使用反应式API，我们可以利用Spring的消息传递概念。

First, we need to create a MessageListenerContainer which will handle sent SubscriptionRequest objects. The synchronous MongoDB driver creates a long-running, blocking task that listens to new documents in the capped collection.

首先，我们需要创建一个MessageListenerContainer，它将处理发送的SubscriptionRequest对象。同步 MongoDB 驱动程序创建了一个长期运行的阻塞任务，用于监听封顶集合中的新文档。

Spring Data MongoDB ships with a default implementation capable of creating and executing Task instances for a TailableCursorRequest:

Spring Data MongoDB带有默认实现，能够为TailableCursorRequest创建和执行Task实例：。

private String collectionName;
private MessageListenerContainer container;
private AtomicInteger counter = new AtomicInteger();

public ErrorLogsCounter(MongoTemplate mongoTemplate,
  String collectionName) {
    this.collectionName = collectionName;
    this.container = new DefaultMessageListenerContainer(mongoTemplate);

    container.start();
    TailableCursorRequest<Log> request = getTailableCursorRequest();
    container.register(request, Log.class);
}

private TailableCursorRequest<Log> getTailableCursorRequest() {
    MessageListener<Document, Log> listener = message -> 
      counter.incrementAndGet();

    return TailableCursorRequest.builder()
      .collection(collectionName)
      .filter(query(where("level").is(LogLevel.ERROR)))
      .publishTo(listener)
      .build();
}

TailableCursorRequest creates a query filtering only the ERROR level logs. Each matching document will be published to the MessageListener that will increment the counter.

TailableCursorRequest创建一个查询，只过滤ERROR级别的日志。每个匹配的文件将被发布到MessageListener，它将增加计数器。

Note that we still need to ensure that the initial query returns some results. Otherwise, the tailable cursor will be immediately closed.

注意，我们仍然需要确保初始查询返回一些结果。否则，可跟踪游标将被立即关闭。

In addition, we should not forget to stop the container once we no longer need it:

此外，我们不应忘记，一旦我们不再需要这个容器，就要停止它。

public void close() {
    container.stop();
}

7. Conclusion

7.结语

MongoDB capped collections with tailable cursors help us receive information from the database in a continuous way. We can run a query that will keep giving results until explicitly closed. Spring Data MongoDB offers us both the blocking and the reactive way of utilizing tailable cursors.

MongoDB的封顶集合与可跟踪游标帮助我们以连续的方式从数据库接收信息。我们可以运行一个查询，该查询将持续给出结果，直到明确关闭。Spring Data MongoDB为我们提供了利用可跟踪游标的阻塞式和反应式方式。

The source code of the complete example is available over on GitHub.

完整示例的源代码可在GitHub上获得，。