Android知名三方库OKHttp(五) - 单线程&多线程文

源代码
GitHub源代码

本文目标

okhttp的单线程和多线程文件下载(仅供学习)

单线程下载文件

1.基本用法

    /**
     * 单线程下载
     */
    public void method1(View view) {
        String url = "https://dl008.liqucn.com/upload/2021/286/e/com.ss.android.ugc.aweme_16.5.0_liqucn.com.apk";

        //1.1创建okHttpClient
        OkHttpClient okHttpClient = new OkHttpClient();

        //1.2创建Request对象
        Request request = new Request.Builder().url(url).build();

        //2.把Request对象封装成call对象
        Call call = okHttpClient.newCall(request);

        //3.发起异步请求
        call.enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                e.printStackTrace();
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                InputStream inputStream = response.body().byteStream();
                final File file = new File(Environment.getExternalStorageDirectory().getAbsolutePath(), "douyin.apk");
                FileOutputStream outputStream = new FileOutputStream(file);
                int len = 0;
                byte[] bytes = new byte[1024 * 10];
                while ((len = inputStream.read(bytes)) != -1) {
                    outputStream.write(bytes, 0, len);
                }
                inputStream.close();
                outputStream.close();
                runOnUiThread(new Runnable() {
                    @Override
                    public void run() {
                        AppInfoUtils.install(file.getAbsolutePath());
                    }
                });
            }
        });
    }

以上就是最基础的okhttp的单线程下载文件,炒鸡简单

多线程下载文件

多线程文件下载，我们会使用到参考okhttp的Dispatcher类来使用线程池来下载文件
首先来看一下我们的基本使用，在这里我们会进行封装

1.基本用法

    /**
     * 多线程下载
     */
    public void method1(View view) {
        String url = "https://dl008.liqucn.com/upload/2021/286/e/com.ss.android.ugc.aweme_16.5.0_liqucn.com.apk";
        DownLoadFacade.getInstance().startDownLoad(url, new DownloadCallback() {
            @Override
            public void onFailure(IOException e) {
                e.printStackTrace();
            }
            @Override
            public void onSucceed(final File file) {
                runOnUiThread(new Runnable() {
                    @Override
                    public void run() {
                        AppInfoUtils.install(file.getAbsolutePath());
                    }
                });
            }
        });
    }

我们先看下对外是怎么使用的,从DownLoadFacade这个类开始

2.对外API调用类DownLoadFacade

public class DownLoadFacade {

    private DownLoadFacade() {
    }

    public static DownLoadFacade getInstance() {
        return SingleHolder.INSTANCE;
    }

    private static class SingleHolder {
        private static final DownLoadFacade INSTANCE = new DownLoadFacade();
    }

    /**
     * 初始化
     * @param context
     */
    public void init(Context context){
        FileManager.getInstance().init(context.getApplicationContext());
    }

    /**
     * 开始下载
     * @param url
     * @param callback
     */
    public void startDownLoad(String url,DownloadCallback callback){
        DownLoadDispatcher.getInstance().startDownLoad(url,callback);
    }
}

可以看出来该类就是一个单例,然后有几个方法,其中一个是开始下载
来让我们看一下startDownLoad这个方法

3.DownLoadDispatcher

public class DownLoadDispatcher {

    //下载正在运行队列
    private final Deque<DownLoadTask> runningTasks = new ArrayDeque<>();

    //专门开了个线程池来更新进度条
    private static ExecutorService sLocalProgressPool = Executors.newFixedThreadPool(THREAD_SIZE);

    private DownLoadDispatcher() {
    }

    public static DownLoadDispatcher getInstance() {
        return SingleHolder.INSTANCE;
    }

    private static class SingleHolder {
        private static final DownLoadDispatcher INSTANCE = new DownLoadDispatcher();
    }

