Reactive-MongoDB 异步 Java Driver 解读

电你發表於 2019-10-22 11:54:00

<h2>一、关于异步驱动</h2>
从3.0 版本开始，MongoDB 开始提供异步方式的驱动(Java Async Driver)，这为应用提供了一种更高性能的选择。 但实质上，使用同步驱动(Java Sync Driver)的项目也不在少数，或许是因为先入为主的原因(同步Driver的文档说明更加的完善)，又或者是为了兼容旧的 MongoDB 版本。 无论如何，由于 Reactive 的发展，未来使用异步驱动应该是一个趋势。
在使用 Async Driver 之前，需要对 Reactive 的概念有一些熟悉。
<h2>二、理解 Reactive (响应式)</h2>
响应式(Reactive)是一种异步的、面向数据流的开发方式，最早是来自于.NET 平台上的 Reactive Extensions 库，随后被扩展为各种编程语言的实现。 在著名的 Reactive Manifesto(响应式宣言) 中，对 Reactive 定义了四个特征：
<img class="alignnone size-full wp-image-28753 lazyload" alt="reactive-specs" width="796" height="404" data-src="https://www.mongochina.com/wp-content/uploads/2019/10/383cf74d-1c42-4058-a325-e04913cd0a42.jpg">
<ul>
<li>及时响应(Responsive)：系统能及时的响应请求。</li>
<li>有韧性(Resilient)：系统在出现异常时仍然可以响应，即支持容错。</li>
<li>有弹性(Elastic)：在不同的负载下，系统可弹性伸缩来保证运行。</li>
<li>消息驱动(Message Driven)：不同组件之间使用异步消息传递来进行交互，并确保松耦合及相互隔离。</li>

</ul>
在响应式宣言的所定义的这些系统特征中，无一不与响应式的流有若干的关系，于是乎就有了 2013年发起的 响应式流规范(Reactive Stream Specification)。
https://www.reactive-streams.org/
其中，对于响应式流的处理环节又做了如下定义：
<ul>
<li>具有处理无限数量的元素的能力，即允许流永不结束</li>
<li>按序处理</li>
<li>异步地传递元素</li>
<li>实现非阻塞的负压(back-pressure)</li>

</ul>
Java 平台则是在 JDK 9 版本上发布了对 Reactive Streams 的支持。
下面介绍响应式流的几个关键接口：
<ul>
<li>Publisher Publisher 是数据的发布者。Publisher 接口只有一个方法 subscribe，用于添加数据的订阅者，也就是 Subscriber。</li>
<li>Subscriber Subscriber 是数据的订阅者。Subscriber 接口有4个方法，都是作为不同事件的处理器。在订阅者成功订阅到发布者之后，其 onSubscribe(Subscription s) 方法会被调用。 Subscription 表示的是当前的订阅关系。</li>

</ul>
当订阅成功后，可以使用 Subscription 的 request(long n) 方法来请求发布者发布 n 条数据。发布者可能产生3种不同的消息通知，分别对应 Subscriber 的另外3个回调方法。
数据通知：对应 onNext 方法，表示发布者产生的数据。 错误通知：对应 onError 方法，表示发布者产生了错误。 结束通知：对应 onComplete 方法，表示发布者已经完成了所有数据的发布。 在上述3种通知中，错误通知和结束通知都是终结通知，也就是在终结通知之后，不会再有其他通知产生。
<ul>
<li>Subscription Subscription 表示的是一个订阅关系。除了之前提到的 request 方法之外，还有 cancel 方法用来取消订阅。需要注意的是，在 cancel 方法调用之后，发布者仍然有可能继续发布通知。但订阅最终会被取消。</li>

</ul>
这几个接口的关系如下图所示：
<img class="alignnone size-full wp-image-28754 lazyload" alt="reactive interfaces" width="1116" height="654" data-src="https://www.mongochina.com/wp-content/uploads/2019/10/0.6877509870459404.png">
图片出处：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/
<h2>三、使用示例</h2>
接下来，通过一个简单的例子来演示一下 Reactive 方式的代码风格：
<h3>A. 引入依赖</h3>
<pre><code>
org.mongodb
mongodb-driver-reactivestreams
1.11.0

</code></pre>
> 引入mongodb-driver-reactivestreams 将会自动添加 reactive-streams, bson, mongodb-driver-async组件
<h3>B. 连接数据库</h3>
<pre><code>//服务器实例表
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());

</code></pre>
<h3>C. 实现文档查询</h3>
<pre><code>//获得数据库对象
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.");
}
});

</code></pre>
注意到，与使用同步驱动不同的是，collection.find()方法返回的不是 Cursor，而是一个 FindPublisher对象，这是Publisher接口的一层扩展。 而且，在返回 Publisher 对象时，此时并没有产生真正的数据库IO请求。真正发起请求需要通过调用 Subscription.request()方法。 在上面的代码中，为了读取由 Publisher 产生的结果，通过自定义一个Subscriber，在onSubscribe 事件触发时就执行数据库的请求，之后分别对 onNext、onError、onComplete进行处理。
尽管这种实现方式是纯异步的，但在使用上比较繁琐。试想如果对于每个数据库操作都要完成一个Subscriber 逻辑，那么开发的工作量是巨大的。
为了尽可能复用重复的逻辑，可以对Subscriber的逻辑做一层封装，包含如下功能：
<ul>
<li>使用 List 容器对请求结果进行缓存</li>
<li>实现阻塞等待结果的方法，可指定超时时间</li>
<li>捕获异常，在等待结果时抛出</li>

</ul>
代码如下：
<pre><code>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;
}
}
</code></pre>
借助这个基础的工具类，我们对于文档的异步操作就变得简单多了。 比如对于文档查询的操作可以改造如下：
<pre><code>ObservableSubscriber subscriber = new ObservableSubscriber();
collection.find().subscribe(subscriber);

//结果处理
subscriber.get(15, TimeUnit.SECONDS).forEach( d -&gt; {
System.out.println("Document:" + d.toJson());
});
</code></pre>
当然，这个例子还有可以继续完善，比如使用 List 作为缓存，则要考虑数据量的问题，避免将全部(或超量) 的文档一次性转入内存。
原文地址：https://www.mongochina.com/article/655.html 
来源：https://www.cnblogs.com/wefeng/p/11718970.html

頁: [1]

琼殿技术论坛's Archiver

Reactive-MongoDB 异步 Java Driver 解读