Reactive-MongoDB 异步 Java Driver 解读

时间 2019-11-07

标签 reactive mongodb 异步 java driver 解读栏目 MongoDB 繁體版

原文原文链接

1、关于异步驱动

从3.0 版本开始，MongoDB 开始提供异步方式的驱动(Java Async Driver)，这为应用提供了一种更高性能的选择。
但实质上，使用同步驱动(Java Sync Driver)的项目也不在少数，或许是由于先入为主的缘由(同步Driver的文档说明更加的完善)，又或者是为了兼容旧的 MongoDB 版本。
不管如何，因为 Reactive 的发展，将来使用异步驱动应该是一个趋势。php

在使用 Async Driver 以前，须要对 Reactive 的概念有一些熟悉。html

2、理解 Reactive (响应式)

响应式(Reactive)是一种异步的、面向数据流的开发方式，最先是来自于.NET 平台上的 Reactive Extensions 库，随后被扩展为各类编程语言的实现。
在著名的 Reactive Manifesto(响应式宣言) 中，对 Reactive 定义了四个特征：java

及时响应(Responsive)：系统能及时的响应请求。
有韧性(Resilient)：系统在出现异常时仍然能够响应，即支持容错。
有弹性(Elastic)：在不一样的负载下，系统可弹性伸缩来保证运行。
消息驱动(Message Driven)：不一样组件之间使用异步消息传递来进行交互，并确保松耦合及相互隔离。

在响应式宣言的所定义的这些系统特征中，无一不与响应式的流有若干的关系，因而乎就有了 2013年发起的响应式流规范(Reactive Stream Specification)。react

https://www.reactive-streams.org/git

其中，对于响应式流的处理环节又作了以下定义：github

具备处理无限数量的元素的能力，即容许流永不结束
按序处理
异步地传递元素
实现非阻塞的负压(back-pressure)

Java 平台则是在 JDK 9 版本上发布了对 Reactive Streams 的支持。mongodb

下面介绍响应式流的几个关键接口：数据库

Publisher
Publisher 是数据的发布者。Publisher 接口只有一个方法 subscribe，用于添加数据的订阅者，也就是 Subscriber。
Subscriber
Subscriber 是数据的订阅者。Subscriber 接口有4个方法，都是做为不一样事件的处理器。在订阅者成功订阅到发布者以后，其 onSubscribe(Subscription s) 方法会被调用。
Subscription 表示的是当前的订阅关系。

当订阅成功后，可使用 Subscription 的 request(long n) 方法来请求发布者发布 n 条数据。发布者可能产生3种不一样的消息通知，分别对应 Subscriber 的另外3个回调方法。编程

数据通知：对应 onNext 方法，表示发布者产生的数据。
错误通知：对应 onError 方法，表示发布者产生了错误。
结束通知：对应 onComplete 方法，表示发布者已经完成了全部数据的发布。
在上述3种通知中，错误通知和结束通知都是终结通知，也就是在终结通知以后，不会再有其余通知产生。缓存

Subscription
Subscription 表示的是一个订阅关系。除了以前提到的 request 方法以外，还有 cancel 方法用来取消订阅。须要注意的是，在 cancel 方法调用以后，发布者仍然有可能继续发布通知。但订阅最终会被取消。

这几个接口的关系以下图所示：

图片出处：http://wiki.jikexueyuan.com/index.php/project/reactor-2.0/05.html

MongoDB 的异步驱动为 mongo-java-driver-reactivestreams 组件，其实现了 Reactive Stream 的上述接口。

> 除了 reactivestream 以外，MongoDB 的异步驱动还包含 RxJava 等风格的版本，有兴趣的读者能够进一步了解

http://mongodb.github.io/mongo-java-driver-reactivestreams/1.11/getting-started/quick-tour-primer/

3、使用示例

接下来，经过一个简单的例子来演示一下 Reactive 方式的代码风格：

A. 引入依赖

org.mongodb
    mongodb-driver-reactivestreams
    1.11.0

> 引入mongodb-driver-reactivestreams 将会自动添加 reactive-streams, bson, mongodb-driver-async组件

B. 链接数据库

//服务器实例表
List servers = new ArrayList();
servers.add(new ServerAddress("localhost", 27018));

//配置构建器
MongoClientSettings.Builder settingsBuilder = MongoClientSettings.builder();

//传入服务器实例
settingsBuilder.applyToClusterSettings(
        builder -&gt; builder.hosts(servers));

