blocks|key|1670063|text|事实证明，并行运算符在几乎所有用例中都是一个问题，并且没有完成大多数人对它的期望，因此在1.0.0.rc.4版本中删除了它：https://github.com/ReactiveX/RxJava/pull/1716|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|1670064|here是一个很好的例子，说明了如何执行这种类型的行为并获得并行执行。|1670065|在您的示例代码中，不清楚searchServiceClient是同步的还是异步的。它会稍微影响如何解决问题，就好像它已经是异步的一样，不需要额外的调度。如果需要同步额外调度。|style|CODE|1670066|首先，这里有一些简单的示例，展示了同步和异步行为：|1670067|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();
++++++++++++});
++++}
}|code-block|syntax|javascript|1670068|下面是一个与您的代码更接近的示例：|1670069|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["+%2B+t.id+%2B+"]+completed+",+startTime));
++++++++++++Observable<String>+imageUrl+=+getProductImage(t.getProductId())
++++++++++++++++++++.doOnCompleted(()+->+logTime("getProductImage["+%2B+t.id+%2B+"]+completed+",+startTime));

++++++++++++return+Observable.zip(reviews,+imageUrl,+(r,+u)+->+{
++++++++++++++++return+new+TileResponse(t,+r,+u);
++++++++++++}).doOnCompleted(()+->+logTime("zip["+%2B+t.id+%2B+"]+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_"+%2B+id);
++++}

++++private+static+void+logTime(String+message,+long+startTime)+{
++++++++System.out.println(message+%2B+"+=>+"+%2B+(System.currentTimeMillis()+-+startTime)+%2B+"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+{

++++}
}|1670070|这将输出以下内容：|1670071|Search+started++=>+65ms
Search+completed++=>+1094ms
getProductImage[1]+completed++=>+2095ms
getSellerReviews[2]+completed++=>+2095ms
getProductImage[3]+completed++=>+2095ms
zip[1]+completed++=>+2096ms
zip[2]+completed++=>+2096ms
getProductImage[2]+completed++=>+2096ms
getSellerReviews[1]+completed++=>+2096ms
zip[3]+completed++=>+2096ms
All+Tiles+Completed++=>+2097ms
getSellerReviews[3]+completed++=>+2097ms|1670072|我已经将每个IO调用模拟为耗时1000ms，因此很明显延迟在哪里，并且是并行发生的。它以毫秒为单位打印出进度。|1670073|这里的诀窍是flatMap合并异步调用，所以只要被合并的观察值是异步的，它们都将被并发执行。|1670074|如果像getProductImage(t.getProductId())这样的调用是同步的，那么它可以像这样异步进行:+getProductImage(t.getProductId()).subscribeOn(Schedulers.io).|1670075|以下是上面示例的重要部分，不包含所有的日志记录和样板类型：|1670076|++++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();|1670077|我希望这能帮到你。|1670078|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/ReactiveX/RxJava/pull/1716|1|https://github.com/Netflix/ReactiveLab/blob/master/reactive-lab-gateway/src/main/java/io/reactivex/lab/gateway/routes/RouteForDeviceHome.java^0|1Q|19|0|0|0|4|1|0|C|J|0|0|0|0|0|0|0|0|0|3|X|0|0|0|0^^$0|@$1|2|3|4|5|6|7|1K|8|@]|9|@$A|1L|B|1M|1|1N]]|C|$]]|$1|D|3|E|5|6|7|1O|8|@]|9|@$A|1P|B|1Q|1|1R]]|C|$]]|$1|F|3|G|5|6|7|1S|8|@$A|1T|B|1U|H|I]]|9|@]|C|$]]|$1|J|3|K|5|6|7|1V|8|@]|9|@]|C|$]]|$1|L|3|M|5|N|7|1W|8|@]|9|@]|C|$O|P]]|$1|Q|3|R|5|6|7|1X|8|@]|9|@]|C|$]]|$1|S|3|T|5|N|7|1Y|8|@]|9|@]|C|$O|P]]|$1|U|3|V|5|6|7|1Z|8|@]|9|@]|C|$]]|$1|W|3|X|5|N|7|20|8|@]|9|@]|C|$O|P]]|$1|Y|3|Z|5|6|7|21|8|@]|9|@]|C|$]]|$1|10|3|11|5|6|7|22|8|@]|9|@]|C|$]]|$1|12|3|13|5|6|7|23|8|@$A|24|B|25|H|I]]|9|@]|C|$]]|$1|14|3|15|5|6|7|26|8|@]|9|@]|C|$]]|$1|16|3|17|5|N|7|27|8|@]|9|@]|C|$O|P]]|$1|18|3|19|5|6|7|28|8|@]|9|@]|C|$]]|$1|1A|3|-4|5|6|7|29|8|@]|9|@]|C|$]]]|1B|$1C|$5|1D|1E|1F|C|$1G|1H]]|1I|$5|1D|1E|1F|C|$1G|1J]]]]

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: <a href="https://github.com/ReactiveX/RxJava/pull/1716">https://github.com/ReactiveX/RxJava/pull/1716</a> 

