在这一章节里，我们将用Vert.X构建我们的第一个微服务。像大多数采用http交互的微服务系统，我们打算从http微服务开始。因为系统由多个互相通信的微服务构成，我们将构建另一个微服务，它作为第一个微服务的消费者。然后，我们将展示为何这样的设计并不完全符合响应式微服务。最后，我们将实现基于消息的微服务，看看消息是怎样提升了响应性。
第一个微服务
在这一章节，我们打算实现同类的微服务两个。第一个微服务暴露一个hello服务，我们称它为hello微服务。另一个消费这个服务两次(并发地)。消费者将被称为hello消费者微服务。这个小系统不仅展示了一个服务是怎样提供服务的，而且展示它是怎样被消费的。在图3-1的左边，微服务用http交互，hello消费者微服务作为http客户端向hello微服务发请求；在图的右边，hello消费者微服务用消息与hello微服务交互。这个不同影响了系统的响应性。
图3-1
在前一章节，我们看到两种不同的方式使用Vert.X API: 回调和RxJava。展示它们的不同有助于你发现更佳途径，hello微服务是使用基于回调开发模式实现，而hello消费者微服务是用RxJava实现。
实现http微服务
微服务通常通过http暴露他们的API，通过http请求来消费。让我们看看用Vert.X怎样实现这些http交互。这个部分开发的代码在代码仓库的microservices/hello-microservice-http目录下可获得。
开始
创建hello-microservice-http目录，然后生成工程结构：
mkdir hello-microservice-http
cd hello-microservice-http
mvn io.fabric8:vertx-maven-plugin:1.0.5:setup \
-DprojectGroupId=io.vertx.microservice \
-DprojectArtifactId=hello-microservice-http \
-Dverticle=io.vertx.book.http.HelloMicroservice \
-Ddependencies=web
这个命令生成maven工程，配置Vert.X Maven插件。另外，它加上vertx-web依赖。Vert.X Web是一个模块，它提供你基于Vert.X构建流行的web应用的一切。
Verticle
打开src/main/java/io/vertx/book/http/HelloMicroservice.java，这个被生成的verticle代码没做什么很有趣的事，但它是一个起点：
package io.vertx.book.http;
import io.vertx.core.AbstractVerticle;
public class HelloMicroservice extendsAbstractVerticle {
@Override
publicvoid start() {
}
}
现在，执行下面的maven命令：
mvn compile vertx:run
你现在可以编辑verticle，每次你保存文件后，应用将被重新编译并自动重启。
http微服务
是时候让MyVerticle做点什么了。让我们启动一个http server。正如你前面章节看到的，用Vert.X创建一个http server仅仅：
@Override
public void start() {
vertx.createHttpServer()
.requestHandler(req-> req.response().end("hello"))
.listen(8080);
}
一旦加上这些代码并保存，在浏览上访问http://localhost:8080你应该能看到hello。这段代码创建一个http server监听端口8080，注册了一个请求处理器，每一个http请求进来时它会被调用。现在，我们仅仅输出hello到http响应。
使用路由和参数
许多服务是通过web url调用的，因此，检查路径是重要的，以知道请求在要求什么。然而，在请求处理器里面做路径检查以实现不同的动作可能会变得复杂。幸运地，Vert.X Web提供了一个路由器，通过它你可以注册路由。路由是Vert.X Web检查路径、调用相关动作的机制。让我们重写start方法，用两个路由：
@Override
public void start() {
Routerrouter = Router.router(vertx);
router.get("/").handler(rc-> rc.response().end("hello"));
router.get("/:name").handler(rc-> rc.response().end("hello " + rc.pathParam("name")));
vertx.createHttpServer()
.requestHandler(router::accept)
.listen(8080);
}
我们创建了路由器对象后，我们注册了两个路由：第一个处理根路径的请求仅仅输出hello，第二个路由有一个路径参数(:name)，处理器追加参数值到欢迎中。最后，我们更改请求处理器(requestHandler)，使用路由器的accept方法。
如果你没有停止vertx:run，你打开浏览器：
访问http://localhost:8080，你应该会看到hello
访问http://localhost:8080/vert.x，你应该会看到hello vert.x
生成JSON
在微服务里，JSON是常用的。让我们修改前一个类，生成JSON:
@Override
public void start() {
Routerrouter = Router.router(vertx);
router.get("/").handler(this::hello);
router.get("/:name").handler(this::hello);
vertx.createHttpServer()
.requestHandler(router::accept)
.listen(8080);
}
private void hello(RoutingContext rc) {
Stringmessage = "hello";
if(rc.pathParam("name") != null) {
message+= " " + rc.pathParam("name");
}
JsonObjectjson = new JsonObject().put("message", message);
rc.response()
.putHeader(HttpHeaders.CONTENT_TYPE,"application/json")
.end(json.encode());
}
Vert.X提供一个JsonObject类来创建和操作JSON。放上这段代码，你打开浏览器：
访问http://localhost:8080，你应该会看到{“message”:“hello”}
访问http://localhost:8080/vert.x，你应该会看到{“message”: “hello vert.x”}
打包和运行
按CTRL+C，停止vertx:run的执行，在同一目录下执行下面的命令：
mvn package
这生成一个fat jar在target目录下：hellomicroservice-http-1.0-SNAPSHOT.jar。fat jar之所以胖，因为jar包有一个合理的大小(约6.3MB)，包含了运行应用所需的一切：
java -jar target/hello-microservice-http-1.0-SNAPSHOT.jar
你可以通过访问http://localhost:8080来检查确定它是运行起来的。保持住运行，因为下一个微服务将调用它。
消费http微服务
一个微服务不构成一个应用，你需要一个微服务系统。现在我们有了一个运行中的微服务，让我们写第二个微服务来消费它。第二个微服务也提供了一个http请求接口，每一个请求会调用我们刚刚实现的微服务。这个章节展示的代码可以从代码仓库的microservices/helloconsumer-microservice-http目录获得。
创建工程
一样地，让我们创建一个新工程：
mkdir hello-consumer-microservice-http
cd hello-consumer-microservice-http
mvn io.fabric8:vertx-maven-plugin:1.0.5:setup
\
-DprojectGroupId=io.vertx.microservice \
-DprojectArtifactId=hello-consumer-microservice-http \
-Dverticle=io.vertx.book.http.HelloConsumerMicroservice \
-Ddependencies=web,web-client,rx
最后的命令增加了其它的依赖：Vert.X web客户端，异步的http客户端。我们将使用这个客户端来向第一个微服务发请求。这个命令也增加的Vert.X RxJava绑定，我们打算在后面使用它。
现在编辑src/main/java/io/vertx/book/http/HelloConsumerMicroservice.java文件，更改它的内容：
package io.vertx.book.http;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.*;
import io.vertx.ext.web.client.*;
import io.vertx.ext.web.codec.BodyCodec;
public class HelloConsumerMicroservice extends AbstractVerticle {
private WebClientclient;
@Override
public void start(){
client = WebClient.create(vertx);
Routerrouter = Router.router(vertx);
router.get("/").handler(this::invokeMyFirstMicroservice);
vertx.createHttpServer()
.requestHandler(router::accept)
.listen(8081);
}
private voidinvokeMyFirstMicroservice(RoutingContext rc) {
HttpRequestrequest = client
.get(8080,"localhost","/vert.x")
.as(BodyCodec.jsonObject());
request.send(ar-> {
if(ar.failed()) {
rc.fail(ar.cause());
} else {
rc.response().end(ar.result().body().encode());
}
});
}
}
在start方法，我们创建一个WebClient，一个Router，我们注册了一个根路径的route，启动http server，传递router的accept方法给requestHandler。这个方法用web客户端来调用第一个微服务的指定路径(/vert.x)，输出结果到http响应。
一旦http请求被创建，我们调用send方法来发出请求，无论是响应返回或者是有错误发生，我们设定的处理器会被调用。If-else块检查请求成功与否。不要忘了这是一个远程交互，有很多原因导致失败。例如，第一个微服务可能没有运行。当它成功时，我们输入收到的数据到响应，否则，我们返回一个500的http响应。
多次调用服务
现在让我们改变当前的行动，用两个不同的路径参数请求hello微服务两次：
HttpRequest request1 = client
.get(8080, "localhost", "/Luke")
.as(BodyCodec.jsonObject());
HttpRequest request2 = client
.get(8080, "localhost", "/Leia")
.as(BodyCodec.jsonObject());
这两个请求是独立的，能够并发地执行。可是这里我们想输出一个把这两个请求的结果装配起来的响应。需要调用两次服务、把两个结果装配起来的代码可以变得复杂。当我们接收到其中一个响应时，我们需要检查另一个请求完成与否。当然，对于两个请求，这个代码仍然是可管理的。当我们需要处理更多的时候，它变得极其复杂。幸运地，正如前一章节所讲，我们能够使用响应式编程，RxJava使代码变得简单。
我们介绍了vertx-mavan-plugin插件引入Vert.X RxJava API。在HelloConsumerMicroservice里，我们替换重要的import语句：
import io.vertx.core.json.JsonObject;
import io.vertx.rxjava.core.AbstractVerticle;
import io.vertx.rxjava.ext.web.*;
import io.vertx.rxjava.ext.web.client.*;
import io.vertx.rxjava.ext.web.codec.BodyCodec;
import rx.Single;
用RX, 我们要写的调用两个请求、构造他们的结果成一个响应的复杂代码变得比较简单：
private voidinvokeMyFirstMicroservice(RoutingContext rc) {
HttpRequestrequest1 = client
.get(8080,"localhost", "/Luke")
.as(BodyCodec.jsonObject());
HttpRequestrequest2 = client
.get(8080,"localhost", "/Leia")
.as(BodyCodec.jsonObject());
Singles1 = request1.rxSend().map(HttpResponse::body);
Singles2 = request2.rxSend().map(HttpResponse::body);
Single.zip(s1,s2, (luke, leia) -> {
//We have the results of both requests in Luke and Leia
returnnew JsonObject()
.put("Luke",luke.getString("message"))
.put("Leia",leia.getString("message"));
})
.subscribe(
result-> rc.response().end(result.encodePrettily()),
error-> {
error.printStackTrace();
rc.response()
.setStatusCode(500).end(error.getMessage());
}
);
}
注意rxSend方法调用。Vert.X里，RxJava方法加上了rx前缀，以更容易识别。rxSend方法的结果是一个Single，可订阅的单个元素，代表一个操作的延期结果，single.zip方法用一组Single作为参数，一旦所有的Single收到它们的值，就用这些值来调用一个函数。最后，订阅(subscribe)，这个方法用两个函数作为参数：
第1个函数是用zip函数的结果(一个json对象)作为参数被调用，我们输出接收到的json内容到http响应；
第2个函数是某种失败(超时，异常等等)发生时被调用，在这里，我们用空的json对象做出响应。
这段代码生效后，hello微服务仍在运行，如果我们打开http://localhost:8081我们应该会看到：
{
"Luke" : "hello Luke",
"Leia" : "hello Leia"
}
这是响应式微服务吗？
现在我们有两个微服务。他们是可独立部署和修改的。他们也使用轻量级的http协议交互。但是他们是响应式微服务吗？不，他们不是。记住，响应式微服务必须是：
. 自治的
. 异步的
. 可恢复的
. 弹性的
当前设计的主要问题是两个微服务是紧耦合的。Web客户端是显示地配置了第一个微服务的地址。如果第一个微服务失败了，我们不能够通过请求另一个来恢复服务。我们想降低负载，创建一个新的hello微服务实例将帮助不了我们。感谢Vert.X web客户端，交互是异步的。然而，我们没有用一个虚拟的目标地址来调用微服务、而是直接用它的URL，这不能提供我们需要的可恢复性和弹性。
不要失望，在下一章节我们将朝响应式微服务迈出一大步。
Vert.X事件总线---一个消息后端
Vert.X提供了事件总线，允许一个应用的不同组件用消息来交互。消息被送到地址，有消息头和消息体。一个地址是一个字符串，代表一个目标地址。消息消费者注册它们自己到这个地址、以接收消息。事件总线也是集群的，意味着它能够跨越网络、在分布的发送者和消费者之间传递消息。以集群模式启动Vert.X应用，被连接的节点可以共享数据结构、做停止失败检查、负载均衡。事件总线能够在集群的所有节点间传递消息。为了创建这样一个集群，你可以用Apache Ignite、Apache Zookeeper、Infinispan或者是Hazelcast。在这本书里，我们打算用Infinispan，但是，我们不做高级的配置。如果需要，可参考Infinispan文档(http://infnispan.org/)。Infinispan(或者你选的其他技术)管理节点的发现和存储。事件总线用直接的p2p
tcp连接通讯。
事件总线提供了三种传递语法：第一种，send方法允许一个组件送一个消息到一个地址，
单个消费者将接收它。假如不止一个消费者注册到这个地址，Vert.X将采用轮询策略来选择一个消费者：
// Consumer
vertx.eventBus().consumer("address",message -> {
System.out.println("Received:'" + message.body() + "'");
});
// Sender
vertx.eventBus().send("address","hello");
与send不同，你可以用publish方法传递消息给所有注册在这个地址的消费者。
最后，send方法能够带一个应答处理器，这个请求/响应机制允许两个组件间实现基于消息的异步交互:
// Consumer
vertx.eventBus().consumer("address",message -> {
message.reply("pong");
});
// Sender
vertx.eventBus().send("address","ping", reply -> {
if(reply.succeeded()) {
System.out.println("Received:" + reply.result().body());
}else {
//No reply or failure
reply.cause().printStackTrace();
}
});
如果你用RX API，你能够用rxSend方法，它返回一个Single，当应用被收到时这个Single接收一个值。我们将很快看到这个方法。
基于消息的微服务
让我们重新实现hello微服务，这次用事件总线代替http server来接收请求。微服务应答消息、提供响应。
创建工程
让我们创建一个新工程。这次我们将加上Infnispan依赖，一个内存数据网格，被用来管理集群：
mkdir hello-microservice-message
cd hello-microservice-message
mvn io.fabric8:vertx-maven-plugin:1.0.5:setup
\
-DprojectGroupId=io.vertx.microservice \
-DprojectArtifactId=hello-microservice-message \
-Dverticle=io.vertx.book.message.HelloMicroservice \
-Ddependencies=infinispan
一旦生成后，为了构建集群，我们需要配置Infinispan。缺省的配置是用组播的方式来发现节点。
如果你的网络支持组播，就可以。否则，检查代码仓库的resource/cluster目录。
写消息驱动的Verticle
编辑src/main/java/io/vertx/book/message/HelloMicroservice.java文件，修改start方法：
@Override
public void start() {
// Receive messagefrom the address 'hello'
vertx.eventBus().consumer("hello",message -> {
JsonObject json =new JsonObject()
.put("served-by",this.toString());
// Check whether wehave received a payload in the
// incoming message
if (message.body().isEmpty()){
message.reply(json.put("message","hello"));
} else {
message.reply(json.put("message","hello" + message.body()));
}
});
}
这段代码从vertx对象获取事件总线(eventBus)，注册一个消费者到地址hello。当接收到一个消息时，它应答它。取决于进来的消息是否有一个空的消息体，我们给以不同的响应。像前面章节的例子一样，我们返回一个json对象。你可能想知道为什么我们在json里面加了served-by。你很快就会明白为什么。现在verticle写好了，是时候启动它：
mvn compile vertx:run \
-Dvertx.runArgs="-cluster -Djava.net.preferIPv4Stack=true"
-cluster选项告诉Vert.X以集群模式启动。
现在让我们写一个微服务来消费这个服务。
初始化基于消息的交互
在这一节里，我们将创建另一个微服务来调用hello微服务，通过送一个消息到hello地址并获得应答。微服务将重新实现与前一章节同样的逻辑，调用服务两次。
同样地，让我们创建一个新的工程：
mkdir hello-consumer-microservice-message
cd hello-consumer-microservice-message
mvn io.fabric8:vertx-maven-plugin:1.0.5:setup
\
-DprojectGroupId=io.vertx.microservice \
-DprojectArtifactId=hello-consumer-microservice-message \
-Dverticle=io.vertx.book.message.HelloConsumerMicroservice \
-Ddependencies=infinispan,rx
这里我们也加了Vert.X RxJava以便于使用Vert.X提供的RX API。如果在前一节中你更改了Infinispan的配置，你需要拷贝它到这个新工程。
现在编辑io.vertx.book.message.HelloConsumerMicroservice。因为我们打算用RxJava，改变相应的引入语句为io.vertx.rxjava.core.AbstractVerticle，然后实现start方法：
@Override
public void start() {
EventBus bus =vertx.eventBus();
Singleobs1 = bus
.rxSend("hello","Luke")
.map(Message::body);
Singleobs2 = bus
.rxSend("hello","Leia")
.map(Message::body);
Single.zip(obs1,obs2, (luke, leia) ->
newJsonObject()
.put("Luke",luke.getString("message"))
.put("Leia",leia.getString("message"))
)
subscribe(x ->System.out.println(x.encode()), Throwable::printStackTrace);
}
这段代码与前一章节是很类似的，替代用WebClient请求http，我们用事件总线发送消息到hello地址、获取应答内容。我们用zip操作获得两个响应并且构建最终的结果。在subscribe方法，我们打印最终结果到控制台或者是输出异常堆栈。
让我们把这段代码和http server合并在一起，当接收到http请求时，我们调用hello服务两次、把构建结果作为响应返回：
@Override
public void start() {
vertx.createHttpServer()
.requestHandler(req-> {
EventBus bus =vertx.eventBus();
Singleobs1 = bus
.rxSend("hello","Luke")
.map(Message::body);
Singleobs2 = bus
.rxSend("hello","Leia")
.map(Message::body);
Single.zip(obs1,obs2, (luke, leia) ->
newJsonObject()
.put("Luke",luke.getString("message") + " from " +luke.getString("served-by"))
.put("Leia",leia.getString("message") + " from " +leia.getString("served-by"))
)
.subscribe(
x ->req.response().end(x.encodePrettily()),
t -> {
t.printStackTrace();
req.response().setStatusCode(500).end(t.getMessage());
}
);
})
.listen(8082);
这段代码仅仅是打包事件总线的交互到请求处理器(requestHandler)并且处理http响应。在失败的情况下，我们返回一个包含错误信息的json对象。
如果你运行这段代码用
mvn compile vertx:run -Dvertx.runArgs="-cluster-Djava.net.preferIPv4Stack=true"
打开你的浏览器访问http://localhost:8082，你应该会看到像这样：
{
"Luke" : "hello Luke from ...HelloMicroservice@39721ab",
"Leia" : "hello Leia from ...HelloMicroservice@39721ab"
}
现在是响应式吗？
这段代码与我们前面写的基于http的微服务很类似，唯一的不同是我们用事件总线代替http。这改变了响应性？的确是，让我们看看为什么。
弹性
弹性是http版本的微服务没有的一个特性。因为微服务是被定位到一个指定的微服务实例(用硬编码的URL)，它没有提供我们需要的弹性。但是我们现在采用送到一个地址的消息，这改变了游戏。让我们看看这个微服务系统的表现。
记得前面执行的输出。返回的json对象显示verticle有处理hello消息。输出总是显示是同一个verticle。这个信息表明是同一个实例。我们预计这是因为我们只有一个实例在运行。现在让我们看看用两个实例将发生什么。
停止hello微服务的vertx:run，运行：
mvn clean package
然后，打开两个不同的终端，在hello-microservice-message目录里执行下面的命令：
java -jar target/hello-microservice-message-1.0-SNAPSHOT.jar \
--cluster -Djava.net.preferIPv4Stack=true
这将启动两个Hello微服务实例，返回到你的浏览器刷新页面你应该看到类似这样：
{
"Luke" : "hello Luke from...HelloMicroservice@16d0d069",
"Leia" : "hello Leia from...HelloMicroservice@411fc4f"
}
两个Hello实例被调用。Vert.X集群连接不同的节点，事件总线也是被集群的。感谢事件总线轮循，Vert.X事件总线分发消息到可用的实例、在监听同一地址的不同节点间均衡负载。
因此，通过用事件总线，我们有了我们所需要的弹性特征。
可恢复性
可恢复性又如何呢？在当前的代码中，如果hello微服务失败了，我们将得到一个失败、执行这个代码：
t -> {
t.printStackTrace();
req.response().setStatusCode(500).end(t.getMessage());
}
尽管用户得到了错误信息，我们没有崩溃，我们没有限制伸缩性，仍然能够处理请求。然而，为了提升用户体验，我们应该总是在适当的时间内响应，即使我们没有从服务中接收到响应。实现这个逻辑，我们可以用timeout增加代码。
为了展示，让我们修改hello微服务、注入失败。这段代码放在代码仓库的microservices/hello-microservice-faulty目录下。
这个新的start方法随机地选择3个策略中的一个：1. 用一个显示的失败响应，2.忘了响应，3.发送正确的结果：
@Override
public void start() {
vertx.eventBus().consumer("hello", message-> {
double chaos = Math.random();
JsonObject json = new JsonObject().put("served-by",this.toString());
if (chaos < 0.6) {
// Normal behavior
if (message.body().isEmpty()) {
message.reply(json.put("message", "hello"));
} else {
message.reply(json.put("message", "hello "+message.body()));
}
} else if (chaos < 0.9) {
System.out.println("Returning a failure");
// Reply with a failure
message.fail(500, "message processing failure");
} else {
System.out.println("Not replying");
// Just do not reply, leading to a timeout on the
// consumer side.
}
});
}
重新打包并重启两个hello微服务的实例。
使用这个故障注入的服务，我们需要改进消费方的容错性。事实上，消费方可能得到超时或是接收到一个显示的失败。在hello消费者微服务里，改变请求hello服务的代码为：
EventBus bus = vertx.eventBus();
Single obs1 = bus
.rxSend("hello", "Luke")
.subscribeOn(RxHelper.scheduler(vertx))
.timeout(3, TimeUnit.SECONDS)
.retry()
.map(Message::body);
Single obs2 = bus.
rxSend("hello", "Leia")
.subscribeOn(RxHelper.scheduler(vertx))
.timeout(3, TimeUnit.SECONDS)
.retry()
.map(Message::body);
这段代码放在代码仓库的microservices/hello-consumer-microservice-timeout目录下。如果有给定的时间内没有接收到响应，timeout方法发出一个失败。如果得到一个超时失败或是一个显示的失败，retry方法将试图重试去获取值。subScribeOn方法指明请求需要在哪一个线程上执行。我们用Vert.X事件轮循器来调用callback。没有指定的话，方法将被从缺省的RxJava线程池中取一个线程来执行，破坏了Vert.X的线程模式。RxHelper类是Vert.X提供的。盲目地重试服务调用不是明智的容错策略，它甚至可能是有害的。下一章节阐述不同的方法。
现在你可以重新加载页面。你总能获得一个结果，即使是失败或者超时。记住当调用服务时线程是不阻塞的，因此，你总是能够接收新的请求、在一个合适的时间内响应。然而，超时重试经常是有害而不是有益，正如我们在下一章节将看到的那样。
小结
在这一章节，我们学习了怎样用Vert.X开发一个http微服务，怎样消费它。正如我们所学的，在代码里硬编码被消费服务的URL不是一个明智的主意，因为它破坏了响应式特征之一。在第二部分，我们用消息替换http交互，这展示了消息和Vert.X事件总线怎样构建响应式微服务。
那么，现在是yes还是no。是的，我们知道怎样构建响应式微服务，但是，这里仍然有一些我们需要关注的缺点：首先，如果你仅仅有http服务，你怎样避免硬编码位置？可恢复性呢？在这一章节我们已经看到了超时和重试，但是熔断器(circuit breaker)、故障转移(failover)、隔仓(bulkhead)呢？让我们继续我们的旅程。
如果你想更深入这些topic:
. Vert.X Web文档(http://vertx.io/docs/vertx-web/java)
. Vert.X Web客户端文档(http://vertx.io/docs/vertx-web-client/java)
. Vert.X响应式微服务
网友评论