RxJava2 zip、flatmap 批量处理图片裁剪、上传

版本依赖：

implementation "io.reactivex.rxjava2:rxjava:2.2.14"
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'

Observable.zip批量裁剪图片示例

    class Picture {
        public String localPath;    //本地路径
        public String cutPath;      //裁剪路径
        public int index;           //图片顺序
    }

List<Picture> pictures = new ArrayList<>();

for (int i = 0; i < 3; i++) {
            Picture picture = new Picture();
            picture.localPath = "www.baidu.com/picture/" + i;
            pictures.add(picture);
        }
        Observable.zip(getObservableList(), new Function<Object[], Object>() {
            @Override
            public Object apply(Object[] objects) {
                return "";
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {
                Log.e(TAG, JSON.toJSONString(pictures));
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });

List<Observable<Picture>> getObservableList() {
        List<Observable<Picture>> observables = new ArrayList<>();
        for (int i = 0; i < pictures.size(); i++) {
            final int finalI = i;
            Observable<Picture> observable = Observable.create(new ObservableOnSubscribe<Picture>() {
                @Override
                public void subscribe(ObservableEmitter<Picture> emitter) throws InterruptedException {
                    Thread.sleep(1000);//模拟图片裁剪耗时
                    Picture picture = pictures.get(finalI);
                    picture.cutPath = picture.localPath + "_cut";
                    emitter.onNext(picture);
                    emitter.onComplete();
                }
            })
          .subscribeOn(Schedulers.io());//此处不加所有执行变成串行
            observables.add(observable);
        }
        return observables;
    }

2019-11-05 10:36:35.692 30420-30420/? E/MainActivity: [{"cutPath":"www.baidu.com/picture/0_cut","localPath":"www.baidu.com/picture/0"},{"cutPath":"www.baidu.com/picture/1_cut","localPath":"www.baidu.com/picture/1"},{"cutPath":"www.baidu.com/picture/2_cut","localPath":"www.baidu.com/picture/2"}]

去除.subscribeOn(Schedulers.io())将变成串行执行,输出3次log

2019-11-05 10:48:45.571 31906-31906/com.code.layoutanalyze E/MainActivity: [{"cutPath":"www.baidu.com/picture/0_cut","localPath":"www.baidu.com/picture/0"},{"localPath":"www.baidu.com/picture/1"},{"localPath":"www.baidu.com/picture/2"}]
2019-11-05 10:48:46.516 31906-31906/com.code.layoutanalyze E/MainActivity: [{"cutPath":"www.baidu.com/picture/0_cut","localPath":"www.baidu.com/picture/0"},{"cutPath":"www.baidu.com/picture/1_cut","localPath":"www.baidu.com/picture/1"},{"localPath":"www.baidu.com/picture/2"}]
2019-11-05 10:48:47.521 31906-31906/com.code.layoutanalyze E/MainActivity: [{"cutPath":"www.baidu.com/picture/0_cut","localPath":"www.baidu.com/picture/0"},{"cutPath":"www.baidu.com/picture/1_cut","localPath":"www.baidu.com/picture/1"},{"cutPath":"www.baidu.com/picture/2_cut","localPath":"www.baidu.com/picture/2"}]

Observable.flatmap变换数据结构并上传

    class Picture {
        public String localPath;    //本地路径
        public String cutPath;      //裁剪路径
        public int index;           //图片顺序
    }

    class PictureServer {
        public String localPath;    //本地路径
        public String serverPath;   //服务端路径
        public int index;           //图片顺序
    }

    private PictureServer transform(Picture picture) {
        PictureServer pictureServer = new PictureServer();
        pictureServer.localPath = picture.localPath;
        pictureServer.index = picture.index;
        return pictureServer;
    }

for (int i = 0; i < 3; i++) {
            Picture picture = new Picture();
            picture.localPath = "www.baidu.com/picture/" + i;
            picture.index = i;
            pictures.add(picture);
        }
        Observable.fromIterable(pictures).flatMap(new Function<Picture, ObservableSource<PictureServer>>() {
            @Override
            public ObservableSource<PictureServer> apply(final Picture picture) throws Exception {
                Observable<PictureServer> observable = Observable.create(new ObservableOnSubscribe<PictureServer>() {
                    @Override
                    public void subscribe(ObservableEmitter<PictureServer> emitter) throws Exception {
                        PictureServer pictureServer = transform(picture);   //转换数据
                        Thread.sleep(1000); //模拟图片上传
                        pictureServer.serverPath = picture.localPath + "_server";
                        emitter.onNext(pictureServer);
                        emitter.onComplete();
                    }
                }).subscribeOn(Schedulers.io());
                return observable;
            }
        }).toList().toObservable().subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<List<PictureServer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }
                    @Override
                    public void onNext(List<PictureServer> objects) {
                        Log.e(TAG, JSON.toJSONString(objects));
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onComplete() {
                    }
                });

2019-11-05 11:08:05.778 32476-32476/com.code.layoutanalyze E/MainActivity: [{"index":0,"localPath":"www.baidu.com/picture/0","serverPath":"www.baidu.com/picture/0_server"},{"index":1,"localPath":"www.baidu.com/picture/1","serverPath":"www.baidu.com/picture/1_server"},{"index":2,"localPath":"www.baidu.com/picture/2","serverPath":"www.baidu.com/picture/2_server"}]

将flatmap改成concatMap或者去除subscribeOn(Schedulers.io())将变成串行上传.

RxJava自定义线程池

在一些低配机器上，如果过多进行并发很有可能出现OOM事件，需要根据手机配置或者内存状况自定义并发线程数量

//自定义线程池
        final Scheduler schedulers = Schedulers.from(Executors.newFixedThreadPool(2));
        Observable.fromIterable(pictures).flatMap(new Function<Picture, ObservableSource<PictureServer>>() {
            @Override
            public ObservableSource<PictureServer> apply(final Picture picture) throws Exception {
                Observable<PictureServer> observable = Observable.create(new ObservableOnSubscribe<PictureServer>() {
                    @Override
                    public void subscribe(ObservableEmitter<PictureServer> emitter) throws Exception {
                        PictureServer pictureServer = transform(picture);   //转换数据
                        Thread.sleep(1000); //模拟图片上传
                        Log.e(TAG,picture.index+"");
                        pictureServer.serverPath = picture.localPath + "_server";
                        emitter.onNext(pictureServer);
                        emitter.onComplete();
                    }
                }).subscribeOn(schedulers);//使用自定义线程池
                return observable;
            }
        })

2019-11-05 11:20:58.804 2753-2787/com.code.layoutanalyze E/MainActivity: 0
2019-11-05 11:20:58.804 2753-2788/com.code.layoutanalyze E/MainActivity: 1
2019-11-05 11:20:59.804 2753-2788/com.code.layoutanalyze E/MainActivity: 2
2019-11-05 11:20:59.868 2753-2753/com.code.layoutanalyze E/MainActivity: [{"index":0,"localPath":"www.baidu.com/picture/0","serverPath":"www.baidu.com/picture/0_server"},{"index":1,"localPath":"www.baidu.com/picture/1","serverPath":"www.baidu.com/picture/1_server"},{"index":2,"localPath":"www.baidu.com/picture/2","serverPath":"www.baidu.com/picture/2_server"}]