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

vert.x

中断了将近半年后发表了一篇新文章。 在本文中,我们将快速了解如何开始使用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

vert.x

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

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

相关文章

如何通过Rultor将Maven工件部署到CloudRepo

在我以前的文章中 &#xff0c;我描述了如何在Amazon S3中设置私有Maven存储库并通过Rultor进行部署。 如果您熟悉管理Amazon Web Services&#xff08;AWS&#xff09;&#xff0c; S3和AWS Identity and Access Management&#xff08;IAM&#xff09;的话&#xff0c;这是一…

java里面自行车的属性_11、Java基础知识

1、安装jdk&#xff0c;配置环境变量2、public class HelloWorld{publicstatic void main(String[] args){System.out.println(‘HelloWorld’);}}3、编译过程&#xff1a;通过javac编译java文件&#xff0c;生成.class文件&#xff0c;使用java命令运行class文件&#xff0c;注…

布线问题分支限界法java_大型布线:Java云应用程序缺少的技术

布线问题分支限界法java您是否曾经想过&#xff0c;为什么大多数Java框架中的依赖项注入仅用于本地进程内服务而不是分布式服务&#xff1f; 我最近在2013年EMC世界大会上遇到了Paul Maritz的主题演讲 &#xff08;跳至第32分钟&#xff09;&#xff0c;这使我在云平台的背景下…

Spring Boot微服务,Docker和Kubernetes研讨会–第2部分

在上一篇文章中&#xff0c;我们使用SpringBoot和Docker创建了第一个微服务“ ProductService”。 在这一部分中&#xff0c;我们将详细介绍如何使用Spring Cloud&#xff0c;netflix库&#xff0c;API网关来管理多个微服务。 假设对于我们的订单管理系统&#xff0c;最小关系…

jboss5.1安全性配置_使用Java EE安全性和JBoss AS 7.x保护JSF应用程序的安全

jboss5.1安全性配置企业应用程序的一个常见要求是在登录页面后面保护所有JSF页面。 有时&#xff0c;您甚至希望在应用程序内部具有保护区&#xff0c;只有拥有特定角色的用户才能访问这些保护区。 Java EE标准附带了实现受某些安全性约束保护的Web应用程序所需的所有方法。 在…

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

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

比较Java REST文档框架

确定在记录REST API时选择哪种Java框架可能很麻烦。 在本博文中&#xff0c;我们将简要比较我们自己使用的REST Web服务的三种文档框架&#xff0c;以及它们如何与Spring框架&#xff08;这是Foreach最常使用的Java框架&#xff09;集成。 这些是RESTful API建模语言&#xff0…

jaVa游戏三国志英杰传,《三国志英杰传》到底是怎样的一款游戏

原标题&#xff1a;《三国志英杰传》到底是怎样的一款游戏介绍作为PC平台上经典的战棋策略类游戏&#xff0c;英杰传系列可谓把这一类型游戏在战略性和资源获取上的精髓发挥的淋漓尽致。系列初代的《三国志英杰传》诞生在1995年的DOS系统上&#xff0c;虽然我接触英杰传时已经是…

jvm 内存镜像_镜像镜像–使用反射在运行时查看JVM内部

jvm 内存镜像开发人员&#xff1a;Takipi会告诉您何时新代码在生产中中断– 了解更多 我们都习惯于在我们的日常工作中直接或通过利用反射的框架来运用反射。 它是Java和Scala编程的主要方面&#xff0c;它使我们使用的库可以与我们的代码进行交互&#xff0c;而无需对其进行硬…

谁去过顽皮,谁去过尼斯? 圣诞老人为您提供Java 11建议!

有没有想过圣诞老人如何为世界各地的孩子们送上节日礼物&#xff1f; 有20亿个孩子&#xff0c;每个孩子都有自己的愿望清单&#xff0c;他会在24小时内完成。 这意味着每个孩子平均需要43微秒&#xff0c;他需要检查每个孩子是否顽皮或好。 您无需再怀疑了。 我会透露这个秘密…

php时间格式函数,PHP函数之日期时间函数date()使用详解_php基础_脚本

