RxJava并行获取可观测数据
RxJava并行获取可观测数据
提问于 2014-10-08 12:04:54
我需要一些在RxJava中实现并行异步调用的帮助。我选择了一个简单的用例，其中第一个调用获取(而不是搜索)要显示的产品(Tile)列表。随后的调用将获取(A)评论和(B)产品图像
经过几次尝试，我来到了这个地方。
1 Observable<Tile> searchTile = searchServiceClient.getSearchResults(searchTerm);
2 List<Tile> allTiles = new ArrayList<Tile>();
3 ClientResponse response = new ClientResponse();
4 searchTile.parallel(oTile -> {
5 return oTile.flatMap(t -> {
6 Observable<Reviews> reviews = reviewsServiceClient.getSellerReviews(t.getSellerId());
7 Observable<String> imageUrl = reviewsServiceClient.getProductImage(t.getProductId());
8 return Observable.zip(reviews, imageUrl, (r, u) -> {
9 t.setReviews(r);
10 t.setImageUrl(u);
11 return t;
12 });
13 });
14 }).subscribe(e -> {
15 allTiles.add((Tile) e);
16 });第1行:获取要显示的产品(Tile)
第4行:我们获取可观察对象的列表，并将其分片以获取评论和imageUrls
6,7:获取可观察的评论和可观察的url
第8行:最后将两个可观察对象压缩，以返回更新后的可观察对象
第15行:最后，第15行整理了集合中要显示的所有单个产品，该集合可以返回给调用层
虽然观察点已经被切分，并且在我们的测试中运行了4个不同的线程；但获取评论和图像似乎是一个接一个的。我怀疑第8行上的zip步骤基本上导致了对两个观察值(review和url)的顺序调用。
这个小组是否有任何建议来并行获取评论和图像urls。本质上，上面附加的瀑布图应该看起来更垂直堆叠。对评论和图片的调用应该是并行的
谢谢阿南德·拉曼
回答 2
Stack Overflow用户
回答已采纳
发布于 2014-10-08 14:09:26
事实证明，并行运算符在几乎所有用例中都是一个问题，并且没有完成大多数人对它的期望，因此在1.0.0.rc.4版本中删除了它：https://github.com/ReactiveX/RxJava/pull/1716
here是一个很好的例子，说明了如何执行这种类型的行为并获得并行执行。
在您的示例代码中，不清楚searchServiceClient是同步的还是异步的。它会稍微影响如何解决问题，就好像它已经是异步的一样，不需要额外的调度。如果需要同步额外调度。
首先，这里有一些简单的示例，展示了同步和异步行为：
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();
});
}
}下面是一个与您的代码更接近的示例：
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 {
}
}这将输出以下内容：
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我已经将每个IO调用模拟为耗时1000ms，因此很明显延迟在哪里，并且是并行发生的。它以毫秒为单位打印出进度。
这里的诀窍是flatMap合并异步调用，所以只要被合并的观察值是异步的，它们都将被并发执行。
如果像getProductImage(t.getProductId())这样的调用是同步的，那么它可以像这样异步进行: getProductImage(t.getProductId()).subscribeOn(Schedulers.io).
以下是上面示例的重要部分，不包含所有的日志记录和样板类型：
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();我希望这能帮到你。
Stack Overflow用户
发布于 2015-01-27 09:13:05
那些仍然是@ JDK 7的人，他们的IDE还不能自动检测JDK 8源代码，以及如何尝试@benjchristensen的上述精彩响应(和解释)，您可以使用这个不知羞耻地修改过的JDK 7代码。致敬@benjchristensen一个令人惊叹的解释和例子！
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
{
}
} 页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/26249030
复制相关文章
相似问题