我如何使用rxjava进行异步改造调用。我必须异步拨打100个以上的电话

Question

这里是我一直在研究的代码示例

项目包含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);
    }
}

AndroidTest

@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调用超时时，可能会为此API调用发出onError。 onError将进一步传播给订户。即使仅API调用失败，也不会获得任何结果。建议确保使用一些onErrorResumeNext或其他错误处理运算符，以确保一个API调用不会取消总体结果。