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，服务端部分处理形如：
链接：https://juejin.cn/post/7183887127992598585
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
