RxJava串行执行任务

Posted 2022-11-29 白玉梁

RxJava是什么不在多说，上一篇讲过RxJava的并发请求实现：RxJava的并发，这一篇讲RxJava串行的实现，主要用到contact和contactArray操作符，直接上代码：

        Observable<Integer> task1 = Observable.create(emitter -> 
            //...
            //这里执行一些任务
            //执行完成后发射到下一个任务并通知本任务完成
            emitter.onNext(1);
            emitter.onComplete();
        );
        Observable<Integer> task2 = Observable.create(emitter -> 
            //...
            //这里执行一些任务
            //执行完成后发射到下一个任务并通知本任务完成
            emitter.onNext(2);
            emitter.onComplete();
        );
        Observable<Integer> task3 = Observable.create(emitter -> 
            //...
            //这里执行一些任务
            //执行完成后发射到下一个任务并通知本任务完成
            emitter.onNext(3);
            emitter.onComplete();
        );

使用contact执行：

        Observable.concat(task1,task2,task3)
                .subscribeOn(androidSchedulers.mainThread())
                .subscribe(index -> 
                    LogUtil.e("index>>" + index);
                );

执行结果：

这是一个最原始最基础的例子，非常简单了！

contact和contactArray的区别是：contact最多只能执行三个串行任务，而contactArray可以执行多个，并且contactArray是可以直接传入数组的，这样就为一些动态的数量未知的任务提供了方便，下面我们来模拟串行执行下载任务（某些特殊需求，不能用并发，只能用串行）：

       String[] filePaths = new String[]//文件下载地址
                "https://imtt.dd.qq.com/sjy.10001/16891/apk/9A8E683EEB620438BA39F1BCB6EF1262.apk",
                "https://imtt.dd.qq.com/sjy.10001/16891/apk/9A8E683EEB620438BA39F1BCB6EF1262.apk",
                "https://imtt.dd.qq.com/sjy.10001/16891/apk/9A8E683EEB620438BA39F1BCB6EF1262.apk",
        ;
        Observable[] tasks = new Observable[filePaths.length];
        for (int i = 0; i < filePaths.length; i++) 
            int finalI = i;
            tasks[i] = Observable.create(emitter -> 
                new DownLoadUtils().download(filePaths[finalI],
                        new File(File10Utils.getDownloadDirPath(mContext), File10Utils.md5("" + System.currentTimeMillis()) + ".apk"),
                        new DownLoadUtils.DownLoadListener() 
                            @Override
                            public void progress(DownLoadUtils.ProgressInfo progressInfo) 
                                LogUtil.e("正在下载第" + (finalI + 1) + "个文件:" + progressInfo.getProgress());
                            

                            @Override
                            public void OnSuccess() 
                                LogUtil.e("第" + (finalI + 1) + "个文件下载成功");
                                emitter.onNext(finalI + 1);
                                emitter.onComplete();
                            

                            @Override
                            public void OnFailure(String errorMsg) 
                                LogUtil.e("第" + (finalI + 1) + "个文件下载失败");
                                emitter.onNext(finalI + 1);
                                emitter.onComplete();
                            
                        );
            );
        

注意，这模拟了一个动态任务数，任务数量不定，每个下载任务都是异步的，而subscribe则在主线程，执行任务：

   Observable.concatArray(tasks)
                    .subscribeOn(AndroidSchedulers.mainThread())
                    .subscribe(index -> 
                        LogUtil.e("index>>" + index);
                    );

执行结果：

DownLoadUtils是自己封装的一个下载工具类，很简单，支持取消任务，支持下载进度，支持下载回调！下载