使用OkHttp，Okio和RxJava下载文件 [英] Downloading files using OkHttp, Okio and RxJava

问题描述

我正在尝试使用OkHttp下载文件，并使用Okio写入磁盘.我也为此过程创建了一个可观察的rx.它正在工作，但是它比我以前使用的(库什的离子库)慢得多.

I'm trying to download files using OkHttp and writing to disk with Okio. Also I've created an rx observable for this process. It is working, however it is noticeably slower than what I had previously used (Koush's Ion library).

这是我创建可观察对象的方法:

Here's how I create the observable:

public Observable<FilesWrapper> download(List<Thing> things) {
    return Observable.from(things)
        .map(thing -> {
            File file = new File(getExternalCacheDir() + File.separator + thing.getName());

            if (!file.exists()) {
                Request request = new Request.Builder().url(thing.getUrl()).build();
                Response response;
                try {
                    response = client.newCall(request).execute();
                    if (!response.isSuccessful()) new IOException();
                    else {
                        BufferedSink sink = Okio.buffer(Okio.sink(file));
                        sink.writeAll(response.body().source());
                        sink.close();
                    }
                } catch (IOException e) {
                    new IOException();
                }
            }

            return file;
        })
        .toList()
        .map(files -> new FilesWrapper(files);
}

有人知道是什么原因导致速度慢，或者我是否错误地使用了操作员?

Does anyone know what could be causing the slow speed, or if I am using the operators incorrectly?

推荐答案

使用flatMap而不是map将允许您并行执行下载:

Using flatMap instead of map will allow you to execute the downloads in parallel:

public Observable<FilesWrapper> download(List<Thing> things) {
    return Observable.from(things)
            .flatMap(thing -> {
                File file = new File(getExternalCacheDir() + File.separator + thing.getName());
                if (file.exists()) {
                    return Observable.just(file);
                }

                final Observable<File> fileObservable = Observable.create(sub -> {
                    if (sub.isUnsubscribed()) {
                        return;
                    }

                    Request request = new Request.Builder().url(thing.getUrl()).build();

                    Response response;
                    try {
                        response = client.newCall(request).execute();
                        if (!response.isSuccessful()) { throw new IOException(); }
                    } catch (IOException io) {
                        throw OnErrorThrowable.from(OnErrorThrowable.addValueAsLastCause(io, thing));
                    }

                    if (!sub.isUnsubscribed()) {
                        try (BufferedSink sink = Okio.buffer(Okio.sink(file))) {
                            sink.writeAll(response.body().source());
                        } catch (IOException io) {
                            throw OnErrorThrowable.from(OnErrorThrowable.addValueAsLastCause(io, thing));
                        }
                        sub.onNext(file);
                        sub.onCompleted();
                    }

                });
                return fileObservable.subscribeOn(Schedulers.io());
            }, 5)
            .toList()
            .map(files -> new FilesWrapper(files));
}

我们使用flatMap上的maxConcurrent限制每个订阅者的并发请求数.

We limit the number of simultaneous requests per subscriber using the maxConcurrent on flatMap.

这篇关于使用OkHttp，Okio和RxJava下载文件的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！