    /**
     * 开始下载
     */
    public void startDownLoad(final String url, final DownloadCallback callback) {
        Call call = OkHttpManager.getInstance().asyncCall(url);
        call.enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                callback.onFailure(e);
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                //获取文件的总大小
                final long contentLength = response.body().contentLength();
                if (contentLength <= -1) {
                    // 没有获取到文件的大小，
                    // 1. 跟后台商量
                    // 2. 只能采用单线程去下载
                    return;
                }
                DownLoadTask downLoadTask = new DownLoadTask(url, contentLength, callback);
                //开启线程池去下载
                downLoadTask.init();
                //添加到运行队列
                runningTasks.add(downLoadTask);
                //更新进度
                updateProgress(url, callback, contentLength);
            }
        });
    }

    /**
     * 进度更新
     */
    private void updateProgress(final String url, final DownloadCallback callback, final long contentLength) {
        sLocalProgressPool.execute(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(300);
                        File file = FileManager.getInstance().getFile(url);
                        long fileSize = file.length();
                        int progress = (int) (fileSize * 100.0 / contentLength);
                        if (progress >= 100) {
                            callback.progress(progress);
                            return;
                        }
                        callback.progress(progress);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }

    /**
     * 移除正在运行的task
     */
    public void recyclerTask(DownLoadTask downloadTask) {
       // 参考 OkHttp 的 Dispatcher 的源码,如果还有需要下载的开始下一个的下载
        runningTasks.remove(downloadTask);
    }
}

可以看出在startDownLoad方法中首先先拿到Okhttp的call对象,然后发起异步请求,在这个请求中我们先拿到文件的总长度,然后创建DownLoadTask类,在这个类中我们会开启线程池去进行多线程下载,至于updateProgress进度更新我们最后说

4.DownLoadTask

public class DownLoadTask {

    //CPU核心数,参考来自AsyncTask源码
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    //线程池中的线程总数量
    public static final int THREAD_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));

    //线程池
    private ExecutorService executorService;

    private String mUrl;
    private long mContentLength;
    private DownloadCallback mCallback;

    private volatile int mSucceedNumber;

    public DownLoadTask(String url, long contentLength, DownloadCallback callback) {
        this.mUrl = url;
        this.mContentLength = contentLength;
        this.mCallback = callback;
    }

    /**
     * 创建线程池,线程总数量为THREAD_SIZE个,然后存活时间为30s,参考自okhttp的Dispatcher类
     */
    public synchronized ExecutorService executorService() {
        if (executorService == null) {
            executorService = new ThreadPoolExecutor(0, THREAD_SIZE, 60, TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>(), new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r, "YD_DownLoadTask");
                    thread.setDaemon(false);
                    return thread;
                }
            });
        }
        return executorService;
    }

    /**
     * 初始化,并开始执行任务
     */
    public void init() {
        for (int x = 0; x < THREAD_SIZE; x++) {
            //总长度 / 线程数 = 每个线程数要下载的字节内容长度
            long threadSize = mContentLength / THREAD_SIZE;

            /**
             * 假如当前线程数为3,然后总mContentLength是300字节,那每个线程需要下载100字节
             * 那第0个区间就是[0-99],第1个区间[100-199],第2个区间[200-299]
             */
            //开始下载的字节点
            long start = x * threadSize;
            //结束下载的字节点
            long end = (threadSize + x * threadSize) - 1;

            if (x == THREAD_SIZE - 1) {
                end = mContentLength - 1;
            }

            //字节总数=22347996   threadSize = 7449332   start=0  end =7449331
            //字节总数=22347996   threadSize = 7449332   start=7449332  end =14898663
            //字节总数=22347996   threadSize = 7449332   start=14898664  end =22347995
            System.out.println("字节总数="+mContentLength+"   threadSize = " + threadSize + "   start=" + start + "  end =" + end);

            //创建runnable对象
            DownloadRunnable runnable = new DownloadRunnable(mUrl, x, start, end, new DownloadCallback() {
                @Override
                public void onFailure(IOException e) {
                    // 一个apk 下载里面有一个线程异常了，处理异常,把其他线程停止掉
                    mCallback.onFailure(e);
                }

                @Override
                public void onSucceed(File file) {
                    synchronized (DownLoadTask.class) {
                        mSucceedNumber += 1;
                        if (mSucceedNumber == THREAD_SIZE) {
                            mCallback.onSucceed(file);
                            DownLoadDispatcher.getInstance().recyclerTask(DownLoadTask.this);
                        }
                    }
                }

                @Override
                public void progress(int progress) {

                }
            });

            //开启执行这个线程
            executorService().execute(runnable);
        }
    }
}

在这里我们参考了okhttp的Dispatcher类的线程池设计,我们也是调用executorService()方法创建线程池,该线程池中的线程如果不活动只会存活60秒,然后线程数最大数量为THREAD_SIZE个(这个是参考AsyncTask的写法)
当外界调用init()方法的时候我们就会根据开了几个线程,然后一个for循环

