基于RxJava2.0+Retrofit2.0超大文件分块（分片）上传（带进度）

本文主要是介绍基于RxJava2.0+Retrofit2.0超大文件分块（分片）上传（带进度），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

当上传文件过大时，直接上传会增加服务器负载，负载过高会导致机器无法处理其他请求及操作，甚至导致宕机，这时分块上传是最佳选择。本次简单介绍两种分块上传文件方案（暂不考虑文件安全相关问题，如签名）：

方案一： 客户端把单个文件切割成若干块，依次上传，最后由服务端合并
方案二： 客户端读取文件 offset 到 offset+chunk的文件块上传，每次上传完返回新的offset（或者每次上传后，本地保存最新的offset），客户端更新offset值并继续下一次上传，本文为了测试方便，最新的偏移量本地计算好即不通过接口返回最新偏移量

文件分块上传与断点下载相比，断点下载的进度保存在客户端（使用数据库保存），分块上传的进度保存在服务端并由服务端合并

效果图：

两种方案对比，方案二上传时间更短，因为方案一基于文件操作，方案二基于内存操作，所以方案二上传时间更短

方案一

客服端：

    public void uploadTest1(View view) {countUploadSize = 0;clearDir();totalSize = file.length();//分割文件并组装数据上传，文件分割数量根据实际情况调整List<File> cut = FileUtils.split(file.getAbsolutePath(), 10);int totalChunk = cut.size();FileChunkReq[] items = new FileChunkReq[cut.size()];for (int i = 0; i < cut.size(); i++) {FileChunkReq fileChunkReq = new FileChunkReq();fileChunkReq.file = cut.get(i);fileChunkReq.fileName = file.getName();fileChunkReq.fileSize = totalSize;fileChunkReq.chunkNum = i + 1;fileChunkReq.totalChunk = totalChunk;fileChunkReq.uuid = UUID;items[i] = fileChunkReq;}startTime = System.currentTimeMillis();Observable.fromArray(items).concatMap(new Function<FileChunkReq, ObservableSource<ResponseResult>>() {@Overridepublic ObservableSource<ResponseResult> apply(FileChunkReq fcq) throws Exception {return upload1(fcq);}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<ResponseResult>() {@Overridepublic void accept(ResponseResult responseResult) throws Exception {showMsg("分块上传方案1responseResult：" + responseResult.toString());}}, new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {showMsg("分块上传方案1异常:" + throwable.getMessage());}}, new Action() {@Overridepublic void run() throws Exception {showMsg("分块上传方案1完毕");}});}

接口调用

/*** 分块上传1** @param fileChunkReq* @return*/private Observable<ResponseResult> upload1(FileChunkReq fileChunkReq) {//"application/octet-stream"//"multipart/form-data"FileProgressRequestBody filePart = new FileProgressRequestBody(fileChunkReq.file, "application/octet-stream", new FileProgressRequestBody.ProgressListener() {@Overridepublic void progress(long upload, long totalUpload) {synchronized (obj) {showProgress(upload);Log.w(TAG, "上传方案1,countUploadSize:" + countUploadSize + ",percent:" + percent);}}});final MultipartBody requestBody = new MultipartBody.Builder().setType(MultipartBody.FORM).addFormDataPart("file", fileChunkReq.file.getName(), filePart).addFormDataPart("fileName", fileChunkReq.fileName).addFormDataPart("fileSize", String.valueOf(fileChunkReq.fileSize)).addFormDataPart("uuid", fileChunkReq.uuid) //可用于持久化，作为文件存放目录，验证重复上传.addFormDataPart("chunkNum", String.valueOf(fileChunkReq.chunkNum)).addFormDataPart("totalChunk", String.valueOf(fileChunkReq.totalChunk)).build();Observable<ResponseResult> observable = uploadApi.upload1(requestBody);return observable;}

方案二

客服端：

 public void uploadTest2(View view) {countUploadSize = 0;offset = 0;long length = file.length();totalSize = length;startTime = System.currentTimeMillis();Observable.create(new ObservableOnSubscribe<FileChunkReq>() {@Overridepublic void subscribe(ObservableEmitter<FileChunkReq> emitter) throws Exception {int blockSize = 1024 * 1024; //1Mwhile (offset < totalSize) {//基于偏移量获取块大小，blockSize 可根据服务端调整byte[] block = FileUtils.getBlock(offset, file, blockSize);if (block != null) {FileChunkReq fileChunkReq = new FileChunkReq();fileChunkReq.fileByte = block;fileChunkReq.fileName = file.getName();fileChunkReq.fileSize = totalSize;fileChunkReq.offset = offset;fileChunkReq.uuid = UUID;emitter.onNext(fileChunkReq);//下一个偏移量offset += block.length;}}emitter.onComplete();}}).flatMap(new Function<FileChunkReq, ObservableSource<ResponseResult>>() {@Overridepublic ObservableSource<ResponseResult> apply(FileChunkReq fileChunkReq) throws Exception {return upload2(fileChunkReq);}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<ResponseResult>() {@Overridepublic void accept(ResponseResult responseResult) throws Exception {showMsg("分块上传方案2responseResult：" + responseResult.toString());}}, new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {showMsg("分块上传方案2异常:" + throwable.getMessage());}}, new Action() {@Overridepublic void run() throws Exception {showMsg("分块上传方案2完毕");}});}

接口调用

/*** 分块上传2** @param fileChunkReq* @return*/private ObservableSource<ResponseResult> upload2(FileChunkReq fileChunkReq) {ByteProgressRequestBody filePart = new ByteProgressRequestBody(fileChunkReq.fileByte, "application/octet-stream", new FileProgressRequestBody.ProgressListener() {@Overridepublic void progress(long upload, long totalUpload) {synchronized (obj) {showProgress(upload);Log.w(TAG, "上传方案2,countUploadSize:" + countUploadSize + ",percent:" + percent);}}});final MultipartBody requestBody = new MultipartBody.Builder().setType(MultipartBody.FORM).addFormDataPart("file", fileChunkReq.fileName, filePart).addFormDataPart("offset", String.valueOf(fileChunkReq.offset)).addFormDataPart("fileName", fileChunkReq.fileName).addFormDataPart("fileSize", String.valueOf(fileChunkReq.fileSize)).addFormDataPart("uuid", fileChunkReq.uuid).build();Observable<ResponseResult> observable = uploadApi.upload2(requestBody);return observable;}