//构建 Client 实例
MongoClient mongoClient = MongoClients.create(settingsBuilder.build());

C. 实现文档查询

//得到数据库对象
MongoDatabase database = client.getDatabase(databaseName);

//得到集合
MongoCollection collection = database.getCollection(collectionName);

//异步返回Publisher
FindPublisher publisher = collection.find();

//订阅实现
publisher.subscribe(new Subscriber() {
    @Override
    public void onSubscribe(Subscription s) {
        System.out.println("start...");
        //执行请求
        s.request(Integer.MAX_VALUE);

    }
    @Override
    public void onNext(Document document) {
        //得到文档
        System.out.println("Document:" + document.toJson());
    }

    @Override
    public void onError(Throwable t) {
        System.out.println("error occurs.");
    }

    @Override
    public void onComplete() {
        System.out.println("finished.");
    }
});

注意到，与使用同步驱动不一样的是，collection.find()方法返回的不是 Cursor，而是一个 FindPublisher对象，这是Publisher接口的一层扩展。
并且，在返回 Publisher 对象时，此时并无产生真正的数据库IO请求。真正发起请求须要经过调用 Subscription.request()方法。
在上面的代码中，为了读取由 Publisher 产生的结果，经过自定义一个Subscriber，在onSubscribe 事件触发时就执行数据库的请求，以后分别对 onNext、onError、onComplete进行处理。

尽管这种实现方式是纯异步的，但在使用上比较繁琐。试想若是对于每一个数据库操做都要完成一个Subscriber 逻辑，那么开发的工做量是巨大的。

为了尽量复用重复的逻辑，能够对Subscriber的逻辑作一层封装，包含以下功能：

使用 List 容器对请求结果进行缓存
实现阻塞等待结果的方法，可指定超时时间
捕获异常，在等待结果时抛出

代码以下：

public class ObservableSubscriber implements Subscriber {

    //响应数据
    private final List received;
    //错误信息
    private final List errors;
    //等待对象
    private final CountDownLatch latch;
    //订阅器
    private volatile Subscription subscription;
    //是否完成
    private volatile boolean completed;

    public ObservableSubscriber() {
        this.received = new ArrayList();
        this.errors = new ArrayList();
        this.latch = new CountDownLatch(1);
    }

    @Override
    public void onSubscribe(final Subscription s) {
        subscription = s;
    }

    @Override
    public void onNext(final T t) {
        received.add(t);
    }

    @Override
    public void onError(final Throwable t) {
        errors.add(t);
        onComplete();
    }

    @Override
    public void onComplete() {
        completed = true;
        latch.countDown();
    }

    public Subscription getSubscription() {
        return subscription;
    }

    public List getReceived() {
        return received;
    }

    public Throwable getError() {
        if (errors.size() &gt; 0) {
            return errors.get(0);
        }
        return null;
    }

    public boolean isCompleted() {
        return completed;
    }

    /**
     * 阻塞必定时间等待结果
     *
     * @param timeout
     * @param unit
     * @return
     * @throws Throwable
     */
    public List get(final long timeout, final TimeUnit unit) throws Throwable {
        return await(timeout, unit).getReceived();
    }

    /**
     * 一直阻塞等待请求完成
     *
     * @return
     * @throws Throwable
     */
    public ObservableSubscriber await() throws Throwable {
        return await(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    }

    /**
     * 阻塞必定时间等待完成
     *
     * @param timeout
     * @param unit
     * @return
     * @throws Throwable
     */
    public ObservableSubscriber await(final long timeout, final TimeUnit unit) throws Throwable {
        subscription.request(Integer.MAX_VALUE);
        if (!latch.await(timeout, unit)) {
            throw new MongoTimeoutException("Publisher onComplete timed out");
        }
        if (!errors.isEmpty()) {
            throw errors.get(0);
        }
        return this;
    }
}

借助这个基础的工具类，咱们对于文档的异步操做就变得简单多了。
好比对于文档查询的操做能够改造以下：

ObservableSubscriber subscriber = new ObservableSubscriber();
collection.find().subscribe(subscriber);

//结果处理
subscriber.get(15, TimeUnit.SECONDS).forEach( d -&gt; {
    System.out.println("Document:" + d.toJson());
});

固然，这个例子还有能够继续完善，好比使用 List 做为缓存，则要考虑数据量的问题，避免将所有(或超量) 的文档一次性转入内存。

原文地址：https://www.mongochina.com/article/655.html