RxJava + Java8 + Java EE 7 + Arquillian =幸福

消防反应管理_DigitalStorm_226x150

微服务是一种体系结构样式，其中每个服务都实现为一个独立的系统。他们可以使用自己的持久性系统（尽管不是强制性的），部署，语言等。

由于系统由一个以上的服务组成，因此每个服务将与其他服务通信，通常使用轻量级协议（如HTTP）并遵循Restful Web方法。您可以在这里阅读有关微服务的更多信息： http : //martinfowler.com/articles/microservices.html

让我们看一个非常简单的例子。假设我们有一家预订商店，用户可以在其中导航目录，当他们找到想要查看更多信息的书时，他们单击isbn，然后打开一个新屏幕，其中包含该书的详细信息和有关该书的评论由读者撰写。

该系统可能由两个服务组成：

一种获取书籍详细信息的服务。可以从任何传统系统（如RDBMS）中检索它们。
一种将所有评论写在书中的服务，在这种情况下，信息可以存储在文档库中。

这里的问题是，对于用户的每个请求，我们需要打开两个连接，每个服务一个。当然，我们需要一种并行执行这些工作的方法来提高性能。这是一个问题，我们如何处理异步请求？第一个想法是使用Future类。对于两个服务可能很好，但是如果您需要四个或五个服务，则代码将变得越来越复杂，例如，您可能需要从一个服务获取数据并将其用于另一服务或将一个服务的结果改编为输入另一个。因此，存在线程和同步管理的成本。

有一种干净整洁的方式来解决这个问题的方法真是太棒了。这正是RxJava所做的。 RxJava是Reactive Extensions的Java VM实现：该库用于通过使用可观察的序列来组成异步和基于事件的程序。

使用RxJava而不是从结构中提取数据，而是将数据推送到它，该数据与订阅者侦听的事件做出反应并一致地起作用。您可以在https://github.com/Netflix/RxJava中找到更多信息。

因此，在这种情况下，我们将实现的示例是此处描述的示例，该示例使用RxJava ， Java EE 7 ， Java 8和Arquillian进行测试。

这篇文章假定您知道如何使用Java EE规范编写Rest服务。

因此，让我们从两个服务开始：

@Singleton
@Path("bookinfo")
public class BookInfoService {@GET@Path("{isbn}")@Produces(MediaType.APPLICATION_JSON)@Consumes(MediaType.APPLICATION_JSON)public JsonObject findBookByISBN(@PathParam("isbn") String isbn) {return Json.createObjectBuilder().add("author", "George R.R. Martin").add("isbn", "1111").add("title", "A Game Of Thrones").build();}}

@Singleton
@Path("comments")
public class CommentsService {@GET@Path("{isbn}")@Produces(MediaType.APPLICATION_JSON)public JsonArray bookComments(@PathParam("isbn") String isbn) {return Json.createArrayBuilder().add("Good Book").add("Awesome").build();}}

@ApplicationPath("rest")
public class ApplicationResource extends Application {
}

最后是时候创建第三个外观服务，该服务从客户端接收通信，并行向两个服务发送请求，最后压缩两个响应。 zip是将通过指定函数发出的一组项目组合在一起并将其发送回客户端的过程（不要与压缩混淆！）。

