RxJava加Retrofit文件分段上传实现详解

时间:2023-02-03 Chavin 人气:0

前言

本文基于 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，服务端部分处理形如：

加载全部内容