RxJava Fetching Observables In Parallel
The parallel operator proved to be a problem for almost all use cases and does not do what most expect from it, so it was removed in the 1.0.0.rc.4 release: https://github.com/ReactiveX/RxJava/pull/1716
A good example of how to do this type of behavior and get parallel execution can be seen here.
In your example code it is unclear if searchServiceClient is synchronous or asynchronous. It affects how to solve the problem slightly as if it is already async no extra scheduling is needed. If synchronous extra scheduling is needed.
First here are some simple examples showing synchronous and asynchronous behavior:
import rx.Observable;import rx.Subscriber;import rx.schedulers.Schedulers;public class ParallelExecution { public static void main(String[] args) { System.out.println("------------ mergingAsync"); mergingAsync(); System.out.println("------------ mergingSync"); mergingSync(); System.out.println("------------ mergingSyncMadeAsync"); mergingSyncMadeAsync(); System.out.println("------------ flatMapExampleSync"); flatMapExampleSync(); System.out.println("------------ flatMapExampleAsync"); flatMapExampleAsync(); System.out.println("------------"); } private static void mergingAsync() { Observable.merge(getDataAsync(1), getDataAsync(2)).toBlocking().forEach(System.out::println); } private static void mergingSync() { // here you'll see the delay as each is executed synchronously Observable.merge(getDataSync(1), getDataSync(2)).toBlocking().forEach(System.out::println); } private static void mergingSyncMadeAsync() { // if you have something synchronous and want to make it async, you can schedule it like this // so here we see both executed concurrently Observable.merge(getDataSync(1).subscribeOn(Schedulers.io()), getDataSync(2).subscribeOn(Schedulers.io())).toBlocking().forEach(System.out::println); } private static void flatMapExampleAsync() { Observable.range(0, 5).flatMap(i -> { return getDataAsync(i); }).toBlocking().forEach(System.out::println); } private static void flatMapExampleSync() { Observable.range(0, 5).flatMap(i -> { return getDataSync(i); }).toBlocking().forEach(System.out::println); } // artificial representations of IO work static Observable<Integer> getDataAsync(int i) { return getDataSync(i).subscribeOn(Schedulers.io()); } static Observable<Integer> getDataSync(int i) { return Observable.create((Subscriber<? super Integer> s) -> { // simulate latency try { Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } s.onNext(i); s.onCompleted(); }); }}Following is an attempt at providing an example that more closely matches your code:
import java.util.List;import rx.Observable;import rx.Subscriber;import rx.schedulers.Schedulers;public class ParallelExecutionExample { public static void main(String[] args) { final long startTime = System.currentTimeMillis(); Observable<Tile> searchTile = getSearchResults("search term") .doOnSubscribe(() -> logTime("Search started ", startTime)) .doOnCompleted(() -> logTime("Search completed ", startTime)); Observable<TileResponse> populatedTiles = searchTile.flatMap(t -> { Observable<Reviews> reviews = getSellerReviews(t.getSellerId()) .doOnCompleted(() -> logTime("getSellerReviews[" + t.id + "] completed ", startTime)); Observable<String> imageUrl = getProductImage(t.getProductId()) .doOnCompleted(() -> logTime("getProductImage[" + t.id + "] completed ", startTime)); return Observable.zip(reviews, imageUrl, (r, u) -> { return new TileResponse(t, r, u); }).doOnCompleted(() -> logTime("zip[" + t.id + "] completed ", startTime)); }); List<TileResponse> allTiles = populatedTiles.toList() .doOnCompleted(() -> logTime("All Tiles Completed ", startTime)) .toBlocking().single(); } private static Observable<Tile> getSearchResults(String string) { return mockClient(new Tile(1), new Tile(2), new Tile(3)); } private static Observable<Reviews> getSellerReviews(int id) { return mockClient(new Reviews()); } private static Observable<String> getProductImage(int id) { return mockClient("image_" + id); } private static void logTime(String message, long startTime) { System.out.println(message + " => " + (System.currentTimeMillis() - startTime) + "ms"); } private static <T> Observable<T> mockClient(T... ts) { return Observable.create((Subscriber<? super T> s) -> { // simulate latency try { Thread.sleep(1000); } catch (Exception e) { } for (T t : ts) { s.onNext(t); } s.onCompleted(); }).subscribeOn(Schedulers.io()); // note the use of subscribeOn to make an otherwise synchronous Observable async } public static class TileResponse { public TileResponse(Tile t, Reviews r, String u) { // store the values } } public static class Tile { private final int id; public Tile(int i) { this.id = i; } public int getSellerId() { return id; } public int getProductId() { return id; } } public static class Reviews { }}This outputs:
Search started => 65msSearch completed => 1094msgetProductImage[1] completed => 2095msgetSellerReviews[2] completed => 2095msgetProductImage[3] completed => 2095mszip[1] completed => 2096mszip[2] completed => 2096msgetProductImage[2] completed => 2096msgetSellerReviews[1] completed => 2096mszip[3] completed => 2096msAll Tiles Completed => 2097msgetSellerReviews[3] completed => 2097msI have made each IO call be simulated to take 1000ms so it is obvious where the latency is and that it is happening in parallel. It prints out the progress is makes in elapsed milliseconds.
The trick here is that flatMap merges async calls, so as long as the Observables being merged are async, they will all be executed concurrently.
If a call like getProductImage(t.getProductId()) was synchronous, it can be made asynchronous like this: getProductImage(t.getProductId()).subscribeOn(Schedulers.io).
Here is the important part of the above example without all the logging and boilerplate types:
Observable<Tile> searchTile = getSearchResults("search term");; Observable<TileResponse> populatedTiles = searchTile.flatMap(t -> { Observable<Reviews> reviews = getSellerReviews(t.getSellerId()); Observable<String> imageUrl = getProductImage(t.getProductId()); return Observable.zip(reviews, imageUrl, (r, u) -> { return new TileResponse(t, r, u); }); }); List<TileResponse> allTiles = populatedTiles.toList() .toBlocking().single();I hope this helps.
People who are still @ JDK 7, whose IDE doesn't automatically detect JDK 8 source just yet and what to try out the above brilliant response (and explanation) by @benjchristensen can use this shamelessly refractored, JDK 7, code. Kudos to @benjchristensen for an amazing explanation and example !
import java.util.List;import rx.Observable;import rx.Subscriber;import rx.functions.Action0;import rx.functions.Func1;import rx.functions.Func2;import rx.schedulers.Schedulers;public class ParallelExecutionExample{ public static void main(String[] args) { final long startTime = System.currentTimeMillis(); Observable<Tile> searchTile = getSearchResults("search term") .doOnSubscribe(new Action0() { @Override public void call() { logTime("Search started ", startTime); } }) .doOnCompleted(new Action0() { @Override public void call() { logTime("Search completed ", startTime); } }); Observable<TileResponse> populatedTiles = searchTile.flatMap(new Func1<Tile, Observable<TileResponse>>() { @Override public Observable<TileResponse> call(final Tile t) { Observable<Reviews> reviews = getSellerReviews(t.getSellerId()) .doOnCompleted(new Action0() { @Override public void call() { logTime("getSellerReviews[" + t.id + "] completed ", startTime); } }); Observable<String> imageUrl = getProductImage(t.getProductId()) .doOnCompleted(new Action0() { @Override public void call() { logTime("getProductImage[" + t.id + "] completed ", startTime); } }); return Observable.zip(reviews, imageUrl, new Func2<Reviews, String, TileResponse>() { @Override public TileResponse call(Reviews r, String u) { return new TileResponse(t, r, u); } }) .doOnCompleted(new Action0() { @Override public void call() { logTime("zip[" + t.id + "] completed ", startTime); } }); } }); List<TileResponse> allTiles = populatedTiles .toList() .doOnCompleted(new Action0() { @Override public void call() { logTime("All Tiles Completed ", startTime); } }) .toBlocking() .single(); } private static Observable<Tile> getSearchResults(String string) { return mockClient(new Tile(1), new Tile(2), new Tile(3)); } private static Observable<Reviews> getSellerReviews(int id) { return mockClient(new Reviews()); } private static Observable<String> getProductImage(int id) { return mockClient("image_" + id); } private static void logTime(String message, long startTime) { System.out.println(message + " => " + (System.currentTimeMillis() - startTime) + "ms"); } private static <T> Observable<T> mockClient(final T... ts) { return Observable.create(new Observable.OnSubscribe<T>() { @Override public void call(Subscriber<? super T> s) { try { Thread.sleep(1000); } catch (Exception e) { } for (T t : ts) { s.onNext(t); } s.onCompleted(); } }) .subscribeOn(Schedulers.io()); // note the use of subscribeOn to make an otherwise synchronous Observable async } public static class TileResponse { public TileResponse(Tile t, Reviews r, String u) { // store the values } } public static class Tile { private final int id; public Tile(int i) { this.id = i; } public int getSellerId() { return id; } public int getProductId() { return id; } } public static class Reviews { }}