在RxJava中组合Observables

1. Introduction

In this quick tutorial, we’ll discuss different ways of combining Observables in RxJava.

If you’re new to RxJava, definitely check out this intro tutorial first.

Now, let’s jump right in.

2. Observables

Observable sequences, or simply Observables, are representations of asynchronous data streams.

These’re based on the Observer pattern wherein an object called an Observer, subscribes to items emitted by an Observable.

The subscription is non-blocking as the Observer stands to react to whatever the Observable will emit in the future. This, in turn, facilitates concurrency.

Here’s a simple demonstration in RxJava:

Observable
  .from(new String[] { "John", "Doe" })
  .subscribe(name -> System.out.println("Hello " + name))

3. Combining Observables

When programming using a reactive framework, it’s a common use-case to combine various Observables.

In a web application, for example, we may need to get two sets of asynchronous data streams that are independent of each other.

Instead of waiting for the previous stream to complete before requesting the next stream, we can call both at the same time and subscribe to the combined streams.

In this section, we’ll discuss some of the different ways we can combine multiple Observables in RxJava and the different use-cases to which each method applies.

3.1. Merge

We can use the merge operator to combine the output of multiple Observables so that they act like one:

@Test
public void givenTwoObservables_whenMerged_shouldEmitCombinedResults() {
    TestSubscriber<String> testSubscriber = new TestSubscriber<>();

    Observable.merge(
      Observable.from(new String[] {"Hello", "World"}),
      Observable.from(new String[] {"I love", "RxJava"})
    ).subscribe(testSubscriber);

    testSubscriber.assertValues("Hello", "World", "I love", "RxJava");
}

3.2. MergeDelayError

The mergeDelayError method is the same as merge in that it combines multiple Observables into one, but if errors occur during the merge, it allows error-free items to continue before propagating the errors:

@Test
public void givenMutipleObservablesOneThrows_whenMerged_thenCombineBeforePropagatingError() {
    TestSubscriber<String> testSubscriber = new TestSubscriber<>();
        
    Observable.mergeDelayError(
      Observable.from(new String[] { "hello", "world" }),
      Observable.error(new RuntimeException("Some exception")),
      Observable.from(new String[] { "rxjava" })
    ).subscribe(testSubscriber);

    testSubscriber.assertValues("hello", "world", "rxjava");
    testSubscriber.assertError(RuntimeException.class);
}

The above example emits all the error-free values:

hello
world
rxjava

Note that if we use merge instead of mergeDelayError, the String “rxjava” won’t be emitted because merge immediately stops the flow of data from Observables when an error occurs.

3.3. Zip

The zip extension method brings together two sequences of values as pairs:

@Test
public void givenTwoObservables_whenZipped_thenReturnCombinedResults() {
    List<String> zippedStrings = new ArrayList<>();

    Observable.zip(
      Observable.from(new String[] { "Simple", "Moderate", "Complex" }), 
      Observable.from(new String[] { "Solutions", "Success", "Hierarchy"}),
      (str1, str2) -> str1 + " " + str2).subscribe(zippedStrings::add);
        
    assertThat(zippedStrings).isNotEmpty();
    assertThat(zippedStrings.size()).isEqualTo(3);
    assertThat(zippedStrings).contains("Simple Solutions", "Moderate Success", "Complex Hierarchy");
}

3.4. Zip With Interval

In this example, we will zip a stream with interval which in effect will delay the emission of elements of the first stream:

@Test
public void givenAStream_whenZippedWithInterval_shouldDelayStreamEmmission() {
    TestSubscriber<String> testSubscriber = new TestSubscriber<>();
        
    Observable<String> data = Observable.just("one", "two", "three", "four", "five");
    Observable<Long> interval = Observable.interval(1L, TimeUnit.SECONDS);
        
    Observable
      .zip(data, interval, (strData, tick) -> String.format("[%d]=%s", tick, strData))
      .toBlocking().subscribe(testSubscriber);
        
    testSubscriber.assertCompleted();
    testSubscriber.assertValueCount(5);
    testSubscriber.assertValues("[0]=one", "[1]=two", "[2]=three", "[3]=four", "[4]=five");
}

4. Summary

In this article, we’ve seen a few of the methods for combining Observables with RxJava. You can learn about other methods like combineLatest, join, groupJoin, switchOnNext, in the official RxJava documentation.

As always, the source code for this article is available in our GitHub repo.