• 1.文件总长度 / 线程数 = 每个线程数要下载的字节内容长度 long threadSize = mContentLength / THREAD_SIZE;
• 2. 计算出来开始下载的字节数 long start = x * threadSize;
• 3.计算出来结束下载的字节数 long end = (threadSize + x * threadSize) - 1;
• 4. 创建DownloadRunnable然后开始去下载(该内部是同步请求去下载指定区间的字节)
• 5.开始执行线程池中的runnable
  在这里,假如当前线程数为3,然后总mContentLength是300字节,那每个线程需要下载100字节,那第0个区间就是[0-99],第1个区间[100-199],第2个区间[200-299],我们来看一下DownloadRunnable对象

5.DownloadRunnable

public class DownloadRunnable implements Runnable {

    private static final int STATUS_DOWNLOADING = 1;
    private static final int STATUS_STOP = 2;
    private int mStatus = STATUS_DOWNLOADING;

    private String mUrl;
    private final int mThreadId;
    private final long mStart;
    private final long mEnd;
    private DownloadCallback mCallback;


    public DownloadRunnable(String url, int threadId, long start, long end, DownloadCallback callback) {
        this.mUrl = url;
        this.mThreadId = threadId;
        this.mStart = start;// 1M-2M 0.5M  1.5M - 2M
        this.mEnd = end;
        this.mCallback = callback;
    }

    @Override
    public void run() {
        RandomAccessFile accessFile = null;
        InputStream inputStream = null;
        try {
            Response response = OkHttpManager.getInstance().syncResponse(mUrl, mStart, mEnd);
            Log.e("TAG", this.toString());

            inputStream = response.body().byteStream();

            // 写数据
            File file = FileManager.getInstance().getFile(mUrl);
            accessFile = new RandomAccessFile(file, "rwd");
            // 从这里开始
            accessFile.seek(mStart);

            int len = 0;
            byte[] buffer = new byte[1024 * 10];
            while ((len = inputStream.read(buffer)) != -1) {
                if (mStatus == STATUS_STOP) {
                    break;
                }
                accessFile.write(buffer, 0, len);
            }
            //数据写完,回调出去
            mCallback.onSucceed(file);
        } catch (IOException e) {
            mCallback.onFailure(e);
        } finally {
            Utils.close(inputStream);
            Utils.close(accessFile);
        }
    }

    public void stop() {
        mStatus = STATUS_STOP;
    }

    @Override
    public String toString() {
        return "DownloadRunnable{" +
                "mUrl='" + mUrl + '\'' +
                ", mThreadId=" + mThreadId +
                ", mStart=" + mStart +
                ", mEnd=" + mEnd +
                ", mCallback=" + mCallback +
                '}';
    }
}

发现给内部的run()方法首先是调用了下面这行代码得到Response

Response response = OkHttpManager.getInstance().syncResponse(mUrl, mStart, mEnd);

具体可以看下里面都怎么写的

public class OkHttpManager {
   public Response syncResponse(String url, long start, long end) throws IOException {
        Request request = new Request
                .Builder()
                .url(url)
                .addHeader("Range", "bytes=" + start + "-" + end)
                .build();
        return okHttpClient.newCall(request).execute();
    }
}

是在header中添加Range然后指定从哪开始到哪结束,这里是个同步请求,然后我们拿到Response对象后就是开启IO流,把数据写到文件中,直到所有的线程都跑完并把数据写完就算完全下载好了
我们会在DownLoadTask类中的DownloadRunnable接口回调成功中用字段mSucceedNumber字段表示成功了几个下载数,当mSucceedNumber == 线程总数的时候说明全部下载好了

6.进度更新

//专门开了个线程池来更新进度条
private static ExecutorService sLocalProgressPool = Executors.newFixedThreadPool(THREAD_SIZE);

public class DownLoadDispatcher {

    /**
     * 开始下载
     */
    public void startDownLoad(final String url, final DownloadCallback callback) {
        Call call = OkHttpManager.getInstance().asyncCall(url);
        call.enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
               ``````
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
               ``````
                //更新进度
                updateProgress(url, callback, contentLength);
            }
        });
    }

    /**
     * 进度更新
     */
    private void updateProgress(final String url, final DownloadCallback callback, final long contentLength) {
        sLocalProgressPool.execute(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(300);
                        File file = FileManager.getInstance().getFile(url);
                        long fileSize = file.length();
                        int progress = (int) (fileSize * 100.0 / contentLength);
                        if (progress >= 100) {
                            callback.progress(progress);
                            return;
                        }
                        callback.progress(progress);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }
}

在这里我们采用一个简单的办法,通过开启一个新的线程池来更新进度条,没个300毫米就扫描下载的文件大小,这样不停的通过回调接口返回出去
具体的可以参考demo