java - 我如何与 rxAndroid 一起下载多个图像并在 OnNext 方法中获得单独的回调

Question

我想一起下载多张图片(在不同的线程中)，并希望为每个图片下载完成(在 onNext 方法中)获得一个单独的回调。

但是从这个代码图像是一个一个地下载，需要很多时间。如果这在 rxAndroid 中是不可能的。那我该怎么做呢？

public void downloadImages(Observer<String> ob){
    getUrlObservable(imageUrls).
            subscribeOn(Schedulers.io()).
            observeOn(AndroidSchedulers.mainThread()).subscribe(ob);
}

private Observable<String> getUrlObservable(List<String> imageUrls) {

    return getEachUrl(imageUrls).flatMap(new Function<String, ObservableSource<String>>() {
        @Override
        public ObservableSource<String> apply(String s) throws Exception {
            return saveImageForProduct(s);
        }
    });

}


private Observable<String> getEachUrl(List<String> imageUrls){
    return Observable.fromArray(imageUrls).flatMapIterable(new Function<List<String>, Iterable<? extends String>>() {
        @Override
        public Iterable<? extends String> apply(List<String> imageurls) throws Exception {
            return imageUrls;
        }
    });

}



   private Observable<String> saveImageForProduct(final String imageUrl) {

    return Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> emitter) throws Exception {
            FileDownloader.downloadFile(imageUrl, Constants.getFileName(imageUrl) + ".jpg");
            emitter.onNext(imageUrl);
            emitter.onComplete();
        }
    }).onErrorResumeNext(Observable.just(""));

}

我的 react 是这样的——

I/Thread: RxCachedThreadScheduler-1
I/next:https://cdn.shopify.com/s/files1.jpg
I/Thread: RxCachedThreadScheduler-1
I/next:https://cdn.shopify.com/s/files1.jpg

我想变成这样--

I/Thread: RxCachedThreadScheduler-1
I/Thread: RxCachedThreadScheduler-2
I/next:https://cdn.shopify.com/s/files1.jpg
I/next:https://cdn.shopify.com/s/files.jpg

Answer 1

创建一个包含下载 url 的列表并使用 flatMap 来实现它。

Observable
.from(imageList)
.flatMap(listItem -> 
    Observable
        .from(listItem)
        .doOnNext(item -> downloadImage(listItem.imageUrl))
        .subscribeOn(Schedulers.io())
)
.doOnError(t -> logError(t))
.subscribe();