如何使用 rxjava 放置异步改造调用。我必须异步调用 100 多个

Question

这是我一直在处理的代码示例

items包含100个元素，因此使用同步调用获取数据会占用大量时间。有人可以建议一种方法来提高此操作的速度，以便花费更少的时间。目前这需要 15-20 秒来执行。我是 rxjava 的新手，所以如果可能的话请提供这个问题的详细解决方案。 dataResponses 包含 100 个项目中每个项目的 RouteDistance 对象。

for(int i = 0 ; i<items.size();i++){

    Map<String, String> map2 = new HashMap<>();

    map2.put("units", "metric");
    map2.put("origin", currentLocation.getLatitude()+","+currentLocation.getLongitude());
    map2.put("destination", items.get(i).getPosition().get(0)+","+items.get(i).getPosition().get(1));
    map2.put("transportMode", "car");
    requests.add(RetrofitClient4_RouteDist.getClient().getRouteDist(map2));
}

Observable.zip(requests,  new Function<Object[], List<RouteDist>>() {
    @Override
    public List<RouteDist> apply(Object[] objects) throws Exception {
        Log.i("onSubscribe", "apply: " + objects.length);
        List<RouteDist> dataaResponses = new ArrayList<>();
        for (Object o : objects) {
            dataaResponses.add((RouteDist) o);
        }
        return dataaResponses;
    }
})
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeOn(Schedulers.io())
        .subscribe(
                new Consumer<List<RouteDist>>() {
                    @Override
                    public void accept(List<RouteDist> dataaResponses) throws Exception {
                        Log.i("onSubscribe", "YOUR DATA IS HERE: "+dataaResponses.toString());
                        recyclerViewAdapter_profile = new RecyclerViewAdapter_Profile(items,dataaResponses);
                        recyclerView.setAdapter(recyclerViewAdapter_profile);
                    }
                },

                new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable e) throws Exception {
                        Log.e("onSubscribe", "Throwable: " + e);
                    }
                });

Answer 1

API

interface Client {
    Observable<RouteDist> routeDist();
}

final class RouteDist {

}


final class ClientImpl implements Client {
    @Override
    public Observable<RouteDist> routeDist() {
        return Observable.fromCallable(() -> {
            // with this log, you see, that each subscription to an Observable is executed on the ThreadPool
            // Log.e("---------------------", Thread.currentThread().getName());
            return new RouteDist();
        });
    }
}

通过 subscribeOn 应用线程

final class ClientProxy implements Client {
    private final Client api;
    private final Scheduler scheduler;

    ClientProxy(Client api, Scheduler scheduler) {
        this.api = api;
        this.scheduler = scheduler;
    }

    @Override
    public Observable<RouteDist> routeDist() {
        // apply #subscribeOn in order to move subscribeAcutal call on given Scheduler
        return api.routeDist().subscribeOn(scheduler);
    }
}

Android测试

@Test
public void name() {
    // CachedThreadPool, in order to avoid creating 100-Threads or more. It is always a good idea to use own Schedulers (e.g. Testing)
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(0, 10,
            60L, TimeUnit.SECONDS,
            new SynchronousQueue<>());

    // wrap real client with Proxy, in order to move the subscribeActual call to the ThreadPool
    Client client = new ClientProxy(new ClientImpl(), Schedulers.from(threadPool));

    List<Observable<RouteDist>> observables = Arrays.asList(client.routeDist(), client.routeDist(), client.routeDist());

    TestObserver<List<RouteDist>> test = Observable.zip(observables, objects -> {
        return Arrays.stream(objects).map(t -> (RouteDist) t).collect(Collectors.toList());
    })
            .observeOn(AndroidSchedulers.mainThread())
            .test();

    test.awaitCount(1);

    // verify that onNext in subscribe is called in Android-EventLoop
    assertThat(test.lastThread()).isEqualTo(Looper.getMainLooper().getThread());
    // verify that 3 calls were made and merged into one List
    test.assertValueAt(0, routeDists -> {
        assertThat(routeDists).hasSize(3);
        return true;
    });
}

进一步阅读：

http://tomstechnicalblog.blogspot.de/2016/02/rxjava-understanding-observeon-and.html

注意：不建议同时调用 API 100 次。此外，当使用 Zip 时，当您拥有足够大的 ThreadPool 时，这就是实际发生的情况。当一个 API-call 超时时，一个 onError 可能会为这个 API-calls 发出。 onError 将进一步传播给订阅者。您不会得到任何结果，即使仅在 API 调用失败。建议使用一些 onErrorResumeNext 或其他一些错误处理运算符，以确保一个 API 调用不会取消整体结果。

如何使用 rxjava 放置异步改造调用。我必须异步调用 100 多个

How do I place Asynchronous Retrofit calls using rxjava. I have to place over a 100 calls asynchronously

java

android

rx-java

retrofit2

rx-java2