concat() vs. merge() Operators in RxJava Observables
Last updated: May 31, 2024
Mocking is an essential part of unit testing, and the Mockito library makes it easy to write clean and intuitive unit tests for your Java code.
Get started with mocking and improve your application tests using our Mockito guide:
Handling concurrency in an application can be a tricky process with many potential pitfalls. A solid grasp of the fundamentals will go a long way to help minimize these issues.
Get started with understanding multi-threaded applications with our Java Concurrency guide:
Spring 5 added support for reactive programming with the Spring WebFlux module, which has been improved upon ever since. Get started with the Reactor project basics and reactive programming in Spring Boot:
Since its introduction in Java 8, the Stream API has become a staple of Java development. The basic operations like iterating, filtering, mapping sequences of elements are deceptively simple to use.
But these can also be overused and fall into some common pitfalls.
To get a better understanding on how Streams work and how to combine them with other language features, check out our guide to Java Streams:
Explore Spring Boot 3 and Spring 6 in-depth through building a full REST API with the framework:
Yes, Spring Security can be complex, from the more advanced functionality within the Core to the deep OAuth support in the framework.
I built the security material as two full courses - Core and OAuth, to get practical with these more complex scenarios. We explore when and how to use each feature and code through it on the backing project.
You can explore the course here:
Spring Data JPA is a great way to handle the complexity of JPA with the powerful simplicity of Spring Boot.
Get started with Spring Data JPA through the guided reference course:
Refactor Java code safely — and automatically — with OpenRewrite.
Refactoring big codebases by hand is slow, risky, and easy to put off. That’s where OpenRewrite comes in. The open-source framework for large-scale, automated code transformations helps teams modernize safely and consistently.
Each month, the creators and maintainers of OpenRewrite at Moderne run live, hands-on training sessions — one for newcomers and one for experienced users. You’ll see how recipes work, how to apply them across projects, and how to modernize code with confidence.
Join the next session, bring your questions, and learn how to automate the kind of work that usually eats your sprint time.
Distributed systems often come with complex challenges such as service-to-service communication, state management, asynchronous messaging, security, and more.
Dapr (Distributed Application Runtime) provides a set of APIs and building blocks to address these challenges, abstracting away infrastructure so we can focus on business logic.
In this tutorial, we'll focus on Dapr's pub/sub API for message brokering. Using its Spring Boot integration, we'll simplify the creation of a loosely coupled, portable, and easily testable pub/sub messaging system:
Spring 5 added support for reactive programming with the Spring WebFlux module, which has been improved upon ever since. Get started with the Reactor project basics and reactive programming in Spring Boot:
1. Overview
concat() and merge() are two powerful operators used to combine multiple Observable instances in RxJava.
concat() emits items from each Observable sequentially, waiting for each to complete before moving to the next, while merge() concurrently emits items from all Observable instances as they’re produced.
In this tutorial, we’ll explore scenarios in which concat() and merge() show similar and different behaviors.
2. Synchronous Sources
The concat() and merge() operators behave exactly the same when both sources are synchronous. Let’s simulate this scenario to understand it better.
2.1. Scenario Setup
Let’s start by using the Observable.just() factory method to create three synchronous sources:
Observable<Integer> observable1 = Observable.just(1, 2, 3);
Observable<Integer> observable2 = Observable.just(4, 5, 6);
Observable<Integer> observable3 = Observable.just(7, 8, 9);Further, let’s create two subscribers, namely, testSubscriberForConcat and testSubscriberForMerge:
TestSubscriber<Integer> testSubscriberForConcat = new TestSubscriber<>();
TestSubscriber<Integer> testSubscriberForMerge = new TestSubscriber<>();Great! We’ve got everything we need to test the scenario.
2.2. concat() and merge()
First, let’s apply the concat() operator and subscribe the resultant Observable with testSubscriberForConcat:
Observable.concat(observable1, observable2, observable3)
.subscribe(testSubscriberForConcat);Further, let’s verify that the emissions are in order where items from observable1 appear before observable2, and observable2 before observable3:
testSubscriberForConcat.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9);
Similarly, we can apply the merge() operator and subscribe the outcome with the testSubscriberForMerge:
Observable.merge(observable1, observable2, observable3).subscribe(testSubscriberForMerge);Next, let’s verify that the emissions through merge follow the same order as that from the concatenation:
testSubscriberForMerge.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9);
Lastly, we must note that synchronous Observable instances emit all items immediately and then signal completion. Further, each Observable completes its emission before the next one starts. Consequently, both operators process each Observable sequentially, producing the same output.
As such, whether the sources are synchronous or asynchronous, the general rule is that if we need to maintain the order of emissions by source, we should use concat(). On the other hand, if we want to combine items as they’re emitted from multiple sources, we should use merge().
3. Predictable Asynchronous Sources
In this section, let’s simulate a scenario with asynchronous sources where the order of emissions is predictable.
3.1. Scenario Setup
Let’s create two asynchronous sources, namely, observable1 and observable2:
Observable<Integer> observable1 = Observable.interval(100, TimeUnit.MILLISECONDS)
.map(i -> i.intValue() + 1)
.take(3);
Observable<Integer> observable2 = Observable.interval(30, TimeUnit.MILLISECONDS)
.map(i -> i.intValue() + 4)
.take(7);We must notice that emissions from observable1 arrive after 100ms, 200ms, and 300ms, respectively. On the other hand, emissions from observable2 arrive at intervals of 30ms.
Now, let’s create the testSubscriberForConcat and testSubscriberforMerge as well:
TestSubscriber<Integer> testSubscriberForConcat = new TestSubscriber<>();
TestSubscriber<Integer> testSubscriberForMerge = new TestSubscriber<>();Fantastic! We’re ready to test this scenario.
3.2. concat() vs. merge()
First, let’s apply the concat() operator and call subscribe() with testSubscribeForConcat:
Observable.concat(observable1, observable2)
.subscribe(testSubscriberForConcat);Next, we must call the awaitTerminalEvent() method to ensure that all emissions are received:
testSubscriberForConcat.awaitTerminalEvent();Now, we can validate that the result contains all items from observable1 followed by all items from observable2:
testSubscriber.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
Further, let’s apply the merge() operator and call subscribe() with testSubscriberForMerge:
Observable.merge(observable1, observable2)
.subscribe(testSubscriberForMerge);Lastly, let’s wait for the emissions and check the emitted values:
testSubscriberForMerge.awaitTerminalEvent();
testSubscriberForMerge.assertValues(4, 5, 6, 1, 7, 8, 9, 2, 10, 3);
The result contains all items interleaved together in the order of their actual emission from observer1 and observer2.
4. Asynchronous Sources With Race Conditions
In this section, we’ll simulate a scenario with two asynchronous sources in which the order of combined emission is fairly unpredictable.
4.1. Scenario Setup
First, let’s create the two asynchronous sources with exactly the same delay:
Observable<Integer> observable1 = Observable.interval(100, TimeUnit.MILLISECONDS)
.map(i -> i.intValue() + 1)
.take(3);
Observable<Integer> observable2 = Observable.interval(100, TimeUnit.MILLISECONDS)
.map(i -> i.intValue() + 4)
.take(3);We know that one emission from each source arrives after 100ms, 200ms, and 300ms. However, we can’t predict the exact order because of the race conditions.
Next, let’s create two test subscribers:
TestSubscriber<Integer> testSubscriberForConcat = new TestSubscriber<>();
TestSubscriber<Integer> testSubscriberForMerge = new TestSubscriber<>();Perfect! We’re good to go now.
4.2. concat() vs. merge()
First, let’s apply the concat() operator, followed by a subscription to testSubscribeForConcat:
Observable.concat(observable1, observable2)
.subscribe(testSubscriberForConcat);
testSubscriberForConcat.awaitTerminalEvent();Now, let’s verify that the outcome from the concat() operator remains unchanged:
testSubscriberForConcat.assertValues(1, 2, 3, 4, 5, 6);
Further, let’s merge() and subscribe with testSubscriberForMerge:
Observable.merge(observable1, observable2)
.subscribe(testSubscriberForMerge);
testSubscriberForMerge.awaitTerminalEvent();Next, let’s accumulate all the emissions in a list and verify that it contains all values:
List<Integer> actual = testSubscriberForMerge.getOnNextEvents();
List<Integer> expected = Arrays.asList(1, 2, 3, 4, 5, 6);
assertTrue(actual.containsAll(expected) && expected.containsAll(actual));Lastly, let’s also log the emissions to see it in action:
21:05:43.252 [main] INFO actual emissions: [4, 1, 2, 5, 3, 6]
We can receive a different order for different runs.
5. Conclusion
In this article, we saw how the concat() and merge() operators in RxJava handle synchronous and asynchronous data sources. Further, we compared scenarios involving predictable and unpredictable patterns of emissions, emphasizing the differences between the two operators.
