{"id":678,"date":"2018-06-27T03:02:02","date_gmt":"2018-06-27T01:02:02","guid":{"rendered":"http:\/\/itblog.inginea.eu\/?p=678"},"modified":"2021-11-07T04:37:42","modified_gmt":"2021-11-07T03:37:42","slug":"speed-up-services-with-reactive-api-in-java-ee-8","status":"publish","type":"post","link":"https:\/\/ondro.inginea.eu\/index.php\/speed-up-services-with-reactive-api-in-java-ee-8\/","title":{"rendered":"Speed Up Services With Reactive API in Java EE 8"},"content":{"rendered":"
This entry is part 1 of 1 in the series Java EE 8 Microservices<\/a><\/div>\n

Services can often be optimized with asynchronous processing even without changing their behavior towards the outside world. The reason why some services aren’t efficient is that they need to wait for other services to provide a result to continue further. Let’s look how to call external REST services without waiting for them and also do multiple parallel calls independently and combine their results later with a reactive pipeline in Java EE 8.<\/p>\n\n\n\n\n\n\n\n

This topic and much more will be described in more detail in my upcoming book Java EE 8 Microservices<\/a>, which I co-author with Mert \u00c7aliskan and Pavel Pscheidl<\/p><\/blockquote>\n\n\n\n

If our service calls multiple microservices and waits for each call to finish and return results before doing another call, it’s a good candidate to refactor using reactive API. In order to make the service more efficient, it could do all the calls to external services in parallel if they don’t depend on each other. This would decrease the time spent waiting and thus speed up the microservice.<\/p>\n\n\n\n

In order to call REST services in parallel, we’ll use the new reactive client API in JAX-RS. We’ll combine it with the RxJava library to combine their results when available. This combination will allow us to write clean and efficient code. And with an additional benefit that the current thread can be released for further processing while waiting for results from remote calls.<\/p>\n\n\n\n

We’ll build a pipeline which processes the results as they arrive and finally merges them into a single response. The first part of the pipeline will call each remote service. Instead of waiting for the results, we’ll specify what to do with each received result and continue with calling other services. Using the rx() method on the JAX-RS client request builder allows us to call a version of the get()<\/code> method, which immediately returns instead of waiting for the result. In order to process results when they arrive, we can chain method handlers onto a CompletionStage<\/code> returned from the rx<\/code> version of the get() <\/code> method:<\/p>\n\n\n

CompletionStage<Forecast<\/span>><\/span> stage = temperatureServiceTarget\n  .request()\n  .rx()\n  .get(Temperature.class)\n  .thenApply(temperature -> new Forecast(temperature));<\/code><\/span>Code language:<\/span> HTML, XML<\/span> (<\/span>xml<\/span>)<\/span><\/small><\/pre>\n\n\n
The above code will call a temperature service and then register a lambda expression to process the resulting temperature when it arrives. This maps the temperature to a forecast object, which can be accessed with the stage  variable later.<\/p>\n\n\n\n
With the following code, we will call an external service and return a Flowable:<\/p>\n\n\n
Flowable<Forecast<\/span>><\/span> flowable = temperatureServiceTarget\n  .register(RxFlowableInvokerProvider.class)\n  .request()\n  .rx(RxFlowableInvoker.class)\n  .get(Temperature.class)\n  .map(temperature -> new Forecast(temperature);<\/code><\/span>Code language:<\/span> HTML, XML<\/span> (<\/span>xml<\/span>)<\/span><\/small><\/pre>\n\n\n
We register additional RxFlowableInvokerProvider<\/code>, which allows to request RxFlowableInvoker<\/code>  later. This invoker then gives us the Flowable<\/code>  return type from RxJava. These classes are not in the JAX-RS API and we must add them with the Jersey RxJava2 library:<\/p>\n\n\n
<dependency<\/span>><\/span>\n  <groupId<\/span>><\/span>org.glassfish.jersey.ext.rx<\/groupId<\/span>><\/span>\n  <artifactId<\/span>><\/span>jersey-rx-client-rxjava2<\/artifactId<\/span>><\/span>\n  <version<\/span>><\/span>2.26<\/version<\/span>><\/span>\n<\/dependency<\/span>><\/span><\/code><\/span>Code language:<\/span> HTML, XML<\/span> (<\/span>xml<\/span>)<\/span><\/small><\/pre>\n\n\n
On the first sight it seems we made the code more complicated while doing the same thing. But a Flowable<\/code> instance allows us to combine multiple calls easily:<\/p>\n\n\n
Flowable.concat(flowable1, flowable2)\n  .doOnNext(forecast -> {\n    forecasts.add(forecast);\n  })\n  .doOnComplete(() -> {\n    asyncResponse.resume(Response.ok(forecasts).build());\n  })\n  .doOnError(asyncResponse::resume)\n  .subscribe();\n}<\/code><\/span>Code language:<\/span> PHP<\/span> (<\/span>php<\/span>)<\/span><\/small><\/pre>\n\n\n
For each forecast received from any flowable, we add it to a list of forecasts. Finally, we send the list of forecasts as a response or send an error response. The final call to subscribe()<\/code>  is necessary to register the listeners, otherwise they would be ignored.<\/p>\n\n\n\n
@GET<\/span>\n@Produces(MediaType.APPLICATION_JSON)\npublic void getForecasts(@Suspended AsyncResponse asyncResponse) {\n  \/* ...here come some asynchronous calls to REST services... *\/<\/span>\n  asyncResponse<\/span>.resume<\/span>(...)\n}<\/code><\/span>Code language:<\/span> CSS<\/span> (<\/span>css<\/span>)<\/span><\/small><\/pre>\n\n\n
Final code example<\/h2>\n\n\n\n
The following code will:<\/p>\n\n\n\n
You may have also noticed the asyncResponse<\/code>  variable used to send the final response or signal an error. This is a JAX-RS asynchronous response instance, which is used to complete a REST response at later time, when the data is available, without blocking the initial processing thread. Using the asynchronous response helps us save thread resources while waiting for results from external REST services. In order to turn on asynchronous processing in our REST endpoint, we will inject javax.ws.rs.container.AsyncResponse<\/a> as the REST method argument, together with the @Suspended<\/a> annotation. We will also change the return type to void<\/code> because we’ll be building the response using the AsyncResponse<\/code> instance:<\/p>\n\n\n
However, we want to use another variant of the get()<\/code>  method together with an RxJava Flowable Invoker from the Jersey project to get a Flowable<\/code>  from RxJava instead of a CompletionStage<\/code>. The Flowable<\/a> interface makes it easier to combine multiple asynchronous results with much simpler code than CompletionStage<\/code> and also more efficiently.<\/p>\n\n\n\n