RxJava加Retrofit文件分段上传实现详解
目录
- 前言
- 问题拆解
- 分块并发读取文件
- 文件片段上传
- 完整代码
前言
本文基于 RxJava 和 Retrofit 库,设计并实现了一种用于大文件分块上传的工具,并对其进行了全面的拆解分析。抛砖引玉,对同样有处理文件分块上传诉求的读者,可能会起到一定的启发作用。
文章主体由四部分构成:
- 首先分析问题,问题拆解为:多线程分段读取文件、构建和发出文件片段上传请求
- 基于 JDK 随机读取文件的类库,设计本地多线程分段读取文件的单元
- 基于 Retrofit 设计由文件片段构建上传的网络请求
- 从上述设计演变而来的完整代码实现
另外,在文章提供的完整代码中,还附了一段由 PHP 编写,用来接收多线程分段数据的服务端接口实现,其中处理了因客户端都线程上传片段,导致服务端接收的文件片段无序,故需在适当时机合并分块构成目标文件。
受限于笔者的开发经验与理论理解,文章的思路和代码难免可能有偏颇,对于有改进和优化的部分,欢迎大家讨论区提出。
问题拆解
要完成文件分段上传到服务端,第一步是分段读取本地文件。通常分段是为了多线程同时执行上传,提高设备计算和网络资源利用率,减少上传时间优化体验,这样即需要一个支持多线程的文件分段读取工具。由于文件可能超过设备内存大小,在读取这类超大文件时需要控制最大读取量防止内存溢出。此时文件已从磁盘数据转换为内存中的字节数据,只需要将这些内存数据传给服务端即可。这样问题被分成 3 个子问题:
- 分段读取文件到内存中
- 控制多线程数量
- 将文件片段传给服务端
问题 1 很好解决,利用 Java 的 RandomAccessFile
可对文件的随机读取的特性,即可按需读取文件片段到内存中。
问题 2 相对复杂一点,但如果有阅读过 JDK 中线程池源码的读者,就会发现这个问题的和控制线程池中线程数量其实是类似的。
问题 3 就不复杂了,Retrofit 基于 OKhttp ,OkHttp是很容易基于字节数组构建 multipart/form-data
请求的。
分块并发读取文件
根据上述对问题 1、2 的拆解,可将读取抽象为一个文件读取器,构建时传入文件对象和分段大小以及最大并发数,以及分段数据的回调。当外部启动读取时将根据文件大小和配置的分段大小构建若干个 Task 用于读取对应片段的数据。
public BlockReader(@NotNull File file, @NotNull BlockCallback callback, int poolSize, int blockSize) { mFile = file; mCallback = callback; mPoolSize = poolSize; mBlockSize = blockSize; } public void start(@Nullable BlockFilter filter) { Observable.empty().observeOn(Schedulers.computation()).doOnComplete(() -> { long length = mFile.length(); for (long offset = 0; offset < length; offset += mBlockSize) { if (null != filter && filter.ignore(offset)) { continue; } mQueue.offer(new ReadTask(offset)); } for (int i = 0; i < Math.min(mPoolSize, mQueue.size()); i++) { Observable.empty().observeOn(Schedulers.io()).doOnComplete(this::schedule).subscribe(); } }).subscribe(); }
多线程调度部分,可通过加锁和记录状态变量统计当前正运行的线程数,则可控制字节数组数,这样就相当于控制住了最大内存占用。
private void schedule() { if (mRunning.get() >= mPoolSize) { return; } ReadTask task; synchronized (mQueue) { if (mRunning.get() >= mPoolSize) { return; } task = mQueue.poll(); if (null != task) { mRunning.incrementAndGet(); } } if (null != task) { task.run(); } }
最后是文件随机读取,直接调用 RandomAccessFile
的 API 即可:
private class ReadTask implements Action { @Override public void run() { try (RandomAccessFile raf = new RandomAccessFile(mFile, RAF_MODE); ByteArrayOutputStream out = new ByteArrayOutputStream(mBlockSize)) { raf.seek(mOffset); byte[] buf = new byte[DEF_BLOCK_SIZE]; long cnt = 0; for (int bytes = raf.read(buf); bytes != -1 && cnt < mBlockSize; bytes = raf.read(buf)) { out.write(buf, 0, bytes); cnt += bytes; } out.flush(); mCallback.onFinished(mOffset, out.toByteArray()); } catch (IOException e) { mCallback.onFinished(mOffset, null); } finally { mRunning.decrementAndGet(); schedule(); } } }
文件片段上传
上传部分则使用 Retrofit 提供的注解和 OKHttp 的类库构建请求。但值得一提的是需要在磁盘IO线程同步完成网络IO,这样可以避免网络IO速度落后磁盘IO太多而导致任务堆积造成内存溢出。
public interface BlockUploader { @POST("test/upload.php") @Multipart Single<Response<ResponseBody>> upload(@Header("filename") String filename, @Header("total") long total, @Header("offset") long offset, @Part List<MultipartBody.Part> body); } private static void syncUpload(String fileName, long fileLength, long offset, byte[] bytes) { RequestBody data = RequestBody.create(MediaType.parse("application/octet-stream"), bytes); MultipartBody body = new MultipartBody.Builder() .addFormDataPart("file", fileName, data) .setType(MultipartBody.FORM) .build(); retrofit.create(BlockUploader.class).upload(fileName, fileLength, offset, body.parts()).subscribe(resp -> { if (resp.isSuccessful()) { System.out.println("✓ offset: " + offset + " upload succeed " + resp.code()); } else { System.out.println("✗ offset: " + offset + " upload failed " + resp.code()); } }, throwable -> { System.out.println("! offset: " + offset + " upload failed"); }); }
完整代码
为控制篇幅,完整代码请移步 Github,服务端部分处理形如:
以上就是RxJava加Retrofit文件分段上传示例的详细内容,更多关于RxJava Retrofit文件上传的资料请关注我们其它相关文章!