拆轮子-RxDownload2源码解析(三)

本文为博主原创文章，未经允许不得转载

造轮子者：Season_zlc

轮子用法请戳作者链接 ↑

前言

本文主要讲述 RxDownload2 的多线程断点下载技术。

断点下载技术前提

服务器必须支持按 byte-range 下载，也就是支持 Range: bytes=xxx-xxx 请求头。详见Http协议 rfc2616 - Range。

下载范围分割

很简单，先读取 Content-Length 响应头，获取文件大小，然后用文件大小除以线程数就可计算出每条线程的下载范围。

比如，假设文件大小是 100 bytes，下载线程数为 3。因为 100 / 3 = 33，所以：

线程 0 的下载范围是 0 ~32 即 [0 * 33 ~ (0 + 1) * 33 - 1]

线程 1 的下载范围是 33~65 即 [1 * 33 ~ (1 + 1) * 33 - 1]

线程 2 的下载范围是 66~99 即 [2 * 33 ~ 100 - 1]

上代码：

prepareDownload() [-> FileHelper.java]

    public void prepareDownload(File lastModifyFile, File tempFile, File saveFile,long fileLength, String lastModify)throws IOException, ParseException {// 将响应头中的上次修改时间转为 long 类型的 unix 时间戳，然后保存到文件中writeLastModify(lastModifyFile, lastModify);// 设置下载文件的大小、计算每条线程的下载范围并保存到 tempFile 中prepareFile(tempFile, saveFile, fileLength);}

prepareFile() [-> FileHelper.java]

    private void prepareFile(File tempFile, File saveFile, long fileLength)throws IOException {RandomAccessFile rFile = null;RandomAccessFile rRecord = null;FileChannel channel = null;try {rFile = new RandomAccessFile(saveFile, ACCESS);rFile.setLength(fileLength);//设置下载文件的长度 rRecord = new RandomAccessFile(tempFile, ACCESS);// 下载范围在文件中的记录方式：|start|end|start|end|start|end|...// 数据类型是 long，long类型在 java 中占 8 个字节，所以每个线程的下载范围都占 16 字节// 所以 tempFile 的长度 RECORD_FILE_TOTAL_SIZE = 16 * 线程数rRecord.setLength(RECORD_FILE_TOTAL_SIZE); //设置指针记录文件的大小// NIO 内存映射文件的方式读写二进制文件，速度更快channel = rRecord.getChannel();// 注意映射方式为读写MappedByteBuffer buffer = channel.map(READ_WRITE, 0, RECORD_FILE_TOTAL_SIZE);long start;long end;// 计算并保存每条线程的下载范围，计算方法同上面举的例子int eachSize = (int) (fileLength / maxThreads);for (int i = 0; i < maxThreads; i++) {if (i == maxThreads - 1) {start = i * eachSize;end = fileLength - 1;} else {start = i * eachSize;end = (i + 1) * eachSize - 1;}buffer.putLong(start);buffer.putLong(end);}} finally {closeQuietly(channel);closeQuietly(rRecord);closeQuietly(rFile);}}

读取下载范围

很简单，上面已经将每条线程的下载范围保存到了 tempFile 中，只要再从 tempFile 中按位置读出来就行了。

readDownloadRange() [-> FileHelper.java]

    public DownloadRange readDownloadRange(File tempFile, int i) throws IOException {RandomAccessFile record = null;FileChannel channel = null;try {// 入参 i 表示线程序号record = new RandomAccessFile(tempFile, ACCESS);channel = record.getChannel();MappedByteBuffer buffer = channel.map(READ_WRITE, i * EACH_RECORD_SIZE, (i + 1) * EACH_RECORD_SIZE);long startByte = buffer.getLong();long endByte = buffer.getLong();return new DownloadRange(startByte, endByte);} finally {closeQuietly(channel);closeQuietly(record);}}

注意 MappedByteBuffer buffer = channel.map(READ_WRITE, i * EACH_RECORD_SIZE, (i + 1) * EACH_RECORD_SIZE); 这句代码是有坑的，但是表现不出来，因为这里的文件打开方式为 READ_WRITE。要是改成 READ_ONLY 就有导致读取最后一条线程的下载范围时抛出IllegalArgumentException(代码静态检查工具 Fortify 提示要以合适的权限打开文件，我将其改为了 READ_ONLY ，发现了这一问题)。

错误原因：map() 方法的最后一个参数表示要映射的字节数，以只读方式打开时，若参数大小超过了文件剩余可读字节数，就会抛出 IllegalArgumentException。而以读写方式打开文件时，会自动扩展文件长度，所以不会抛出异常。

因为每段下载范围的长度都是 EACH_RECORD_SIZE = 16 bytes，所以，上述代码应修改为：
MappedByteBuffer buffer = channel.map(READ_WRITE, i * EACH_RECORD_SIZE, EACH_RECORD_SIZE);

Intellij IDEA 示例代码

自己写了个示例代码，测试了一下：

        RandomAccessFile file = new RandomAccessFile("temp.txt", "rw");file.setLength(48);FileChannel channel = file.getChannel();MappedByteBuffer buffer = channel.map(FileChannel.MapMode.READ_WRITE, 0, 48);for (int i = 0; i < 3; i++) {if (i == 2) {buffer.putLong(i * 33).putLong(99);} else {buffer.putLong(i * 33).putLong((i + 1) * 33 - 1);}}channel.close();RandomAccessFile file1 = new RandomAccessFile("temp.txt", "r");FileChannel channel1 = file1.getChannel();for (int i = 0; i < 3; i++) {MappedByteBuffer buffer1 = channel1.map(FileChannel.MapMode.READ_ONLY, i * 16, 16);System.out.println(String.format("long1: %d", buffer1.getLong()));System.out.println(String.format("long2: %d", buffer1.getLong()));}channel1.close();

给 Notepad++ 装个十六进制查看器，查看生成的 temp.txt 中的内容是否和我们代码写的一样：

temp.txt view in HEX

上面是十六进制，换算成十进制就是上面示例代码写的内容。

写下载文件

很简单，利用 RandomAccessFile 可从任意位置读写的属性，分别将每条线程下载的数据写到同一个文件的不同位置。

saveFile() [-> FileHelper.java]

    public void saveFile(FlowableEmitter<DownloadStatus> emitter, int i, File tempFile,File saveFile, ResponseBody response) {RandomAccessFile record = null;FileChannel recordChannel = null;RandomAccessFile save = null;FileChannel saveChannel = null;InputStream inStream = null;try {try {// 1.映射 tempFile 到内存中record = new RandomAccessFile(tempFile, ACCESS);recordChannel = record.getChannel();MappedByteBuffer recordBuffer = recordChannel.map(READ_WRITE, 0, RECORD_FILE_TOTAL_SIZE);// i 代表线程序号，startIndex 代表该线程下载范围的 start 字段在文件中的指针位置int startIndex = i * EACH_RECORD_SIZE;// start 表示该线程的起始下载位置long start = recordBuffer.getLong(startIndex);// 新建一个下载状态对象，用于发射下载进度DownloadStatus status = new DownloadStatus();// totalSize 代表文件总大小，也可以从 saveFile 中读出long totalSize = recordBuffer.getLong(RECORD_FILE_TOTAL_SIZE - 8) + 1;status.setTotalSize(totalSize);int readLen;byte[] buffer = new byte[2048];inStream = response.byteStream();save = new RandomAccessFile(saveFile, ACCESS);saveChannel = save.getChannel();while ((readLen = inStream.read(buffer)) != -1 && !emitter.isCancelled()) {MappedByteBuffer saveBuffer = saveChannel.map(READ_WRITE, start, readLen);saveBuffer.put(buffer, 0, readLen);// 成功下载一段数据后，将已下载位置写回 start 字段start += readLen;recordBuffer.putLong(startIndex, start);// 计算已下载字节数 = 文件长度 - 每条线程剩余未下载字节数status.setDownloadSize(totalSize - getResidue(recordBuffer));// 发射下载进度emitter.onNext(status);}// 发射下载完成emitter.onComplete();} finally {closeQuietly(record);closeQuietly(recordChannel);closeQuietly(save);closeQuietly(saveChannel);closeQuietly(inStream);closeQuietly(response);}} catch (IOException e) {emitter.onError(e);}}

总结

下载流程就不分析了，只要熟练使用下图所示两个快捷键，什么源码分析都是手到擒来：
RxDownload2 源码解析系列至此结束，虽然框架比较简单，但是还是有很多值得学习的东西。尤其是作者对 RxJava2 的使用，可以说非常之六了。他写的十篇 Rxjava2 教程也非常的通俗易懂，感兴趣的可以看一看。