使用vert.x 2.0,RxJava和mongoDB创建simpe RESTful服务

中断了将近半年后发表了一篇新文章。 在本文中,我们将快速了解如何开始使用vert.x,更有趣的是,如何使用RxJava简化异步系统的编程。 我们将涵盖以下主题:

  • 使用Maven创建一个空的vert.x项目
  • 导入IntelliJ并创建一个简单的HTTP服务器
  • 使用vert.x mongoDB持久性模块从mongoDB加载数据
  • 通过REST界面公开拉链
  • 用RxJava观察者替换回调

首先要做的很简单,我们只是使用标准的Maven原型创建一个vert.x项目。 (请注意,可以从github下载完整的最终示例: https : //github.com/josdirksen/smartjava/tree/master/vertx-demo-1 )

使用Maven创建一个空的vert.x项目

转到要在其中创建vert.x项目的目录,键入以下内容,然后按Enter键:

jos@Joss-MacBook-Pro.local:~/Dev/playground$ mvn archetype:generate -Dfilter=io.vertx:

这显示了所有可用的io.vertx原型(在这种情况下只有1个)

[INFO] Scanning for projects...
[INFO]                                                                         
[INFO] ------------------------------------------------------------------------
[INFO] Building Maven Stub Project (No POM) 1
[INFO] ------------------------------------------------------------------------
[INFO] 
[INFO] >>> maven-archetype-plugin:2.2:generate (default-cli) @ standalone-pom >>>
[INFO] 
[INFO] <<< maven-archetype-plugin:2.2:generate (default-cli) @ standalone-pom <<<
[INFO] 
[INFO] --- maven-archetype-plugin:2.2:generate (default-cli) @ standalone-pom ---
[INFO] Generating project in Interactive mode
[INFO] No archetype defined. Using maven-archetype-quickstart (org.apache.maven.archetypes:maven-archetype-quickstart:1.0)
Choose archetype:
1: remote -> io.vertx:vertx-maven-archetype (-)
Choose a number or apply filter (format: [groupId:]artifactId, case sensitive contains): :

由于只有一个,因此只需输入“ 1”,然后按Enter。 接下来,它将显示您可以选择的版本。 在此示例中,我选择了2.0.1-最终版本。

Choose io.vertx:vertx-maven-archetype version: 
1: 1.0.0-beta1
2: 1.0.0-beta2
3: 1.0.0-beta3
4: 1.0.0-CR1
5: 1.0.0-CR2
6: 2.0.0-final
7: 2.0.1-final
Choose a number: 7:

输入“ 7”,然后按Enter。 接下来的步骤使您可以定义项目的名称和版本:

Define value for property 'groupId': : org.smartjava
Define value for property 'artifactId': : vertx-demo-1
Define value for property 'version':  1.0-SNAPSHOT: : 
Define value for property 'package':  org.smartjava: : 
Confirm properties configuration:
groupId: org.smartjava
artifactId: vertx-demo-1
version: 1.0-SNAPSHOT
package: org.smartjavaY: : Y

输入您可以在上面看到的值(或使用自己的值),最后输入“ Y”以确认您的选择。 现在将创建一个项目:

[INFO] ----------------------------------------------------------------------------
[INFO] Using following parameters for creating project from Archetype: vertx-maven-archetype:2.0.1-final
[INFO] ----------------------------------------------------------------------------
[INFO] Parameter: groupId, Value: org.smartjava
[INFO] Parameter: artifactId, Value: vertx-demo-1
[INFO] Parameter: version, Value: 1.0-SNAPSHOT
[INFO] Parameter: package, Value: org.smartjava
[INFO] Parameter: packageInPathFormat, Value: org/smartjava
[INFO] Parameter: package, Value: org.smartjava
[INFO] Parameter: version, Value: 1.0-SNAPSHOT
[INFO] Parameter: groupId, Value: org.smartjava
[INFO] Parameter: artifactId, Value: vertx-demo-1
[INFO] project created from Archetype in dir: /Users/jos/Dev/playground/vertx-demo-1
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 5:37.710s
[INFO] Finished at: Sun Nov 24 14:55:12 CET 2013
[INFO] Final Memory: 9M/24M
[INFO] ------------------------------------------------------------------------

要测试一切是否正确,只需转到刚刚创建的目录,然后运行“ mvn install”。 这将下载所有必需的库,运行一些测试并将您的项目安装到本地Maven存储库。 现在我们有了一个maven项目,我们可以将其加载到我们最喜欢的IDE中。 就我而言,我使用IntelliJ,但是Eclipse的工作方式几乎相同。

导入IntelliJ并创建一个简单的HTTP服务器

启动IntelliJ并选择“文件->导入项目”,导航到maven创建的目录并导入项目。

选择要导入的文件或目录

只需在所有问题上单击“下一步”,您就可以在IntelliJ中拥有一个项目。 如果您基于此原型创建项目,则会自动获得许多可以试用的顶点。 groovy中定义了其中的几个。 IntelliJ会自动尝试编译它们,但是由于找不到合适的编译器,因此编译/生成过程失败。 在此示例中,我们将首先关注vert.x的Java部分,因此只需从“ src / main / resources”和“ test / resources / integration_tests / groovy”目录中删除.groovy文件。

现在,我们可以通过使用maven安装模块,然后调用'vertx:runModIDEA'目标,直接通过maven使用提供的处理程序运行vert.x。 请注意,您确实需要先调用'mvn:compile'来查看所做的更改。 如果您不想使用maven从IDE运行项目,则还可以使用其他方法,即使用'org.vertx.java.platform.impl.cli.Starter'类启动vert.x。直接从IDE。 在IntelliJ中,您为此创建以下启动配置:

Run_Debug配置2

如果运行此命令,仍然会看到错误。 像这样:

Exception in thread "main" java.lang.ClassNotFoundException: org.vertx.java.platform.impl.cli.Starterat java.net.URLClassLoader$1.run(URLClassLoader.java:366)at java.net.URLClassLoader$1.run(URLClassLoader.java:355)at java.security.AccessController.doPrivileged(Native Method)at java.net.URLClassLoader.findClass(URLClassLoader.java:354)at java.lang.ClassLoader.loadClass(ClassLoader.java:424)at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)at java.lang.ClassLoader.loadClass(ClassLoader.java:357)at java.lang.Class.forName0(Native Method)at java.lang.Class.forName(Class.java:190)at com.intellij.rt.execution.application.AppMain.main(AppMain.java:113)

