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方法来执行以下操作：

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

如果使用回调进行此操作，则可能需要四个嵌套级别的回调。在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不支持闭包（尚未）这一事实引起的。那么这里发生了什么？

我们首先根据请求创建一个观察者
```
reqDataObservable = RxSupport.toObservable(req)
```
这意味着我们希望在请求缓冲区中的数据可用时得到通知。
由于我们想对这些数据进行处理，因此我们使用
```
reqDataObservable.flatMap
```
功能。这使我们可以指定在先前创建的可观察对象上有可用数据时发生的情况。因此，我们没有嵌套回调，而是通过各种异步调用指定了数据流。收到数据后，我们将使用它来更新数据库。请注意，我们使用
```
rxEventBus.send
```
方法。这还没有进行呼叫，但是再次返回了一个可观察的。
作为第三步，我们使用上一个的输出来（可能）确定更新是否成功。然后，我们从数据库中获取最新版本。再次使用可观察对象完成此操作。
只要我们没有订户，什么都不会发生。由于我们对最终异步操作的结果感兴趣，因此我们使用
```
getLatestObservable.subscribe
```
函数并“等待”（仍然是非阻塞的）以获取最终数据库读取的结果。一旦收到，我们将根据收到的消息将响应发送回去。

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