Reactive Java (RxJava) Tutorial: Advanced

Reactive Java or RxJava is an implementation and enhancement of the observer pattern. It was intended for use in event driven schemes where nesting synchronous or asynchronous callback methods becomes overly complex. It abstracts some of the more complex tasks associated with asynchronous operations including threading and concurrency.

This example builds on the basics described in Part One of this topic. If you are not familiar with the basics, please look back at this previous article before moving forward with these more advanced concepts. This article will examine asynchronous or concurrency tasks that are made easier with RxJava. The concurrency problems that RxJava is suitable to solve and we will look at in this example include:

• Nested callbacks
• Making asynchronous calls
• Aggregating or combining asynchronous calls
• Streaming

You can also check this tutorial in the following video:

1. Setup

We used Eclipse Neon, Java 8, Maven 3.3.9 and RxJava 2.0.0.  At the time of this sample, we ran into issues using the last version, 2.0.7, with Eclipse as the source was not available. This example uses a simple Java application to demonstrate Reactive functionality.

pom.xml

2. Simple Asynchronous Call

Let’s start by exploring the asynchronous capabilities in RxJava. In the next few examples we will use RxJava to spawn new threads to do various tasks. The default behavior of an Observable is to observe on the same thread where the subscribe method is called. You can introduce an asynchronous process using the subscribeOn method. Here we will look at a simple asynchronous call in RxJava.

Notice that in this example we are subscribing on the Schedulers.newThread() scheduler and using lambda notation to execute the serialize method of the FlowableEmitter interface. The FlowableEmitter interface requires that we call the onNext, onError and onComplete methods sequentially. In this example, we will simplify things and use lambda notation to call the serialize method which will ensure that these methods are serialized. The body of this method will act as the worker thread and will sleep for three seconds while the process moves ahead. You can see from the output that the flowable messages, “RxJavaExampleAdvanced:51 – Executing async flowable…”, are logged after the main thread finishes “RxJavaExampleAdvanced:59 – Finished simple async”.

Output

3. Asynchronous Web Service Call

A common usage of RxJava is to make long running calls or calls with unpredictable finish times asynchronously. This allows the code to handle other tasks while it is waiting for the long running call to finish. You might see a client user interface make a web service call for data asynchronously so it can finish displaying the components to the user that are not dependent on the data in the service call. In the next example, we will explore using RxJava to make asynchronous calls to a web service. We make one simple call to a web service or API that returns a String result. You’ll notice that the API is executed on a new thread “RxNewThreadScheduler-1” not the main thread. Again we use the FlowableEmitter interface with lambda notation to execute the serialize method which makes a rest call to our API on a new thread.

Output

4. Multiple Asynchronous Web Service Calls

Many times you’ll need to make multiple calls to a web service. In this next example we’ll leverage the map function of RxJava to execute and return the response from multiple API calls. As a reminder, the map function returns the data type of the Flowable while flatMap will return the Flowable object. Using the Flowable that is returned from a flatMap call would allow you to take further reactive actions on the response. You’ll see this in later examples but in this case we are just outputing the result and will not need to make use of the Flowable.

Output

While this code will operate the emitted calls on another thread it does not create a new thread for each item. If we want to have our process execute on a new thread for each item we will need to do something slightly different. As a developer leveraging these reactive mechanisms you need to think carefully about which calls you want to make synchronously versus those that you want to make asynchronously. If you need to use the results of earlier calls in later calls, you will want each to execute synchronously. If you just need all the data returned and you are not concerned with the order in which it is returned, you might have a good candidate to make each call asynchronously which will improve the overall performance.

We will start with a simple asynchronous example where multiple items are emitted and build on that to use the same technique for multiple API calls. There’s a lot going on in this small sample. We are using the flatMap method to operate on an array of items and transform the array to another Observable for each item. We’ll need to do that to ensure that we can subscribeOn each item which will execute those corresponding operations on a different thread. This is how we ensure that each emitted item is handled asychronously rather than the group sequentially executed on a separate thread. Notice that we have a different thread for both of the items in this example; "Thread[RxNewThreadScheduler-1,5,main]", "Thread[RxNewThreadScheduler-2,5,main]".

Output

Now lets take a look at an asynchronous example where we spawn a new thread for each API call. Again we map each emitted value to a new Observable with a single item and subscribeOn to a new thread. Again, you can see in this sample that the thread id that is executing each request is different, i.e. Thread:Thread[RxNewThreadScheduler-4,5,main].

Output

5. Combine Asynchronous Results

To build on this technique, we will make multiple API calls and zip or combine the results. We are making multiple API calls asychronously, each on its own thread, and using the zip method to combine the results.

Output

6. Streaming Results

Finally, let’s examine streaming the results of asynchronous API calls where the results are emitted as they are available. This example just builds on the concepts introduced previously with the key addition of the ObservableHttp calls that leverage a CloseableHttpAsyncClient.

Output

Here is the Spring Boot streaming resource that ties it together.

7. Summary

In this example, we looked at making asynchronous calls using RxJava, including calls to RESTful web services. We examined how certain usages make all asynchronous calls on a single thread and how to use Observables to make each call on a separate thread. We also looked at combining the results of multiple calls and finally streaming results of service calls.

8. Download the Source Code

Here we demonstrated how to use the basic RxJava operations.

Do you want to know how to develop your skillset to become a Java Rockstar?

Subscribe to our newsletter to start Rocking right now!

To get you started we give you our best selling eBooks for FREE!

1. JPA Mini Book

2. JVM Troubleshooting Guide

3. JUnit Tutorial for Unit Testing

4. Java Annotations Tutorial

5. Java Interview Questions

6. Spring Interview Questions

7. Android UI Design

and many more ....

I agree to the Terms and Privacy Policy

Sign up

Thank you!

We will contact you soon.