之所以会这样,是因为在由vert.x原型创建的pom.xml中,将vert.x库指定为“提供”。 作为快速解决方案,请打开pom.xml并将三个io.vertx依赖项的范围从“提供”更改为“编译”。 现在,当您从IntelliJ运行此启动器时,vert.x将正确启动。

/Library/Java/JavaVirtualMachines/jdk1.7.0_25.jdk/Contents/Home/bin/java -Didea.launcher.port=7543 "-Didea.launcher.bin.path=/Applications/IntelliJ IDEA 12.app/bin" -Dfile.encoding=UTF-8 -classpath "..." com.intellij.rt.execution.application.AppMain org.vertx.java.platform.impl.cli.Starter runmod org.smartjava~vertx-demo-1~1.0-SNAPSHOT
log4j:WARN No appenders could be found for logger (io.netty.util.internal.logging.InternalLoggerFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Nov 24, 2013 3:43:26 PM org.vertx.java.core.logging.impl.JULLogDelegate info
INFO: Module org.smartjava~vertx-demo-1~1.0-SNAPSHOT successfully installed
Nov 24, 2013 3:43:26 PM org.vertx.java.core.logging.impl.JULLogDelegate info
INFO: PingVerticle started
Nov 24, 2013 3:43:26 PM org.vertx.java.core.logging.impl.JULLogDelegate info
INFO: Succeeded in deploying module

现在,我们已经在IntelliJ中设置了项目,并且可以直接从IDE轻松运行它(并使用ctrl-F5重新启动它),让我们开始创建一个简单的HTTP服务器,以便我们可以看到浏览器中的一些输出以进行测试更容易(请注意,比我在这里显示的测试vert.x和verticles更好的方法,但这是另一篇文章)。 打开PingVerticle.java文件,并将启动方法替换为以下代码:

package org.smartjava;import org.vertx.java.core.Handler;
import org.vertx.java.core.http.HttpServerRequest;
import org.vertx.java.platform.Verticle;public class PingVerticle extends Verticle {public void start() {vertx.createHttpServer().requestHandler(new Handler<HttpServerRequest>() {@Overridepublic void handle(HttpServerRequest httpServerRequest) {httpServerRequest.response().end("Hello smartjava");}}).listen(8888);container.logger().info("Webserver started, listening on port: 8888");}
}

运行此命令,然后打开浏览器到localhost:8888,您将看到以下内容。

本地主机_8888

这是您在vert.x中创建并直接从IDE运行的Web服务器。 小菜一碟到目前为止。 现在,让我们获取一些数据来处理。

使用vert.x mongoDB持久性模块从mongoDB加载数据

我不会深入探讨如何安装mongoDB,互联网上有足够的文章对此进行了解释。 如果您在Mac上运行并安装了macport,则只需使用以下命令行即可安装mongoDB:

sudo port install mongodb

在本文的其余部分中,我假设您已经安装了mongoDB,并且可以从控制台获得其命令行工具。 我们需要做的第一件事是获取一些数据。 在此示例中,我们将使用可从mongoDB网站下载的邮政编码列表: http ://media.mongodb.org/zips.json。 下载此文件,打开控制台并运行以下命令,首先启动mongoDB,然后将此zip列表导入mongoDB。

jos@Joss-MacBook-Pro.local:~/Dev/playground/vertx-demo-1$ mkdir data
jos@Joss-MacBook-Pro.local:~/Dev/playground/vertx-demo-1$ mongod --dbpath ./data/
Sun Nov 24 16:23:51.765 [initandlisten] MongoDB starting : pid=77755 port=27017 dbpath=./data/ 64-bit host=Joss-MacBook-Pro.local
Sun Nov 24 16:23:51.765 [initandlisten] db version v2.4.5
Sun Nov 24 16:23:51.765 [initandlisten] git version: nogitversion
Sun Nov 24 16:23:51.765 [initandlisten] build info: Darwin Joss-MacBook-Pro.local 12.4.0 Darwin Kernel Version 12.4.0: Wed May  1 17:57:12 PDT 2013; root:xnu-2050.24.15~1/RELEASE_X86_64 x86_64 BOOST_LIB_VERSION=1_54
Sun Nov 24 16:23:51.765 [initandlisten] allocator: tcmalloc
Sun Nov 24 16:23:51.765 [initandlisten] options: { dbpath: "./data/" }
Sun Nov 24 16:23:51.766 [initandlisten] journal dir=./data/journal
Sun Nov 24 16:23:51.766 [initandlisten] recover : no journal files present, no recovery needed
Sun Nov 24 16:23:51.779 [FileAllocator] allocating new datafile ./data/local.ns, filling with zeroes...
Sun Nov 24 16:23:51.779 [FileAllocator] creating directory ./data/_tmp
Sun Nov 24 16:23:51.812 [FileAllocator] done allocating datafile ./data/local.ns, size: 16MB,  took 0.031 secs
Sun Nov 24 16:23:51.853 [FileAllocator] allocating new datafile ./data/local.0, filling with zeroes...
Sun Nov 24 16:23:52.254 [FileAllocator] done allocating datafile ./data/local.0, size: 64MB,  took 0.4 secs
Sun Nov 24 16:23:52.260 [initandlisten] command local.$cmd command: { create: "startup_log", size: 10485760, capped: true } ntoreturn:1 keyUpdates:0  reslen:37 480ms
Sun Nov 24 16:23:52.260 [initandlisten] waiting for connections on port 27017
Sun Nov 24 16:23:52.260 [websvr] admin web console waiting for connections on port 28017

现在我们可以使用mongoImport导入下载的邮政编码:

jos@Joss-MacBook-Pro.local:~/Dev/playground/vertx-demo-1$ wget http://media.mongodb.org/zips.json
--2013-11-24 16:25:45--  http://media.mongodb.org/zips.json
Resolving media.mongodb.org... 54.230.131.14, 54.230.131.51, 54.230.128.129, ...
Connecting to media.mongodb.org|54.230.131.14|:80... connected.
HTTP request sent, awaiting response... 200 OK
Length: 2871006 (2.7M) [application/json]
Saving to: `zips.json'100%[======================================>] 2,871,006   2.20M/s   in 1.2s    2013-11-24 16:25:47 (2.20 MB/s) - `zips.json' saved [2871006/2871006]jos@Joss-MacBook-Pro.local:~/Dev/playground/vertx-demo-1$ mongoimport --db vertx --collection zips --file ./zips.json 
connected to: 127.0.0.1
Sun Nov 24 16:26:28.337 check 9 29470
Sun Nov 24 16:26:28.458 imported 29470 objects
jos@Joss-MacBook-Pro.local:~/Dev/playground/vertx-demo-1$

如果您已在IntelliJ中安装了mongoDB插件,则可以轻松测试它是否正常工作:

PingVerticle.java-[vertx-demo-1]-vertx-demo-1-[〜_Dev_playground_vertx-demo-1]

此时,我们只需要从vert.x调用mongoDB实例并加载数据即可。 为此,我们将使用mongodb持久性库。 首先,我们需要将此模块添加到Maven构建中(请注意,这主要用于我们要在内部进行调试的vert.x解析此模块本身):

<dependency><groupId>io.vertx</groupId><artifactId>mod-mongo-persistor</artifactId><version>2.1.0-SNAPSHOT</version><scope>compile</scope></dependency>

Vert.x有一个非常有趣的模块系统(另一篇文章也有此内容),要使用此mongo-persistor,我们首先需要将其部署为模块。 这实际上很容易做到:

// load the general config object, loaded by using -config on command line
JsonObject appConfig = container.config();// deploy the mongo-persistor module, which we'll use for persistence
container.deployModule("io.vertx~mod-mongo-persistor~2.1.0-SNAPSHOT", appConfig.getObject("mongo-persistor"));

在这里,我们要做的是加载该模块的配置,然后调用该模块的名称以及配置的相关部分的deployModule。 首先,让我们看看我们用于此的配置:

{"mongo-persistor" : {"address": "mongodb-persistor","host": "localhost","port": 27017,"pool_size": 10,"db_name": "vertx"}
}

没什么难的。 我们只是将mongo-persister单元指向我们的mongoDB实例。 您可能会遇到的一个问题是,如何在vert.x中获取此文件。 为此,我们只需要对启动器进行一些小的更改,并从以下项更改程序参数:

runmod org.smartjava~vertx-demo-1~1.0-SNAPSHOT

对此:

runmod org.smartjava~vertx-demo-1~1.0-SNAPSHOT -conf src/main/resources/config.json

config.json文件包含我们刚刚显示的配置。 因此,通过此设置,我们可以在事件总线地址“ mongodb-persistor”上监听mongodb-persistor库。 现在剩下要做的就是以该模块可以理解的格式将消息发送到此端点。 第一步,我们将搜索状态为“ AL”的所有邮政编码。 如果您浏览https://github.com/vert-x/mod-mongo-persistor/的文档,您会发现我们已经告诉该模块要搜索的“集合”以及“操作”的类型我们要使用。 根据操作,需要其他配置。 要搜索“ AL”状态下的所有邮政编码,我们需要创建以下json消息:

{"action": "find","collection": "zips","matcher": {"state": "AL"}
}

让我们更改请求处理程序并查看完整的start方法:

public void start() {// load the general config object, loaded by using -config on command lineJsonObject appConfig = container.config();// deploy the mongo-persistor module, which we'll use for persistencecontainer.deployModule("io.vertx~mod-mongo-persistor~2.1.0-SNAPSHOT", appConfig.getObject("mongo-persistor"));// create and run the serververtx.createHttpServer().requestHandler(new Handler<HttpServerRequest>() {@Overridepublic void handle(final HttpServerRequest httpServerRequest) {// we send the response from the mongo query back to the client.// first create the queryJsonObject matcher = new JsonObject().putString("state", "AL");JsonObject json = new JsonObject().putString("collection", "zips").putString("action", "find").putObject("matcher", matcher);// send it over the busvertx.eventBus().send("mongodb-persistor", json, new Handler<Message<JsonObject>>() {@Overridepublic void handle(Message<JsonObject> message) {// send the response back, encoded as stringhttpServerRequest.response().end(message.body().encodePrettily());}});}}).listen(8888);// output that the server is startedcontainer.logger().info("Webserver started, listening on port: 8888");}

在这里,您可以看到我们创建了正确的json消息,通过总线发送了该消息,然后等待将响应发送回,直到我们从mongoDB获得响应。 我们美化此响应并将其发送回客户端:

本地主机_8888-1

通过REST界面公开拉链

现在我们已经有了基本的后端组件,让我们看看创建一个简单的基于REST的前端需要什么。 我们将跳过特定于媒体类型的过滤(我将在后面的文章中添加),现在我们只看HTTP动词和URL。 对于这一部分,我们希望支持以下REST调用:

* GET /zipsShow all the zipcode information that are stored in mongoDB
* GET /zips/:idShow the information belonging to the specified zip code
* GET /zips?state=:state&city=:citySimple search service, where you can search for zip codes per city or state* POST /zips/:idUpdate existing zip code information

非常简单,但是这里的主要目标是显示其完成方式,而不是如何创建完整的RESTful服务。 为了处理这些各种URL和动词,vert.x提供了一个路由匹配器:(为清晰起见,省略了方法体)

RouteMatcher matcher = new RouteMatcher();// the matcher for the complete list and the searchmatcher.get("/zips", new Handler<HttpServerRequest>() {...}// the matcher for a specific idmatcher.get("/zips/:id", new Handler<HttpServerRequest>() {...}// the matcher for the updatematcher.post("/zips/:id", new Handler<HttpServerRequest>() {...}vertx.createHttpServer().requestHandler(matcher).listen(8888);

对于那些使用诸如sinatra或scalatra之类的库的人来说,这看起来很熟悉。 我们定义了要处理的方法(在这种情况下为获取和发布),我们感兴趣的url以及在收到请求时将被调用的处理程序。 如您在最后一行看到的那样,我们传入此处理程序来处理对我们创建的服务器的请求。

现在让我们快速看一下这些处理程序的实现。 在这里,我们创建了与mongoDB通信的mongo-persistor消息。 我不会过多介绍这些方法,因为它们几乎可以自我解释:

// the matcher for the complete list and the searchmatcher.get("/zips", new Handler<HttpServerRequest>() {public void handle(final HttpServerRequest req) {JsonObject json = new JsonObject();MultiMap params = req.params();if (params.size() > 0 && params.contains("state") || params.contains("city")) {// create the matcher configurationJsonObject matcher = new JsonObject();if (params.contains("state")) matcher.putString("state", params.get("state"));if (params.contains("city")) matcher.putString("city", params.get("city"));// create the message for the mongo-persistor verticlejson = new JsonObject().putString("collection", "zips").putString("action", "find").putObject("matcher", matcher);} else {// create the queryjson = new JsonObject().putString("collection", "zips").putString("action", "find").putObject("matcher", new JsonObject());}JsonObject data = new JsonObject();data.putArray("results", new JsonArray());// and call the event we want to usevertx.eventBus().send("mongodb-persistor", json, new ReplyHandler(req, data));}});

在这种方法中,我们从mongoDB中检索所有邮政编码。 由于mongo-persistor不会返回所有内容,因此我们需要对响应进行迭代。 我们使用以下ReplyHandler进行此操作:

private static class ReplyHandler implements Handler<Message<JsonObject>> {private final HttpServerRequest request;private JsonObject data;private ReplyHandler(final HttpServerRequest request, JsonObject data) {this.request = request;this.data = data;}@Overridepublic void handle(Message<JsonObject> event) {// if the response contains more message, we need to get the restif (event.body().getString("status").equals("more-exist")) {JsonArray results = event.body().getArray("results");for (Object el : results) {data.getArray("results").add(el);}event.reply(new JsonObject(), new ReplyHandler(request, data));} else {JsonArray results = event.body().getArray("results");for (Object el : results) {data.getArray("results").add(el);}request.response().putHeader("Content-Type", "application/json");request.response().end(data.encodePrettily());}}}

在这个ReplyHandler中,我们仅浏览结果并继续要求提供更多内容,直到不再看到“已存在”状态为止。 我将跳过仅检索单个邮政编码的处理程序,因为它并不那么有趣。 下一个处理程序处理post函数,通过该函数我们可以更新现有元素。

matcher.post("/zips/:id", new Handler<HttpServerRequest>() {public void handle(final HttpServerRequest req) {// process the bodyreq.bodyHandler(new Handler<Buffer>() {@Overridepublic void handle(Buffer event) {// normally we'd validate the input, for now just assume it is correct.final String body = event.getString(0,event.length());// create the queryJsonObject newObject = new JsonObject(body);JsonObject matcher = new JsonObject().putString("_id", req.params().get("id"));JsonObject json = new JsonObject().putString("collection", "zips").putString("action", "update").putObject("criteria", matcher).putBoolean("upsert", false).putBoolean("multi",false).putObject("objNew",newObject);// and call the event we want to usevertx.eventBus().send("mongodb-persistor", json, new Handler<Message<JsonObject>>() {@Overridepublic void handle(Message<JsonObject> event) {// we could handle the errors here, but for now// assume everything went ok, and return the original// and updated jsonreq.response().end(body);}});}});}});

代码本身并不复杂。 我们首先使用处理程序来处理请求,然后从该处理程序中创建一个新的处理程序,该处理程序用于获取请求的主体,最后创建一个处理程序,该处理程序更新数据库并发送回响应。 尽管它并不复杂,但是当涉及到越来越多的处理程序时,它会变得有些繁琐且难以阅读。 因此,在本文的最后一部分,我们将介绍如何使用rxjava提供的功能替换嵌套处理程序。

用RxJava观察者替换回调

对于rxjava代码,我们将只添加几个与不同URL匹配的额外处理程序。 因此,该网址将是/ rxzips / 90210,而不是/ zips / 90210。 首先,将以下依赖项添加到您的Maven配置中:

<dependency><groupId>io.vertx</groupId><artifactId>mod-rxjava</artifactId><version>1.0.0-beta2-SNAPSHOT</version><scope>compile</scope></dependency>

在深入探讨如何将rxjava和vert.x一起使用之前(从rxjava站点获取),快速引语解释了为什么这样做很有用:

Java Futures可直接用于单个级别的异步执行,但是嵌套时它们开始增加非同寻常的复杂性。

很难使用Future来最佳地组成条件异步执行流(或者是不可能的,因为每个请求的延迟在运行时会有所不同)。 当然可以这样做,但是很快就会变得复杂(因此容易出错),或者过早地在Future.get()上阻塞,这消除了异步执行的好处。

另一方面,RxJava Observables旨在组成异步数据的流和序列。

使用RxJava Observables,可以很容易地组成异步数据流和序列。 如果查看示例中的最后两个代码示例,您会发现为什么这样做会很有用。 在最后一个“ post”示例中,我们有三个嵌套的回调,通过可观察的函数,可以很容易地编写它并使代码实际告诉正在发生的事情。

现在,让我们扩展post方法来执行以下操作:

  1. 首先得到身体
  2. 拥有主体之后,我们将更新数据库中的元素
  3. 接下来,更新成功后,我们将从数据库中获取最新版本
  4. 在获得最新版本后,我们将在响应中返回此版本。

如果使用回调进行此操作,则可能需要四个嵌套级别的回调。 在rxjava中,我们可以按以下方式进行操作:

matcher.post("/rxzips/:id", new Handler<HttpServerRequest>() {public void handle(final HttpServerRequest req) {// first access the buffer as an observable. We do this this way, since// we want to keep using the matchhandler and we can't do that with rxHttpServerObservable<Buffer> reqDataObservable = RxSupport.toObservable(req);// after we have the body, we update the element in the databaseObservable<RxMessage<JsonObject>> updateObservable = reqDataObservable.flatMap(new Func1<Buffer, Observable<RxMessage<JsonObject>>>() {@Overridepublic Observable<RxMessage<JsonObject>> call(Buffer buffer) {System.out.println("buffer = " + buffer);// create the messageJsonObject newObject = new JsonObject(buffer.getString(0, buffer.length()));JsonObject matcher = new JsonObject().putString("_id", req.params().get("id"));JsonObject json = new JsonObject().putString("collection", "zips").putString("action", "update").putObject("criteria", matcher).putBoolean("upsert", false).putBoolean("multi", false).putObject("objNew", newObject);// and return an observablereturn rxEventBus.send("mongodb-persistor", json);}});// use the previous input again, so we could see whether the update was successful.Observable<RxMessage<JsonObject>> getLatestObservable = updateObservable.flatMap(new Func1<RxMessage<JsonObject>, Observable<RxMessage<JsonObject>>>() {@Overridepublic Observable<RxMessage<JsonObject>> call(RxMessage<JsonObject> jsonObjectRxMessage) {System.out.println("jsonObjectRxMessage = " + jsonObjectRxMessage);// next we get the latest version from the database, after the update has succeeded// this isn't dependent on the previous one. It just has to wait till the previous// one has updated the database, but we could check whether the previous one was successfullyJsonObject matcher = new JsonObject().putString("_id", req.params().get("id"));JsonObject json2 = new JsonObject().putString("collection", "zips").putString("action", "find").putObject("matcher", matcher);return rxEventBus.send("mongodb-persistor", json2);}});// after we've got the latest version we return this in the response.getLatestObservable.subscribe(new Action1<RxMessage<JsonObject>>() {@Overridepublic void call(RxMessage<JsonObject> jsonObjectRxMessage) {req.response().end(jsonObjectRxMessage.body().encodePrettily());}});}});

仍然是一大段代码,但是大多数都是注释,并且是由Java不支持闭包(实际上)引起的。 那么这里发生了什么?

  1. 我们首先根据请求创建一个观察者
    reqDataObservable = RxSupport.toObservable(req)

    这意味着我们希望在请求缓冲区中的数据可用时得到通知。

  2. 由于我们想对这些数据进行处理,因此我们使用
    reqDataObservable.flatMap

    功能。 这使我们可以指定当先前创建的可观察对象上有一些数据可用时会发生什么。 因此,我们没有嵌套回调,而是通过各种异步调用指定了数据流。 收到数据后,我们将使用它来更新数据库。 请注意,我们使用

    rxEventBus.send

    方法。 这还没有进行呼叫,但是再次返回了一个可观察的。

  3. 作为第三步,我们使用上一个的输出来(可能)确定更新是否成功。 然后,我们从数据库中获取最新版本。 再次使用可观察对象完成此操作。
  4. 只要我们没有订户,什么都不会发生。 由于我们对最终异步操作的结果感兴趣,因此我们使用
    getLatestObservable.subscribe

    函数并“等待”(仍然是非阻塞)以获取最终数据库读取的结果。 一旦收到,我们将根据收到的消息将响应发送回去。

在此示例中,我们仅展示了rxjava的几个非常小的部分:

  • 我们使用了可观察对象,使异步操作的工作和排序变得更加容易。
  • 我们使用flatmap函数将结果从一个异步调用传递到另一个异步调用,从而消除了嵌套回调
  • 我们使用rxSupport和rxEventbus rxJava vert.x扩展来轻松创建rxJava可观察对象
  • 然后,我们通过订阅链中的最后一个可观察到的序列开始了完整的序列

在以后的文章中,我们将对rxJava进行更深入的研究,以了解如何组织更复杂的异步流。

参考:从Smart Java博客的JCG合作伙伴 Jos Dirksen 使用vert.x 2.0,RxJava和mongoDB创建simpe RESTful服务 。

翻译自: https://www.javacodegeeks.com/2013/12/create-a-simpe-restful-service-with-vert-x-2-0-rxjava-and-mongodb.html

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/365652.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

搭建react native所遇到的坑

一、所遇问题 在搭建react native环境中,遇到执行react native run-android命令出现如下问题 1 Could not resolve all dependencies for configuration :classpath. 2 > Could not resolve com.github.dcendents:android-maven-gradle-plugin:3.1.4 3 Required b…

sqlalchemy mysql_使用SQLAlchemy操作MySQL

SQLAlchemy是Python编程语言下的一款开源软件&#xff0c;提供了SQL工具包及对象关系映射(ORM)工具&#xff0c;使用MIT许可证发行。SQLAlchemy首次发行于2006年2月&#xff0c;并迅速地在Python社区中最广泛使用的ORM工具之一&#xff0c;不亚于Django的ORM框架。本文将介绍如…

使用phpstorm+wamp实现php代码实时调试审计

转载自&#xff1a;https://www.bugbank.cn/q/article/5853afaffc0bf4f010ee6ac3.html php调试有N多好用的工具&#xff0c;最近研究到phpstorm配合wamp实现php实时断点调试相当不错&#xff0c;推荐给大家&#xff0c;网上有各种配置版本&#xff0c;但没有一个能好使的。本文…

如何使用Play框架为https配置SSL证书

我花了数小时试图使它起作用&#xff0c;最后&#xff0c;问题是我自己没有使用keytool生成CSR&#xff08;证书请求&#xff09;。 尝试通过https访问Play时&#xff0c;我一直收到此错误&#xff1a; javax.net.ssl.SSLPeerUnverifiedException&#xff1a;对等方未通过身份…

js页面加载前执行_做一名合格的前端开发工程师:Javascript加载执行问题探索

做前端开发少不了各种利器。比如我习惯用的还是Google浏览器和重型武器Fiddller。一&#xff1a;原始情况 首先大家看看如下的代码&#xff1a;估计90%的程序员都会把js文件放在head中&#xff0c;但是大家有没有深究过呢&#xff1f;很多浏览器都会使用单一的线程来做“界面UI…

【线性代数】3-5:独立性,基和维度(Independence,Basis and Dimension)

title: 【线性代数】3-5:独立性&#xff0c;基和维度(Independence,Basis and Dimension) categories: MathematicLinear Algebra keywords:IndependenceBasisDimensionSpan toc: true date: 2017-09-25 15:20:46Abstract: 本文是本章最重要的知识点&#xff0c;也是整个线性代…

APP网络测试要点和弱网模拟

当前APP网络环境比较复杂&#xff0c;网络制式有2G、3G、4G网络&#xff0c;还有越来越多的公共Wi-Fi。不同的网络环境和网络制式的差异&#xff0c;都会对用户使用app造成一定影响。另外&#xff0c;当前app使用场景多变&#xff0c;如进地铁、上公交、进电梯等&#xff0c;使…

word公式插件_再也不用担心我的公式写不出来了:一款公式输入神器实测

→ → https://mp.weixin.qq.com/s/gNRWdN-W4_Eb2MyyTpWzjw关注微信公众号&#xff1a;人工智能前沿讲习&#xff0c;重磅干货&#xff0c;第一时间送达本文推荐一个公式输入神器&#xff0c;只要截图就能识别公式&#xff0c;手写的公式都能识别。经过实测&#xff0c;几乎没有…

使用Camel在Amazon上构建分布式工作流应用程序

带有SNS-SQS的管道 工作流由按动态条件确定的特定顺序执行的独立任务组成。 工作流通常代表业务流程&#xff0c;例如电子商务商店中的订单处理步骤。 Amazon Web Services提供了各种工具来构建分布式和可扩展的工作流应用程序。 构建此类应用程序的一种方法是使用主题和队列来…

mysql 高版本检索外键_第05期:外键到底能不能用?

外键的设计初衷是为了在数据库端保证对逻辑上相关联的表数据在操作上的一致性与完整性。外键在大部分企业写的开发规范里会直接规避掉&#xff01;外键有优缺点&#xff0c;也并不是说每种场景都不适用&#xff0c;完全没有必要一刀切。外键到底能不能用&#xff1f;下面会针对…

从Ant Build演进Gradle Build:导入Ant Build文件

在大型项目上更改构建系统可能很困难并且需要大量工作。 幸运的是&#xff0c;对于那些将Ant版本迁移到Gradle版本的人&#xff0c;Gradle提供了特别方便的机制来促进这种迁移 。 由于Gradle基于Groovy构建&#xff0c;并且Groovy通过AntBuilder包含内置的Ant支持&#xff0c;因…

HTML5 之 简单汇总

参考&#xff1a; HTML5的十大新特性 前端面试必备之html5的新特性 HTML5 主要内容&#xff1a;语义化、增强型表单、多媒体标签、Canvas、SVG、地理定位、拖放API、Web Worker、Web Storage、WebSocket、HTML 5 应用程序缓存 1.语义化元素 1.1结构元素 标签描述article表示与上…

个人信息管理系统代码_Thymeleaf+SpringBoot+Mybatis实现的易游网旅游信息管理系统...

项目简介项目源码&#xff1a;麻烦转发后关注JAVA梦想口服液私信回复【源码】即可获取&#xff01;本系统是基于ThymeleafSpringBootMybatis。是非常标准的SSM三大框架(SpringBoot就是一个大框架&#xff0c;里面包含了许多的东西&#xff0c;其中Spring就是最核心的内容&#…

被问到有没有内核开发经验_一个人就是一个开发团队!成电硬核毕业生自制迷你电脑走红!...

△小视频近日&#xff0c;一段长3分多钟的小视频在B站突然爆红&#xff0c;不仅登上首页&#xff0c;获得超过200万的播放量&#xff0c;还被众多观众“膜拜”。有网友在评论区说&#xff1a;“其实这些东西吧&#xff0c;外行看起来很牛&#xff0c;但我这种内行看起来&#x…

matlab2010a连接mysql_MATLAB2010a+OpenCV2.3.1+VS2010运行TLD

出现matlab不显示C编译器的原因主要还是当前Matlab版本相对于VS来说不够新&#xff0c;比如14版的肯定支持10的VS。 本文引用地址&#xff1a; http://blog.csdn.net/shanpohe/article/details/7596401 http://blog.sina.com.cn/s/blog_adfd55190101ejvr.html TLD(Tracking Lea…

更改span标签样式_CSS 内嵌样式

前面一节我们讲了行内样式&#xff0c;但是行内样式的缺点就是样式不能重用。例如当有好多个 标签&#xff0c;我们希望所有的 标签的样式都一致&#xff0c;那么需要在每个标签中都写一遍&#xff0c;这么会很麻烦&#xff0c;也会增加很多代码。那么为了解决这个问题&#…

mysql 6安装当前密码_MySQL8.0 安装踩坑指南

就在昨天上午&#xff0c;刚为云服务器安装好Apache2.4.33和PHP7.2.4环境&#xff0c;准备再来一个最新的MySQL5.7.22。寻找5.7版本的rpm包时下到mysql80xxx.rpm&#xff0c;看人家的教程是mysql57&#xff0c;难道80是MySQL出出…出了8版&#xff0c;一搜新闻2个小时前MySQL发…

Elasticsearch环境搭建和介绍(Windows)

一、Elasticsearch介绍和安装 1.1 介绍 Elastic Elastic官网&#xff1a;https://www.elastic.co/cn/ Elastic有一条完整的产品线&#xff1a;Elasticsearch、Kibana、Logstash等&#xff0c;前面说的三个就是大家常说的ELK技术栈。 Elasticsearch Elasticsearch官网&#xff1…

Jar Hell变得轻松–用jHades揭秘classpath

Java开发人员将不得不面对的最困难的问题是类路径错误&#xff1a; ClassNotFoundException &#xff0c; NoClassDefFoundError &#xff0c;Jar Hell&#xff0c; Xerces Hell和company。 在本文中&#xff0c;我们将探究这些问题的根本原因&#xff0c;并了解最小的工具&am…

程序控制发送文件到邮箱_Kindle电子邮箱推送

Kindle 推送支持的格式目前 Kindle 推送仅支持以下几种格式。需要注意的是&#xff0c;AZW 和 AZW3 是两种不同的格式&#xff0c;虽然这两种格式 Kindle 设备都支持阅读&#xff0c;但是亚马逊的个人文档服务支持推送 AZW 但是不支持 AZW3。Kindle 格式 (.mobi 或 .azw) * 推荐…