OkDownload是一款多线程断点续传下载引擎，它的功能完整，性能高，可配置性高，可以注入自定义组件来修改下载策略、替换网络请求框架等等，而且在项目中已有成熟应用（英语流利说），是一个很不错的开源下载框架。
项目地址：https://github.com/lingochamp/okdownload
OkDownload的简单使用
OkDownload的使用非常简单：
1.引入该开源库：
implementation 'com.liulishuo.okdownload:okhttp:1.0.5' （提供okhttp连接，ps：如果使用的话，需要引入okhttp网络请求库）
implementation 'com.liulishuo.okdownload:okdownload:1.0.5' （下载核心库）
implementation 'com.liulishuo.okdownload:sqlite:1.0.5' （存储断点信息的数据库）
2.开始一个任务：
task = new DownloadTask.Builder(url, parentFile)
.setFilename(filename)
// 下载进度回调的间隔时间（毫秒）
.setMinIntervalMillisCallbackProcess(30)
// 任务过去已完成是否要重新下载
.setPassIfAlreadyCompleted(false)
.build();
//异步执行任务
task.enqueue(listener);
// 取消任务
task.cancel();
// 同步执行任务
task.execute(listener);
当然也可以同时异步执行多个任务
DownloadTask.enqueue(tasks, listener);
3.任务队列的构建、开始和停止
DownloadContext.Builder builder = new DownloadContext.QueueSet()
.setParentPathFile(parentFile)
.setMinIntervalMillisCallbackProcess(150)
.commit();
builder.bind(url1);
builder.bind(url2).addTag(key, value);
builder.bind(url3).setTag(tag);
builder.setListener(contextListener);
DownloadTask task = new DownloadTask.Builder(url4, parentFile)
.setPriority(10).build();
builder.bindSetTask(task);
DownloadContext context = builder.build();
context.startOnParallel(listener);
// stop
context.stop();
4.获取任务状态
Status status = StatusUtil.getStatus(task)
status = StatusUtil.getStatus(url, parentPath, null);
status = StatusUtil.getStatus(url, parentPath, filename);
boolean isCompleted = StatusUtil.isCompleted(task);
isCompleted = StatusUtil.isCompleted(url, parentPath, null);
isCompleted = StatusUtil.isCompleted(url, parentPath, filename);
Status completedOrUnknown = StatusUtil.isCompletedOrUnknown(task);
5.获取断点信息
// 注意:任务完成后，断点信息将会被删除
BreakpointInfo info = OkDownload.with().breakpointStore().get(id);
info = StatusUtil.getCurrentInfo(url, parentPath, null);
info = StatusUtil.getCurrentInfo(url, parentPath, filename);
// 断点信息将被缓存在任务对象中，即使任务已经完成了
info = task.getInfo();
6.设置任务监听
可以为任务设置五种不同类型的监听器，同时，也可以给任务和监听器建立1对1、1对多、多对1、多对多的关联。
给一个任务设置多种监听：
DownloadListener listener1 = new DownloadListener1();
DownloadListener listener2 = new DownloadListener2();
DownloadListener combinedListener = new DownloadListenerBunch.Builder()
.append(listener1)
.append(listener2)
.build();
DownloadTask task = new DownloadTask.build(url, file).build();
task.enqueue(combinedListener);
为多个任务动态设置监听：
UnifiedListenerManager manager = new UnifiedListenerManager();
DownloadListener listener1 = new DownloadListener1();
DownloadListener listener2 = new DownloadListener2();
DownloadListener listener3 = new DownloadListener3();
DownloadListener listener4 = new DownloadListener4();
DownloadTask task = new DownloadTask.build(url, file).build();
manager.attachListener(task, listener1);
manager.attachListener(task, listener2);
manager.detachListener(task, listener2);
// 当一个任务结束时，这个任务的所有监听器都被移除
manager.addAutoRemoveListenersWhenTaskEnd(task.getId());
// enqueue task to start.
manager.enqueueTaskWithUnifiedListener(task, listener3);
manager.attachListener(task, listener4);
下面我们来分析一下这个下载框架的源码：
OkDownload
首先看一下OkDownload这个类，这个类定义了所有的下载策略，我们可以自定义一些下载策略，可以通过OkDownload的Builder构造自定义的一个OkDownload实例，再通过OkDownload.setSingletonInstance进行设置：
OkDownload.Builder builder = new OkDownload.Builder(context)
.downloadStore(downloadStore) //断点信息存储的位置，默认是SQLite数据库
.callbackDispatcher(callbackDispatcher) //监听回调分发器，默认在主线程回调
.downloadDispatcher(downloadDispatcher) //下载管理机制，最大下载任务数、同步异步执行下载任务的处理
.connectionFactory(connectionFactory) //选择网络请求框架，默认是OkHttp
.outputStreamFactory(outputStreamFactory) //构建文件输出流DownloadOutputStream，是否支持随机位置写入
.downloadStrategy(downloadStrategy) //下载策略，文件分为几个线程下载
.processFileStrategy(processFileStrategy) //多文件写文件的方式，默认是根据每个线程写文件的不同位置，支持同时写入。
.monitor(monitor); //下载状态监听
OkDownload.setSingletonInstance(builder.build());
DownloadTask
DownloadTask下载任务类，可通过它的Builder来构造一个下载任务，我们看它是如何执行的：
public void execute(DownloadListener listener) {
this.listener = listener;
OkDownload.with().downloadDispatcher().execute(this);
}
public void enqueue(DownloadListener listener) {
this.listener = listener;
OkDownload.with().downloadDispatcher().enqueue(this);
}
可以看到都是通过downloadDispatcher来执行下载任务的，默认的downloadDispatcher是一个DownloadDispatcher实例，我们以同步执行一个下载任务为例，看它是如何下载的：
public void execute(DownloadTask task) {
Util.d(TAG, "execute: " + task);
final DownloadCall call;
synchronized (this) {
if (inspectCompleted(task)) return;
if (inspectForConflict(task)) return;
call = DownloadCall.create(task, false, store);
runningSyncCalls.add(call);
}
syncRunCall(call);
}
void syncRunCall(DownloadCall call) {
call.run();
}
在execute方法里将一个DownloadTask实例又封装为了一个DownloadCall对象，然后在syncRunCall方法里执行了DownloadCall对象的run方法。通过看DownloadCall源码可以知道该类继承自NamedRunnable，而NamedRunnable实现了Runnable，在run方法里调用了execute方法。（enqueue执行任务最终则是调用 getExecutorService().execute(call);来异步执行的）
那我们看一下DownloadCall这个类。
DownloadCall
先看一下DownloadCall是如何实现execute方法的，该方法比较长，首先执行的是inspectTaskStart：
先看一下这个store是什么：
通过看OkDownload这个类的源码可以知道，DownloadCall的store是调用BreakpointStoreOnSQLite的createRemitSelf方法生成的一个实例：
可以看到是RemitStoreOnSQLite的一个实例，其主要用来保存任务及断点信息至本地数据库。RemitStoreOnSQLite里持有BreakpointStoreOnSQLite对象，BreakpointStoreOnSQLite里面包含了BreakpointSQLiteHelper（用于操作数据）和BreakpointStoreOnCache（用于做数据操作之前的数据缓存）。
@Override public void syncCacheToDB(int id) throws IOException {
sqLiteHelper.removeInfo(id);
final BreakpointInfo info = sqliteCache.get(id);
if (info == null || info.getFilename() == null || info.getTotalOffset() <= 0) return;
sqLiteHelper.insert(info);
}
最终会调用上述syncCacheToDB方法，先删除数据库中的任务信息，若缓存（创建BreakpointStoreOnCache对象时，会调用loadToCache方法将数据库中所有任务信息进行缓存）
this.onCache = new BreakpointStoreOnCache(helper.loadToCache(),
helper.loadResponseFilenameToMap());
中有该任务，则检查任务信息是否合法，若合法则再次将该任务及断点信息保存在本地数据库中。
inspectTaskStart方法结束后，会进入一个do-while循环，首先做一些下载前的准备工作：
do {
//1.判断当前任务的下载链接长度是否大于0，否则就抛出异常；
if (task.getUrl().length() <= 0) {
this.cache = new DownloadCache.PreError(
new IOException("unexpected url: " + task.getUrl()));
break;
}
if (canceled) break;
//2.从缓存中获取任务的断点信息，若没有断点信息，则创建断点信息并保存至数据库；
@NonNull final BreakpointInfo info;
try {
BreakpointInfo infoOnStore = store.get(task.getId());
if (infoOnStore == null) {
info = store.createAndInsert(task);
} else {
info = infoOnStore;
}
setInfoToTask(info);
} catch (IOException e) {
this.cache = new DownloadCache.PreError(e);
break;
}
if (canceled) break;
// 3.创建带缓存的下载输出流；
@NonNull final DownloadCache cache = createCache(info);
this.cache = cache;
// 4.访问下载链接判断断点信息是否合理；
final BreakpointRemoteCheck remoteCheck = createRemoteCheck(info);
try {
remoteCheck.check();
} catch (IOException e) {
cache.catchException(e);
break;
}
//5.确定文件路径后等待文件锁释放；
fileStrategy.getFileLock().waitForRelease(task.getFile().getAbsolutePath());
// 6. 判断缓存中是否有相同的任务，若有则复用缓存中的任务的分块信息；
OkDownload.with().downloadStrategy()
.inspectAnotherSameInfo(task, info, remoteCheck.getInstanceLength());
try {
//7.检查断点信息是否是可恢复的，若不可恢复，则根据文件大小进行分块，重新下载，否则继续进行下一步；
if (remoteCheck.isResumable()) {
// 8.判断断点信息是否是脏数据（文件存在且断点信息正确且下载链接支持断点续传）；
final BreakpointLocalCheck localCheck = createLocalCheck(info,
remoteCheck.getInstanceLength());
localCheck.check();
// 9.若是脏数据则根据文件大小进行分块，重新开始下载，否则从断点位置开始下载；
if (localCheck.isDirty()) {
Util.d(TAG, "breakpoint invalid: download from beginning because of "
+ "local check is dirty " + task.getId() + " " + localCheck);
fileStrategy.discardProcess(task);
assembleBlockAndCallbackFromBeginning(info, remoteCheck,
localCheck.getCauseOrThrow());
} else {
okDownload.callbackDispatcher().dispatch()
.downloadFromBreakpoint(task, info);
}
} else {
Util.d(TAG, "breakpoint invalid: download from beginning because of "
+ "remote check not resumable " + task.getId() + " " + remoteCheck);
fileStrategy.discardProcess(task);
assembleBlockAndCallbackFromBeginning(info, remoteCheck,
remoteCheck.getCauseOrThrow());
}
} catch (IOException e) {
cache.setUnknownError(e);
break;
}
// 10. 开始下载
start(cache, info);
if (canceled) break;
// 11. 错误重试机制
if (cache.isPreconditionFailed()
&& retryCount++ < MAX_COUNT_RETRY_FOR_PRECONDITION_FAILED) {
store.remove(task.getId());
retry = true;
} else {
retry = false;
}
} while (retry);
1.判断当前任务的下载链接长度是否大于0，否则就抛出异常；2.从缓存中获取任务的断点信息，若没有断点信息，则创建断点信息并保存至数据库；3.创建带缓存的下载输出流；4.访问下载链接判断断点信息是否合理；5.确定文件路径后等待文件锁释放； 6. 判断缓存中是否有相同的任务，若有则复用缓存中的任务的分块信息；7.检查断点信息是否是可恢复的，若不可恢复，则根据文件大小进行分块，重新下载，否则继续进行下一步；8.判断断点信息是否是脏数据（文件存在且断点信息正确且下载链接支持断点续传）；9.若是脏数据则根据文件大小进行分块，重新开始下载，否则从断点位置开始下载；10.开始下载。
文件分成多少块进行下载由DownloadStrategy决定的：
// 1 connection: [0, 1MB)
private static final long ONE_CONNECTION_UPPER_LIMIT = 1024 * 1024; // 1MiB
// 2 connection: [1MB, 5MB)
private static final long TWO_CONNECTION_UPPER_LIMIT = 5 * 1024 * 1024; // 5MiB
// 3 connection: [5MB, 50MB)
private static final long THREE_CONNECTION_UPPER_LIMIT = 50 * 1024 * 1024; // 50MiB
// 4 connection: [50MB, 100MB)
private static final long FOUR_CONNECTION_UPPER_LIMIT = 100 * 1024 * 1024; // 100MiB
public ResumeAvailableResponseCheck resumeAvailableResponseCheck(
DownloadConnection.Connected connected,
int blockIndex,
BreakpointInfo info) {
return new ResumeAvailableResponseCheck(connected, blockIndex, info);
}
public int determineBlockCount(@NonNull DownloadTask task, long totalLength) {
if (task.getSetConnectionCount() != null) return task.getSetConnectionCount();
if (totalLength < ONE_CONNECTION_UPPER_LIMIT) {
return 1;
}
if (totalLength < TWO_CONNECTION_UPPER_LIMIT) {
return 2;
}
if (totalLength < THREE_CONNECTION_UPPER_LIMIT) {
return 3;
}
if (totalLength < FOUR_CONNECTION_UPPER_LIMIT) {
return 4;
}
return 5;
}
文件大小在0-1MB、1-5MB、5-50MB、50-100MB、100MB以上时分别开启1、2、3、4、5个线程进行下载。
我们重点看一下下载部分的源码，也就是start(cache，info)方法：
void start(final DownloadCache cache, BreakpointInfo info) throws InterruptedException {
final int blockCount = info.getBlockCount();
final List<DownloadChain> blockChainList = new ArrayList<>(info.getBlockCount());
for (int i = 0; i < blockCount; i++) {
final BlockInfo blockInfo = info.getBlock(i);
if (Util.isCorrectFull(blockInfo.getCurrentOffset(), blockInfo.getContentLength())) {
continue;
}
Util.resetBlockIfDirty(blockInfo);
blockChainList.add(DownloadChain.createChain(i, task, info, cache, store));
}
if (canceled) {
return;
}
startBlocks(blockChainList);
}
可以看到它是分块下载的，每一个分块都是一个DownloadChain实例，DownloadChain实现了Runnable接口，继续看startBlocks方法:
对于每一个分块任务，都调用了submitChain方法，由一个线程池去处理每一个DownloadChain分块，核心代码就在这里:
void start() throws IOException {
final CallbackDispatcher dispatcher = OkDownload.with().callbackDispatcher();
// 处理请求拦截链
final RetryInterceptor retryInterceptor = new RetryInterceptor();
final BreakpointInterceptor breakpointInterceptor = new BreakpointInterceptor();
connectInterceptorList.add(retryInterceptor);
connectInterceptorList.add(breakpointInterceptor);
connectInterceptorList.add(new RedirectInterceptor());
connectInterceptorList.add(new HeaderInterceptor());
connectInterceptorList.add(new CallServerInterceptor());
connectIndex = 0;
final DownloadConnection.Connected connected = processConnect();
if (cache.isInterrupt()) {
throw InterruptException.SIGNAL;
}
dispatcher.dispatch().fetchStart(task, blockIndex, getResponseContentLength());
// 获取数据拦截链
final FetchDataInterceptor fetchDataInterceptor =
new FetchDataInterceptor(blockIndex, connected.getInputStream(),
getOutputStream(), task);
fetchInterceptorList.add(retryInterceptor);
fetchInterceptorList.add(breakpointInterceptor);
fetchInterceptorList.add(fetchDataInterceptor);
fetchIndex = 0;
final long totalFetchedBytes = processFetch();
dispatcher.dispatch().fetchEnd(task, blockIndex, totalFetchedBytes);
}
可以看到它主要使用责任链模式进行了两个链式调用：处理请求拦截链和获取数据拦截链。
处理请求拦截链包含了RetryInterceptor重试拦截器、BreakpointInterceptor断点拦截器、RedirectInterceptor重定向拦截器、HeaderInterceptor头部信息处理拦截器、CallServerInterceptor请求拦截器，该链式调用过程会逐个调用拦截器的interceptConnect方法：
public class RetryInterceptor implements Interceptor.Connect, Interceptor.Fetch {
@NonNull @Override
public DownloadConnection.Connected interceptConnect(DownloadChain chain) throws IOException {
final DownloadCache cache = chain.getCache();
// 如果产生了RetryException，则重新执行该链式调用
while (true) {
try {
if (cache.isInterrupt()) {
throw InterruptException.SIGNAL;
}
return chain.processConnect();
} catch (IOException e) {
if (e instanceof RetryException) {
chain.resetConnectForRetry();
continue;
}
chain.getCache().catchException(e);
throw e;
}
}
}
......
}
public class BreakpointInterceptor implements Interceptor.Connect, Interceptor.Fetch {
private static final String TAG = "BreakpointInterceptor";
@NonNull @Override
public DownloadConnection.Connected interceptConnect(DownloadChain chain) throws IOException {
final DownloadConnection.Connected connected = chain.processConnect();
final BreakpointInfo info = chain.getInfo();
if (chain.getCache().isInterrupt()) {
throw InterruptException.SIGNAL;
}
if (info.getBlockCount() == 1 && !info.isChunked()) {
// 当只有一个线程进行下载文件时，如果断点信息中保存的文件长度和服务端返回的文件长度不一致，则以服务端返回的为准重新进行下载
final long blockInstanceLength = getExactContentLengthRangeFrom0(connected);
final long infoInstanceLength = info.getTotalLength();
if (blockInstanceLength > 0 && blockInstanceLength != infoInstanceLength) {
Util.d(TAG, "SingleBlock special check: the response instance-length["
+ blockInstanceLength + "] isn't equal to the instance length from trial-"
+ "connection[" + infoInstanceLength + "]");
final BlockInfo blockInfo = info.getBlock(0);
boolean isFromBreakpoint = blockInfo.getRangeLeft() != 0;
final BlockInfo newBlockInfo = new BlockInfo(0, blockInstanceLength);
info.resetBlockInfos();
info.addBlock(newBlockInfo);
if (isFromBreakpoint) {
final String msg = "Discard breakpoint because of on this special case, we have"
+ " to download from beginning";
Util.w(TAG, msg);
throw new RetryException(msg);
}
OkDownload.with().callbackDispatcher().dispatch()
.downloadFromBeginning(chain.getTask(), info, CONTENT_LENGTH_CHANGED);
}
}
// update for connected.
final DownloadStore store = chain.getDownloadStore();
try {
if (!store.update(info)) {
throw new IOException("Update store failed!");
}
} catch (Exception e) {
throw new IOException("Update store failed!", e);
}
return connected;
}
......
}
public class RedirectInterceptor implements Interceptor.Connect {
//最大重定向次数
static final int MAX_REDIRECT_TIMES = 10;
private static final int HTTP_TEMPORARY_REDIRECT = 307;
private static final int HTTP_PERMANENT_REDIRECT = 308;
@NonNull @Override
public DownloadConnection.Connected interceptConnect(DownloadChain chain) throws IOException {
int redirectCount = 0;
String url;
DownloadConnection connection;
while (true) {
if (chain.getCache().isInterrupt()) {
throw InterruptException.SIGNAL;
}
final DownloadConnection.Connected connected = chain.processConnect();
final int code = connected.getResponseCode();
if (!isRedirect(code)) {
return connected;
}
//若需要重定向，则根据返回的新的url重新进行网络请求
if (++redirectCount >= MAX_REDIRECT_TIMES) {
throw new ProtocolException("Too many redirect requests: " + redirectCount);
}
url = connected.getResponseHeaderField("Location");
if (url == null) {
throw new ProtocolException(
"Response code is " + code + " but can't find Location field");
}
chain.releaseConnection();
connection = OkDownload.with().connectionFactory().create(url);
chain.setConnection(connection);
chain.setRedirectLocation(url);
}
}
private static boolean isRedirect(int code) {
return code == HttpURLConnection.HTTP_MOVED_PERM
|| code == HttpURLConnection.HTTP_MOVED_TEMP
|| code == HttpURLConnection.HTTP_SEE_OTHER
|| code == HttpURLConnection.HTTP_MULT_CHOICE
|| code == HTTP_TEMPORARY_REDIRECT
|| code == HTTP_PERMANENT_REDIRECT;
}
}
public class HeaderInterceptor implements Interceptor.Connect {
private static final String TAG = "HeaderInterceptor";
@NonNull @Override
public DownloadConnection.Connected interceptConnect(DownloadChain chain) throws IOException {
final BreakpointInfo info = chain.getInfo();
final DownloadConnection connection = chain.getConnectionOrCreate();
final DownloadTask task = chain.getTask();
// 添加User-Agent字段
final Map<String, List<String>> userHeader = task.getHeaderMapFields();
if (userHeader != null) Util.addUserRequestHeaderField(userHeader, connection);
if (userHeader == null || !userHeader.containsKey(USER_AGENT)) {
Util.addDefaultUserAgent(connection);
}
//添加Range字段
final int blockIndex = chain.getBlockIndex();
final BlockInfo blockInfo = info.getBlock(blockIndex);
if (blockInfo == null) {
throw new IOException("No block-info found on " + blockIndex);
}
String range = "bytes=" + blockInfo.getRangeLeft() + "-";
range += blockInfo.getRangeRight();
connection.addHeader(RANGE, range);
Util.d(TAG, "AssembleHeaderRange (" + task.getId() + ") block(" + blockIndex + ") "
+ "downloadFrom(" + blockInfo.getRangeLeft() + ") currentOffset("
+ blockInfo.getCurrentOffset() + ")");
// 如果有Etag信息，则添加If-Match字段
final String etag = info.getEtag();
if (!Util.isEmpty(etag)) {
connection.addHeader(IF_MATCH, etag);
}
if (chain.getCache().isInterrupt()) {
throw InterruptException.SIGNAL;
}
OkDownload.with().callbackDispatcher().dispatch()
.connectStart(task, blockIndex, connection.getRequestProperties());
DownloadConnection.Connected connected = chain.processConnect();
Map<String, List<String>> responseHeaderFields = connected.getResponseHeaderFields();
if (responseHeaderFields == null) responseHeaderFields = new HashMap<>();
OkDownload.with().callbackDispatcher().dispatch().connectEnd(task, blockIndex,
connected.getResponseCode(), responseHeaderFields);
if (chain.getCache().isInterrupt()) {
throw InterruptException.SIGNAL;
}
// 检查Etag字段是否一致
final DownloadStrategy strategy = OkDownload.with().downloadStrategy();
final DownloadStrategy.ResumeAvailableResponseCheck responseCheck =
strategy.resumeAvailableResponseCheck(connected, blockIndex, info);
responseCheck.inspect();
//获取Content-Length、Content-Range字段信息
final long contentLength;
final String contentLengthField = connected.getResponseHeaderField(CONTENT_LENGTH);
if (contentLengthField == null || contentLengthField.length() == 0) {
final String contentRangeField = connected.getResponseHeaderField(CONTENT_RANGE);
contentLength = Util.parseContentLengthFromContentRange(contentRangeField);
} else {
contentLength = Util.parseContentLength(contentLengthField);
}
chain.setResponseContentLength(contentLength);
return connected;
}
}
public class CallServerInterceptor implements Interceptor.Connect {
@NonNull @Override
public DownloadConnection.Connected interceptConnect(DownloadChain chain) throws IOException {
OkDownload.with().downloadStrategy().inspectNetworkOnWifi(chain.getTask());
OkDownload.with().downloadStrategy().inspectNetworkAvailable();
\\进行网络请求，获得响应
return chain.getConnectionOrCreate().execute();
}
}
获取数据拦截链包含了RetryInterceptor重试拦截器、BreakpointInterceptor断点拦截器、RedirectInterceptor重定向拦截器、HeaderInterceptor头部信息处理拦截器、FetchDataInterceptor获取数据拦截器，该链式调用过程会逐个调用拦截器的interceptFetch方法：
public class RetryInterceptor implements Interceptor.Connect, Interceptor.Fetch {
......
@Override
public long interceptFetch(DownloadChain chain) throws IOException {
try {
return chain.processFetch();
} catch (IOException e) {
chain.getCache().catchException(e);
throw e;
}
}
}
public class BreakpointInterceptor implements Interceptor.Connect, Interceptor.Fetch {
......
@Override
public long interceptFetch(DownloadChain chain) throws IOException {
final long contentLength = chain.getResponseContentLength();
final int blockIndex = chain.getBlockIndex();
final boolean isNotChunked = contentLength != CHUNKED_CONTENT_LENGTH;
long fetchLength = 0;
long processFetchLength;
final MultiPointOutputStream outputStream = chain.getOutputStream();
try {
while (true) {
//循环调用FetchDataInterceptor拦截器读写文件
processFetchLength = chain.loopFetch();
if (processFetchLength == -1) {
break;
}
fetchLength += processFetchLength;
}
} finally {
chain.flushNoCallbackIncreaseBytes();
if (!chain.getCache().isUserCanceled()) outputStream.done(blockIndex);
}
if (isNotChunked) {
outputStream.inspectComplete(blockIndex);
if (fetchLength != contentLength) {
throw new IOException("Fetch-length isn't equal to the response content-length, "
+ fetchLength + "!= " + contentLength);
}
}
return fetchLength;
}
......
public class FetchDataInterceptor implements Interceptor.Fetch {
private final InputStream inputStream;
private final byte[] readBuffer;
private final MultiPointOutputStream outputStream;
private final int blockIndex;
private final DownloadTask task;
private final CallbackDispatcher dispatcher;
public FetchDataInterceptor(int blockIndex,
@NonNull InputStream inputStream,
@NonNull MultiPointOutputStream outputStream,
DownloadTask task) {
this.blockIndex = blockIndex;
this.inputStream = inputStream;
this.readBuffer = new byte[task.getReadBufferSize()];
this.outputStream = outputStream;
this.task = task;
this.dispatcher = OkDownload.with().callbackDispatcher();
}
@Override
public long interceptFetch(DownloadChain chain) throws IOException {
if (chain.getCache().isInterrupt()) {
throw InterruptException.SIGNAL;
}
OkDownload.with().downloadStrategy().inspectNetworkOnWifi(chain.getTask());
// 读取数据
int fetchLength = inputStream.read(readBuffer);
if (fetchLength == -1) {
return fetchLength;
}
//写文件
outputStream.write(blockIndex, readBuffer, fetchLength);
// 判断是否回调下载进度
chain.increaseCallbackBytes(fetchLength);
if (this.dispatcher.isFetchProcessMoment(task)) {
chain.flushNoCallbackIncreaseBytes();
}
return fetchLength;
}
}
每一个DownloadChain都完成后，最终会调用inspectTaskEnd方法，从数据库中删除该任务，并回调通知任务完成。这样，一个完整的下载任务就完成了。总体流程如下：
OkDownload的优势在于：
1.OkDownload内部使用的网络请求框架默认为OkHttp，OkHttp底层使用的IO库为Okio，相较于原生Java IO流，它更加简便高效。
2.使用了数据库缓存+内存缓存的二级缓存模式，操作效率更高。
3.功能更完善，除了多线程断点续传外，还提供了暂停功能，多种回调监听功能，多任务管理功能等。
4.更加可靠：下载前有多重检查机制来判断重新下载还是从断点处下载；每次从断点续传时，都会对比响应信息跟之前是否一致；对重定向做了处理；有错误重试机制。
5.可配置性高，可以注入自定义组件。
