Reactive-MongoDB如何异步Java Driver
Reactive-MongoDB如何异步Java Driver,很多新手对此不是很清楚,为了帮助大家解决这个难题,下面小编将为大家详细讲解,有这方面需求的人可以来学习下,希望你能有所收获。
一、关于 异步驱动
从3.0 版本开始,MongoDB 开始提供异步方式的驱动(Java Async Driver),这为应用提供了一种更高性能的选择。
但实质上,使用同步驱动(Java Sync Driver)的项目也不在少数,或许是因为先入为主的原因(同步Driver的文档说明更加的完善),又或者是为了兼容旧的 MongoDB 版本。
无论如何,由于 Reactive 的发展,未来使用异步驱动应该是一个趋势。
在使用 Async Driver 之前,需要对 Reactive 的概念有一些熟悉。
二、理解 Reactive (响应式)
响应式(Reactive)是一种异步的、面向数据流的开发方式,最早是来自于.NET 平台上的 Reactive Extensions 库,随后被扩展为各种编程语言的实现。
在著名的 Reactive Manifesto(响应式宣言) 中,对 Reactive 定义了四个特征:
及时响应(Responsive):系统能及时的响应请求。
有韧性(Resilient):系统在出现异常时仍然可以响应,即支持容错。
有弹性(Elastic):在不同的负载下,系统可弹性伸缩来保证运行。
消息驱动(Message Driven):不同组件之间使用异步消息传递来进行交互,并确保松耦合及相互隔离。
在响应式宣言的所定义的这些系统特征中,无一不与响应式的流有若干的关系,于是乎就有了 2013年发起的 响应式流规范(Reactive Stream Specification)。
https://www.reactive-streams.org/
其中,对于响应式流的处理环节又做了如下定义:
具有处理无限数量的元素的能力,即允许流永不结束
按序处理
异步地传递元素
实现非阻塞的负压(back-pressure)
Java 平台则是在 JDK 9 版本上发布了对 Reactive Streams 的支持。
下面介绍响应式流的几个关键接口:
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/
三、使用示例
接下来,通过一个简单的例子来演示一下 Reactive 方式的代码风格:
A. 引入依赖
org.mongodb mongodb-driver-reactivestreams 1.11.0
> 引入mongodb-driver-reactivestreams 将会自动添加 reactive-streams, bson, mongodb-driver-async组件
B. 连接数据库
//服务器实例表List servers =newArrayList(); servers.add(newServerAddress("localhost",27018));//配置构建器MongoClientSettings.Builder settingsBuilder =MongoClientSettings.builder();//传入服务器实例 settingsBuilder.applyToClusterSettings( builder -> builder.hosts(servers));//构建 Client 实例MongoClient mongoClient =MongoClients.create(settingsBuilder.build());
C. 实现文档查询
//获得数据库对象MongoDatabase database = client.getDatabase(databaseName);//获得集合MongoCollection collection = database.getCollection(collectionName);//异步返回PublisherFindPublisher publisher = collection.find();//订阅实现 publisher.subscribe(newSubscriber(){ @Override publicvoid onSubscribe(Subscription s){ System.out.println("start..."); //执行请求 s.request(Integer.MAX_VALUE); } @Override publicvoid onNext(Document document){ //获得文档 System.out.println("Document:"+ document.toJson()); } @Override publicvoid onError(Throwable t){ System.out.println("error occurs."); } @Override publicvoid onComplete(){ System.out.println("finished."); }});
注意到,与使用同步驱动不同的是,collection.find()方法返回的不是 Cursor,而是一个 FindPublisher对象,这是Publisher接口的一层扩展。
而且,在返回 Publisher 对象时,此时并没有产生真正的数据库IO请求。真正发起请求需要通过调用 Subscription.request()方法。
在上面的代码中,为了读取由 Publisher 产生的结果,通过自定义一个Subscriber,在onSubscribe 事件触发时就执行 数据库的请求,之后分别对 onNext、onError、onComplete进行处理。
尽管这种实现方式是纯异步的,但在使用上比较繁琐。试想如果对于每个数据库操作都要完成一个Subscriber 逻辑,那么开发的工作量是巨大的。
为了尽可能复用重复的逻辑,可以对Subscriber的逻辑做一层封装,包含如下功能:
使用 List 容器对请求结果进行缓存
实现阻塞等待结果的方法,可指定超时时间
捕获异常,在等待结果时抛出
代码如下:
publicclassObservableSubscriberimplementsSubscriber{ //响应数据 privatefinalList received; //错误信息 privatefinalList errors; //等待对象 privatefinalCountDownLatch latch; //订阅器 privatevolatileSubscription subscription; //是否完成 privatevolatileboolean completed; publicObservableSubscriber(){ this.received =newArrayList(); this.errors =newArrayList(); this.latch =newCountDownLatch(1); } @Override publicvoid onSubscribe(finalSubscription s){ subscription = s; } @Override publicvoid onNext(final T t){ received.add(t); } @Override publicvoid onError(finalThrowable t){ errors.add(t); onComplete(); } @Override publicvoid onComplete(){ completed =true; latch.countDown(); } publicSubscription getSubscription(){ return subscription; } publicList getReceived(){ return received; } publicThrowable getError(){ if(errors.size()>0){ return errors.get(0); } returnnull; } publicboolean isCompleted(){ return completed; } /** * 阻塞一定时间等待结果 * * @param timeout * @param unit * @return * @throws Throwable */ publicListget(finallong timeout,finalTimeUnit unit)throwsThrowable{ return await(timeout, unit).getReceived(); } /** * 一直阻塞等待请求完成 * * @return * @throws Throwable */ publicObservableSubscriber await()throwsThrowable{ return await(Long.MAX_VALUE,TimeUnit.MILLISECONDS); } /** * 阻塞一定时间等待完成 * * @param timeout * @param unit * @return * @throws Throwable */ publicObservableSubscriber await(finallong timeout,finalTimeUnit unit)throwsThrowable{ subscription.request(Integer.MAX_VALUE); if(!latch.await(timeout, unit)){ thrownewMongoTimeoutException("Publisher onComplete timed out"); } if(!errors.isEmpty()){ throw errors.get(0); } returnthis; }}
借助这个基础的工具类,我们对于文档的异步操作就变得简单多了。
比如对于文档查询的操作可以改造如下:
ObservableSubscriber subscriber =newObservableSubscriber(); collection.find().subscribe(subscriber);//结果处理 subscriber.get(15,TimeUnit.SECONDS).forEach( d ->{ System.out.println("Document:"+ d.toJson());});
当然,这个例子还有可以继续完善,比如使用 List 作为缓存,则要考虑数据量的问题,避免将全部(或超量) 的文档一次性转入内存。
看完上述内容是否对您有帮助呢?如果还想对相关知识有进一步的了解或阅读更多相关文章,请关注行业资讯频道,感谢您对的支持。