RxJava并行获取可观察对象

ltqd579y  于 2022-12-25  发布在  Java
关注(0)|答案(2)|浏览(157)

我需要一些帮助来实现RxJava中的并行异步调用。我已经选择了一个简单的用例，其中第一个调用获取（而不是搜索）要显示的产品列表（Tile）。随后的调用输出并获取（A）REVIEWS和（B）PRODUCT IMAGES
经过几次尝试，我终于到了这个地方。
第1行：取出要显示的产品（平铺）
第4行：我们获取Observable的列表并将其SHARD以获取评论和imageUrls
谎言6，7：获取可观察评论和可观察url
第8行：最后，压缩2个Observable以返回更新后的Observable
第15行：最后，第15行整理要在集合中显示的所有单个产品，该集合可以返回到调用层
虽然观察已被分割，并在我们的测试运行在4个不同的线程;评论和图片的获取似乎是一个接一个的。我怀疑第8行的zip步骤基本上导致了2个可观察对象（评论和URL）的顺序调用。

这个小组有没有建议并行获取评论和图片的网址？本质上，上面的瀑布图应该看起来更垂直堆叠。对评论和图片的调用应该是并行的
谢谢阿南德·拉曼

Java

2条答案

按热度按时间
lx0bsm1f

lx0bsm1f1#

并行运算符被证明是几乎所有用例的一个问题，并且不能实现大多数人的期望，因此在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合并异步调用，因此只要被合并的Observable是异步的，它们就都将并发执行。
如果像getProductImage(t.getProductId())这样的调用是同步的，则可以像下面这样使其异步：获取产品映像（t.获取产品ID（））.订阅打开（调度程序. 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();
赞(0）分享  回复(0）举报  2022-12-25
z9ju0rcb

z9ju0rcb2#

那些仍然是@ JDK 7的人，那些IDE还没有自动检测JDK 8源代码的人，以及那些想尝试@benjchristensen上面精彩回应（和解释）的人，都可以使用这个无耻的重复，JDK 7，代码。

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
    {

    }
}
赞(0）分享  回复(0）举报  2022-12-25
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com