A good example of how to do this type of behavior and get parallel execution can be seen <a href="https://github.com/Netflix/ReactiveLab/blob/master/reactive-lab-gateway/src/main/java/io/reactivex/lab/gateway/routes/RouteForDeviceHome.java">here</a>.

In your example code it is unclear if <code>searchServiceClient</code> 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:

<pre><code>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 -&gt; {
 return getDataAsync(i);
 }).toBlocking().forEach(System.out::println);
 }

 private static void flatMapExampleSync() {
 Observable.range(0, 5).flatMap(i -&gt; {
 return getDataSync(i);
 }).toBlocking().forEach(System.out::println);
 }

 // artificial representations of IO work
 static Observable&lt;Integer&gt; getDataAsync(int i) {
 return getDataSync(i).subscribeOn(Schedulers.io());
 }

 static Observable&lt;Integer&gt; getDataSync(int i) {
 return Observable.create((Subscriber&lt;? super Integer&gt; s) -&gt; {
 // simulate latency
 try {
 Thread.sleep(1000);
 } catch (Exception e) {
 e.printStackTrace();
 }
 s.onNext(i);
 s.onCompleted();
 });
 }
}
</code></pre>

Following is an attempt at providing an example that more closely matches your code:

<pre><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&lt;Tile&gt; searchTile = getSearchResults("search term")
 .doOnSubscribe(() -&gt; logTime("Search started ", startTime))
 .doOnCompleted(() -&gt; logTime("Search completed ", startTime));

 Observable&lt;TileResponse&gt; populatedTiles = searchTile.flatMap(t -&gt; {
 Observable&lt;Reviews&gt; reviews = getSellerReviews(t.getSellerId())
 .doOnCompleted(() -&gt; logTime("getSellerReviews[" + t.id + "] completed ", startTime));
 Observable&lt;String&gt; imageUrl = getProductImage(t.getProductId())
 .doOnCompleted(() -&gt; logTime("getProductImage[" + t.id + "] completed ", startTime));

 return Observable.zip(reviews, imageUrl, (r, u) -&gt; {
 return new TileResponse(t, r, u);
 }).doOnCompleted(() -&gt; logTime("zip[" + t.id + "] completed ", startTime));
 });

 List&lt;TileResponse&gt; allTiles = populatedTiles.toList()
 .doOnCompleted(() -&gt; logTime("All Tiles Completed ", startTime))
 .toBlocking().single();
 }

 private static Observable&lt;Tile&gt; getSearchResults(String string) {
 return mockClient(new Tile(1), new Tile(2), new Tile(3));
 }

 private static Observable&lt;Reviews&gt; getSellerReviews(int id) {
 return mockClient(new Reviews());
 }

 private static Observable&lt;String&gt; getProductImage(int id) {
 return mockClient("image_" + id);
 }

 private static void logTime(String message, long startTime) {
 System.out.println(message + " =&gt; " + (System.currentTimeMillis() - startTime) + "ms");
 }

 private static &lt;T&gt; Observable&lt;T&gt; mockClient(T... ts) {
 return Observable.create((Subscriber&lt;? super T&gt; s) -&gt; {
 // 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 {

 }
}
</code></pre>

This outputs:

<pre><code>Search started =&gt; 65ms
Search completed =&gt; 1094ms
getProductImage[1] completed =&gt; 2095ms
getSellerReviews[2] completed =&gt; 2095ms
getProductImage[3] completed =&gt; 2095ms
zip[1] completed =&gt; 2096ms
zip[2] completed =&gt; 2096ms
getProductImage[2] completed =&gt; 2096ms
getSellerReviews[1] completed =&gt; 2096ms
zip[3] completed =&gt; 2096ms
All Tiles Completed =&gt; 2097ms
getSellerReviews[3] completed =&gt; 2097ms
</code></pre>

I 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 <code>getProductImage(t.getProductId())</code> 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:

<pre><code> Observable&lt;Tile&gt; searchTile = getSearchResults("search term");;

 Observable&lt;TileResponse&gt; populatedTiles = searchTile.flatMap(t -&gt; {
 Observable&lt;Reviews&gt; reviews = getSellerReviews(t.getSellerId());
 Observable&lt;String&gt; imageUrl = getProductImage(t.getProductId());

 return Observable.zip(reviews, imageUrl, (r, u) -&gt; {
 return new TileResponse(t, r, u);
 });
 });

 List&lt;TileResponse&gt; allTiles = populatedTiles.toList()
 .toBlocking().single();
</code></pre>

I hope this helps.

blocks|key|1670152|text|那些仍然是@+JDK+7的人，他们的IDE还不能自动检测JDK+8源代码，以及如何尝试@benjchristensen的上述精彩响应(和解释)，您可以使用这个不知羞耻地修改过的JDK+7代码。致敬@benjchristensen一个令人惊叹的解释和例子！|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1670153|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["+%2B+t.id+%2B+"]+completed+",+startTime);
++++++++++++++++++++++++++++++++++++}
++++++++++++++++++++++++});
++++++++++++++++Observable<String>+imageUrl+=+getProductImage(t.getProductId())
++++++++++++++++++++++++.doOnCompleted(new+Action0()
++++++++++++++++++++++++++++++++{

++++++++++++++++++++++++++++++++++++@Override
++++++++++++++++++++++++++++++++++++public+void+call()
++++++++++++++++++++++++++++++++++++{
++++++++++++++++++++++++++++++++++++++++logTime("getProductImage["+%2B+t.id+%2B+"]+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["+%2B+t.id+%2B+"]+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_"+%2B+id);
++++}

++++private+static+void+logTime(String+message,+long+startTime)
++++{
++++++++System.out.println(message+%2B+"+=>+"+%2B+(System.currentTimeMillis()+-+startTime)+%2B+"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
++++{

++++}
}+|code-block|syntax|javascript|1670154|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

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 !

<pre><code>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&lt;Tile&gt; 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&lt;TileResponse&gt; populatedTiles = searchTile.flatMap(new Func1&lt;Tile, Observable&lt;TileResponse&gt;&gt;()
 {

 @Override
 public Observable&lt;TileResponse&gt; call(final Tile t)
 {
 Observable&lt;Reviews&gt; reviews = getSellerReviews(t.getSellerId())
 .doOnCompleted(new Action0()
 {

 @Override
 public void call()
 {
 logTime("getSellerReviews[" + t.id + "] completed ", startTime);
 }
 });
 Observable&lt;String&gt; imageUrl = getProductImage(t.getProductId())
 .doOnCompleted(new Action0()
 {

 @Override
 public void call()
 {
 logTime("getProductImage[" + t.id + "] completed ", startTime);
 }
 });
 return Observable.zip(reviews, imageUrl, new Func2&lt;Reviews, String, TileResponse&gt;()
 {

 @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&lt;TileResponse&gt; allTiles = populatedTiles
 .toList()
 .doOnCompleted(new Action0()
 {

 @Override
 public void call()
 {
 logTime("All Tiles Completed ", startTime);
 }
 })
 .toBlocking()
 .single();
 }

 private static Observable&lt;Tile&gt; getSearchResults(String string)
 {
 return mockClient(new Tile(1), new Tile(2), new Tile(3));
 }

 private static Observable&lt;Reviews&gt; getSellerReviews(int id)
 {
 return mockClient(new Reviews());
 }

 private static Observable&lt;String&gt; getProductImage(int id)
 {
 return mockClient("image_" + id);
 }

 private static void logTime(String message, long startTime)
 {
 System.out.println(message + " =&gt; " + (System.currentTimeMillis() - startTime) + "ms");
 }

 private static &lt;T&gt; Observable&lt;T&gt; mockClient(final T... ts)
 {
 return Observable.create(new Observable.OnSubscribe&lt;T&gt;()
 {

 @Override
 public void call(Subscriber&lt;? super T&gt; 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
 {

 }
} 
</code></pre>

I need some help in implementing parallel asynchronous calls in RxJava. I have picked up a simple use case wherein the FIRST call fetches (rather searches) a list of products (Tile) to be displayed. The subsequent calls go out and fetch (A) REVIEWS and (B) PRODUCT IMAGES

After several attempts I got to this place. 

<pre><code> 1 Observable&lt;Tile&gt; searchTile = searchServiceClient.getSearchResults(searchTerm);
 2 List&lt;Tile&gt; allTiles = new ArrayList&lt;Tile&gt;();
 3 ClientResponse response = new ClientResponse();

 4 searchTile.parallel(oTile -&gt; {
 5 return oTile.flatMap(t -&gt; {
 6 Observable&lt;Reviews&gt; reviews = reviewsServiceClient.getSellerReviews(t.getSellerId());
 7 Observable&lt;String&gt; imageUrl = reviewsServiceClient.getProductImage(t.getProductId());

 8 return Observable.zip(reviews, imageUrl, (r, u) -&gt; {
 9 t.setReviews(r);
10 t.setImageUrl(u);

11 return t;
12 });

13 });
14 }).subscribe(e -&gt; {
15 allTiles.add((Tile) e);
16 });
</code></pre>

Line 1: goes out and fetches the product (Tile) to be displayed

Line 4: We take the list of the Observable and SHARD it to fetch reviews and imageUrls

Lie 6,7: Fetch the Observable review and Observable url

Line 8: Finally the 2 observables are zipped up to return an updated Observable

Line 15: finally line 15 collates all the individual products to be displayed in a collection which can be returned back to the calling layer

While the Observable has been sharded and in our tests run over 4 different threads; fetching of reviews and images seems to be one after another. I suspect that the zip step on line 8 is basically causing the sequential invocation of the the 2 observables (reviews and url).

<img src="https://i.stack.imgur.com/YVERX.png" alt="enter image description here">

Does this group have any suggestion to parallely fetch reiews and image urls. In essence the waterfall chart attached above should look more vertically stacked. The calls to reviews and images should be in parallel

thanks
anand raman

RxJava Fetching Observables In Parallel

我需要一些在RxJava中实现并行异步调用的帮助。我选择了一个简单的用例，其中第一个调用获取(而不是搜索)要显示的产品(Tile)列表。随后的调用将获取(A)评论和(B)产品图像经过几次尝试，我来到了这个地方。 1 Observable<Tile> searchTile = searchServiceClient...

问RxJava并行获取可观测数据
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问RxJava并行获取可观测数据EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问RxJava并行获取可观测数据
EN