@Singleton
@Path("book")
public class BookService {private static final String BOOKSERVICE = "http://localhost:8080/bookservice";private static final String COMMENTSERVICE = "http://localhost:8080/bookcomments";@Resource(name = "DefaultManagedExecutorService")ManagedExecutorService executor;Client bookServiceClient;WebTarget bookServiceTarget;Client commentServiceClient;WebTarget commentServiceTarget;@PostConstructvoid initializeRestClients() {bookServiceClient = ClientBuilder.newClient();bookServiceTarget = bookServiceClient.target(BOOKSERVICE + "/rest/bookinfo");commentServiceClient = ClientBuilder.newClient();commentServiceTarget = commentServiceClient.target(COMMENTSERVICE + "/rest/comments");}@GET@Path("{isbn}")@Produces(MediaType.APPLICATION_JSON)public void bookAndComment(@Suspended final AsyncResponse asyncResponse, @PathParam("isbn") String isbn) {//RxJava code shown below}
}

基本上，我们创建一个新服务。在这种情况下，我们将要连接的两个服务的URL都是硬编码的。这样做是出于学术目的，但是您将在类似于生产的代码中从生产者类或属性文件或用于此目的的任何系统中注入它。然后，我们创建javax.ws.rs.client.WebTarget来使用Restful Web Service 。

之后，我们需要使用RxJava API实现bookAndComment方法。

RxJava中使用的主要类是rx.Observabl e。正如他的名字所暗示的那样，该类是可观察的，它是引发事件以推动对象的原因。默认情况下，事件是同步的，开发人员有责任使它们异步。

因此，我们需要为每个服务提供一个异步可观察实例：

public Observable<JsonObject> getBookInfo(final String isbn) {return Observable.create((Observable.OnSubscribe<JsonObject>) subscriber -> {Runnable r = () -> {subscriber.onNext(bookServiceTarget.path(isbn).request().get(JsonObject.class));subscriber.onCompleted();};executor.execute(r);});
}

基本上，我们创建一个Observable ，当订户订阅它时将执行指定的功能。该函数是使用lambda表达式创建的，以避免创建嵌套的内部类。在这种情况下，由于调用bookinfo服务，我们将返回JsonObject 。结果将传递到onNext方法，以便订阅者可以接收结果。因为我们要异步执行此逻辑，所以代码被包装在Runnable块中。

完成所有逻辑后，还需要调用onCompleted方法。

注意，因为除了创建Runnable之外 ，我们还希望使可观察的异步发生，所以我们使用Executor在单独的线程中运行逻辑。 Java EE 7中的一项重大新增功能是在容器内创建线程的一种托管方式。在这种情况下，我们使用容器提供的ManagedExecutorService在当前任务的不同线程中异步跨越任务。

public Observable<JsonArray> getComments(final String isbn) {return Observable.create((Observable.OnSubscribe<JsonArray>) subscriber -> {Runnable r = () -> {subscriber.onNext(commentServiceTarget.path(isbn).request().get(JsonArray.class));subscriber.onCompleted();};executor.execute(r);});
}

与之前的内容类似，但没有获取书籍信息，而是获得了一系列评论。

然后，当两个响应均可用时，我们需要创建一个负责将两个响应压缩的可观察对象。这是通过在Observable类上使用zip方法完成的，该方法接收两个Observable并应用一个函数来合并两个结果。在这种情况下，一个lambda表达式将创建一个新的json对象，附加两个响应。

@GET
@Path("{isbn}")
@Produces(MediaType.APPLICATION_JSON)
public void bookAndComment(@Suspended final AsyncResponse asyncResponse, @PathParam("isbn") String isbn) {//Calling previous defined functionsObservable<JsonObject> bookInfo = getBookInfo(isbn);Observable<JsonArray> comments = getComments(isbn);Observable.zip(bookInfo, comments, (JsonObject book, JsonArray bookcomments) ->Json.createObjectBuilder().add("book", book).add("comments", bookcomments).build()).subscribe(new Subscriber<JsonObject>() {@Overridepublic void onCompleted() {}@Overridepublic void onError(Throwable e) {asyncResponse.resume(e);}@Overridepublic void onNext(JsonObject jsonObject) {asyncResponse.resume(jsonObject);}});
}

让我们看一下以前的服务。我们正在使用@Suspended批注来使用Java EE中的新增功能之一，即Jax-Rs 2.0异步REST端点。基本上，我们正在做的是释放服务器资源，并使用resume方法在响应可用时生成响应。

最后是测试。我们将Wildfly 8.1用作Java EE 7服务器和Arquillian 。因为每个服务可以在不同的服务器上进行部署，我们将在不同的战争，而是同一个服务器内部署各项服务。

因此，在这种情况下，我们将部署三个战争文件，这在Arquillian中非常容易做到。

@RunWith(Arquillian.class)
public class BookTest {@Deployment(testable = false, name = "bookservice")public static WebArchive createDeploymentBookInfoService() {return ShrinkWrap.create(WebArchive.class, "bookservice.war").addClasses(BookInfoService.class, ApplicationResource.class);}@Deployment(testable = false, name = "bookcomments")public static WebArchive createDeploymentCommentsService() {return ShrinkWrap.create(WebArchive.class, "bookcomments.war").addClasses(CommentsService.class, ApplicationResource.class);}@Deployment(testable = false, name = "book")public static WebArchive createDeploymentBookService() {WebArchive webArchive = ShrinkWrap.create(WebArchive.class, "book.war").addClasses(BookService.class, ApplicationResource.class).addAsLibraries(Maven.resolver().loadPomFromFile("pom.xml").resolve("com.netflix.rxjava:rxjava-core").withTransitivity().as(JavaArchive.class));return webArchive;}@ArquillianResourceURL base;@Test@OperateOnDeployment("book")public void should_return_book() throws MalformedURLException {Client client = ClientBuilder.newClient();JsonObject book = client.target(URI.create(new URL(base, "rest/").toExternalForm())).path("book/1111").request().get(JsonObject.class);//assertions}
}

在这种情况下，客户将要求一本书提供所有信息。在服务器部分中， zip方法将等待直到并行检索书和注释，然后将两个响应组合到一个对象中并发送回客户端。

这是RxJava的非常简单的示例。实际上，在这种情况下，我们只看到了如何使用zip方法，但是RxJava提供了许多其他有用的方法，例如take（） ， map（） ， merge（） ，…（ https：// github .com / Netflix / RxJava / wiki / Alphabetical-Observable-Operators列表）

此外，在此示例中，我们仅看到了连接两个服务并并行检索信息的示例，您可能想知道为什么不使用Future类。在此示例中使用Future和Callbacks完全可以，但是在现实生活中，您的逻辑将不像压缩两个服务那样容易。也许您将拥有更多服务，也许您需要从一项服务中获取信息，然后针对每个结果打开一个新的连接。如您所见，您可能从两个Future实例开始，但以一堆Future.get（）方法，超时等结束，因此，在这些情况下， RxJava确实简化了应用程序的开发。

此外，我们还看到了如何使用Java EE 7的一些新增功能，例如如何使用Jax-Rs开发异步Restful服务。

在这篇文章中，我们学习了如何处理服务之间的互连以及如何使它们可伸缩并减少资源消耗。但是，我们没有谈论这些服务之一发生故障时发生的情况。来电者怎么了？我们有办法进行管理吗？当其中一项服务不可用时，是否有一种方法可以不浪费资源？我们将在下一篇关于容错的文章中对此进行介绍。

我们不断学习，

亚历克斯