$ttime();echo date("Y-m-d H:i:s",$t);第一个参数的格式分别表示:a - "am" 或是 "pm"A - "AM" 或是 "PM"d - 几日&#xff0c;二位数字&#xff0c;若不足二位则前面补零; 如: "01" 至 "31"D - 星期几…

play框架配置 拦截器_如何使用Play框架为https配置SSL证书

play框架配置 拦截器我花了几个小时试图使它起作用&#xff0c;最后&#xff0c;问题是我自己没有使用keytool生成CSR&#xff08;证书请求&#xff09;。 当我尝试通过https访问Play时&#xff0c;我一直收到此错误&#xff1a; javax.net.ssl.SSLPeerUnverifiedException&a…

matlab 球坐标绘图,MATLAB绘制地图

1使用向量绘制地图1.1绘制全球海岸线向量数据可以表示一个地图。这种向量存在的形式是一系列的经纬度或投影坐标对&#xff0c;它们代表一个点集、一个线条或者多边形。例如&#xff0c;描绘出行政区域边界的点、公路系统、城市的中心或者以上三个集合放在一起&#xff0c;都可…

php 有 stringbuffer,String、StringBuffer、StringBulider三者介绍

三者都实现了CharSequence接口&#xff0c;因此CharSequence可认为是一个字符串的协议接口1.String类是不可变类&#xff0c;即一旦一个String对象被创建后&#xff0c;包含在这个对象中的字符序列是不可改变的&#xff0c;直至这个对象被销毁&#xff1b;我们常常定义的时候 S…

php生成网页按钮,JavaScript实现自动生成网页元素功能(按钮、文本等)_javascript技巧...

创建元素的方法&#xff1a;1、利用createTextNode()创建一个文本对象2、利用createElement()创建一个标签对象3、直接利用容器标签中的一个属性&#xff1a;innerHTML-----本质上改该标签容器中的“html代码”&#xff0c;不是我们认为的对象树的操作详解代码&#xff1a;这是…

adf 自动输稿器_在ADF实体PK属性中使用MySQL自动增量PK列

adf 自动输稿器大家好。 继续进行ADF MySQL解决方法系列&#xff0c;今天我们将看到需要做些什么才能将MySQL PK自动增量列与ADF实体PK属性一起使用。 如果使用的是Oracle数据库&#xff0c;则可以使用oracle.jbo.domain.DBSequence以及序列和触发器来立即进行操作。 为简单起…

探索适用于Apache Spark的Spline Data Tracker和可视化工具(第1部分)

最近引起我注意的一个有趣且充满希望的开源项目是Spline &#xff0c;它是由Absa维护的Apache Spark的数据沿袭跟踪和可视化工具。 该项目由两部分组成&#xff1a;一个在驱动程序上工作的Scala库&#xff0c;该驱动程序通过分析Spark执行计划来捕获数据沿袭&#xff0c;并提供…

高性能mysql 聚簇索引,高性能MySQL笔记-第5章Indexing for High Performance-005聚集索引...

一、聚集索引介绍1.什么是聚集索引&#xff1f;InnoDB’s clustered indexes actually store a B-Tree index and the rows together in the same structure.2.为什么一张表只能一个聚集索引&#xff1f;When a table has a clustered index, its rows are actually stored in …

PHP应用GD2函数填充几何图形,使用GD2函数绘制几何图形(PHP图形图像的典型应用教程4)...

使用GD2函数绘制几何图形(PHP图形图像的典型应用教程4)本篇主要讲解使用GD2函数实现几何图形的绘制&#xff0c;首先我们需要的事创建一个图像&#xff0c;在之前的文章中我们就说过了&#xff0c;创建图像是所有图像操作的第一步&#xff0c;然后再背景上根据坐标点绘制图形轮…

API测试和自动化101:基本指南

API代表A pplication P AGC软件我覆盖整个院落。 通常&#xff0c;API用于通过使用任何通信方式来促进两个不同应用程序之间的交互。 在网络上使用API​​时&#xff0c;我们将其称为“ Web服务”。 最近&#xff0c;API已成为编程的Struts。 与在应用程序中一样&#xff0c;编…