Java多線程-五種線程池分析以及AnsyncTask源碼分析

時間  2019-11-29
標籤 java 多線程 五種 線程 分析 以及 ansynctask 源碼 欄目 Java 简体版
原文   原文鏈接

這是小編我的的學習總結，還望各位大神多多吐槽~android

一.前言

線程是咱們處理耗時操做的利器可是每一個線程的建立和銷燬都須要必定的開銷。只是偶爾的使用線程去作一些操做的時候仍是能夠直接new的，若是須要大量的線程去作一些操做的時候咱們就要慎重考慮了，好比某個軟件下載APP這樣的。這裏呢Java 1.5中提供了Executor框架用於把任務的提交和執行解耦，任務的提交交給了Runable或者Callable,而Executor框架用來處理任務。ThreadPollExecutor則是Executor框架的核心實現類。下面咱們就來看下這個類。緩存

（1）ThreadPollExecutor

咱們能夠直接用該類去建立一個線程池，咱們先來看一下ThreadPollExecutor類最多的一個參數列表，讀完你就知道怎麼使用了。bash

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable>                    workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {複製代碼

  • corePoolSize: 核心線程數。默認狀況下線程池是空的，只有有任務提交的時候纔會去建立線程。當運行的線程少於corePoolSize的時候，新建任務將建立核心線程，若是的等於或者大於corePoolSize的時候將不會建立核心線程，建立普通線程。能夠調用prestartAllcoreThread的方法來提早建立並啓動全部的核心線程。網絡

  • maximumPoolSize：線程池所容許的最大的線程數量。當任務隊列滿了的話線程數小於maximumPoolSize的值的話，新建任務仍是會建立新的線程的。框架

  • keepAliveTime ： 非核心線程閒置的超時時間，第一是非核心的線程第二是閒置狀態。調用allowCoreThreadTimeOut的時候該設置也會做用再核心的線程上面。異步

  • TimeUnit ： 超時的時間單位。DAYS(天)，HOURS（小時），MINUTES（f分鐘）,SECONDS（秒）,MILLISECONDS（毫秒）.async

  • workQueue ： 任務隊列（阻塞隊列）。當前線程書大於corePoolSize的大小的時候，會將任務加入阻塞隊列裏面。該隊列是BlockingQueue類型的。ide

  • ThreadFactory ： 線程工場。咱們能夠本身去定義。源碼分析

  • RejectedExecutionHandler ： 飽和策略。這是當任務隊列和線程數都滿了的時候所採起的的對應策略默認是AbordPolicy表示沒法處理新的任務，並拋出RejectedExecutionException異常。post

    1.CallerRunsPolicy ： 用調用者所在的線程來處理任務。

    2.DisCardPolicy ： 不能執行任務並將任務刪除。

    2.DisCardOldesPolicy ： 丟棄隊列最近的任務，並執行當前的任務， 會一直執行下去。

（2）FixedThreadPool

FixedThreadPool是可重的固定線程數的線程池。

//建立FixedThreadPool
 Executors.newFixedThreadPool(5);

//建立所須要的參數
  public static ExecutorService newFixedThreadPool(int nThreads) {
 //最後調用的都是ThreadPoolExecutor
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }複製代碼

咱們只須要傳如須要建立的線程的數量，也就是線程池的大小。咱們能夠看到構造方法，傳入的線程的數量就是核心線程的數量。也就是FixedThreadPool會建立固定數量核心線程的線程池，而且這些核心線程不會被回收，任務超過線程的數量將存入隊列中。

（2）CachedThreadPool

該線程池是根據須要去建立線程的。

//建立緩存線程池
Executors.newCachedThreadPool();
//構造方法
  public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }複製代碼

咱們能夠看出，這裏面的核心線程數是0，而線程的最大值是Integer.Max_VALUE。閒置超時間是一分鐘。默認隊列能夠保證任務順序的執行。CacheThreadPool適合大量的須要當即處理而且耗時較少的任務。

（4）SingleThreadExecutor

該類是使用單個線程的線程池。

//建立方法
  Executors.newSingleThreadExecutor();
  //構造方法
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }複製代碼

corePoolSize和maximumPoolSize都是1，意味着SingleThreadExecutor只有一個核心線程。其餘的參數和FixedThreadPool同樣。若是已經建立了一個線程再來一個任務的時候會將該任務加入到任務隊列裏面，確保了全部任務的執行順序。

（5）ScheduledThreadPool

ScheduledThreadPool是一個能實現定是和週期性任務的線程池。

//建立
 Executors.newScheduledThreadPool(5);
//構造方法
     public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE,
              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
              new DelayedWorkQueue());
    }
//最後仍是調用的
ThreadPoolExecutor的構造方法。複製代碼

當執行任務的時候，會先將任務包裝成ScheduledFutrueTask並添加到DelayedWorkQueue裏面，當沒超過corePoolSize的時候，會建立線程，人後去DelayedWorkQueue隊列裏面去拿任務，並非當即的執行。當執行完任務的時候會將ScheduledFutrueTask中的time變量改成下次要執行的時間並放回DelayedWorkQueue中。

二.AsyncTask的源碼分析

有了以前的鋪墊咱們就能夠去看下AsyncTask的源碼了。我相信再很早的時候你們都會接觸到AsyncTask加HttpUrlConnection的組合去下載網絡數據。咱們先來看下構造：

public abstract class AsyncTask<Params, Progress, Result> {
}複製代碼

AsyncTask是一個抽象的泛型類，他有三個泛型，第一個是要傳入的參數的類型Params，第二個是進度會掉的參數類型Progress，第三個是返回值類型額參數Result。AsyncTask有四個核心的方法。

  • onPreExecute(） ： 再任務啓動前調用，最早調用。

  • doInbackground(Parsms..) : 在線程池中執行。再onPreExecute執行後執行，該方法運行再子線程中。再該方法中去處理耗時的操做。

  • onProgressUpdate(Progress...） ： 再主線程中執行。當調用publishProgress（Progress...）的時候，此方法會將進度更新到UI上。

  • onPostExecute（Result result）: 再主線程中。接受doInbackground方法返回的值。

AsyncTask再3.0以前和以後的改動比較大。咱們直接來看3.0以後的版本。先來看下構造方法：

public AsyncTask() {
    //1
        mWorker = new WorkerRunnable<Params, Result>() {
            public Result call() throws Exception {
                mTaskInvoked.set(true);
                Result result = null;
                try {
                    Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
                    //noinspection unchecked
                    result = doInBackground(mParams);
                    Binder.flushPendingCommands();
                } catch (Throwable tr) {
                    mCancelled.set(true);
                    throw tr;
                } finally {
                    postResult(result);
                }
                return result;
            }
        };
//2
        mFuture = new FutureTask<Result>(mWorker) {
            @Override
            protected void done() {
                try {
                    postResultIfNotInvoked(get());
                } catch (InterruptedException e) {
                    android.util.Log.w(LOG_TAG, e);
                } catch (ExecutionException e) {
                    throw new RuntimeException("An error occurred while executing doInBackground()",
                            e.getCause());
                } catch (CancellationException e) {
                    postResultIfNotInvoked(null);
                }
            }
        };
    }複製代碼

從註釋1咱們能夠看到WorkerRunable實現了Callable接口，並實現了Call方法再Call方法裏面調用了doInBackground方法來處理任務並獲得結果，而後統統過postResult方法將結果傳遞出去。這就解釋了爲何doInBackground的方法是在子線程中，可是postResult也在call裏面啊，別急慢慢來。註釋2這裏是一個FutureTask的可管理的異步任務，它實現了Runable和Future接口。所以能夠包裝成Runable和Callable，提供給Executor執行。也能夠直接調用FutureTask.run()方法。這裏new FutureTask(mWorker) 將WorkerRunnable經過參數保存到本身的方法內存中，在執行的時候會用到。當調用了exeture（）方法時：

public final AsyncTask<Params, Progress, Result> execute(Params... params) {
        return executeOnExecutor(sDefaultExecutor, params);
    }

//全局變量定義
    private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;

//調用到
public final AsyncTask<Params, Progress, Result> executeOnExecutor(Executor exec,
            Params... params) {
        if (mStatus != Status.PENDING) {
            switch (mStatus) {
                case RUNNING:
                    throw new IllegalStateException("Cannot execute task:"
                            + " the task is already running.");
                case FINISHED:
                    throw new IllegalStateException("Cannot execute task:"
                            + " the task has already been executed "
                            + "(a task can be executed only once)");
            }
        }

        mStatus = Status.RUNNING;

        onPreExecute();

        mWorker.mParams = params;
        exec.execute(mFuture);

        return this;
    }複製代碼

這裏呢就會看到首先調用的是onPreExecute的方法了。當exec.execute(mFuture);的時候FutureTask做爲一個參數傳進了sDefaultExecutor的execute方法裏面。sDefaultExecutor是一個串行的SerialExecutor；

private static class SerialExecutor implements Executor {
        final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();
        Runnable mActive;

        public synchronized void execute(final Runnable r) {
            mTasks.offer(new Runnable() {
                public void run() {
                    try {
                        r.run();
                    } finally {
                        scheduleNext();
                    }
                }
            });
            if (mActive == null) {
                scheduleNext();
            }
        }

        protected synchronized void scheduleNext() {
            if ((mActive = mTasks.poll()) != null) {
                THREAD_POOL_EXECUTOR.execute(mActive);
            }
        }
    }複製代碼

咱們能夠看出當exec.execute(mFuture);執行的時候，會將FutureTask加入到mTaks中。不管任務執行如何都將調用schedleNext方法，它會從mTasks中取出FutureTask任務並交給THREAD_POOL_EXECUTOR處理。而後再FutrueTask的run方法執行的時候也活調用傳入的WorkerRunable的call方法

//FutureTask的run方法。
 public void run() {
        if (state != NEW ||
            !U.compareAndSwapObject(this, RUNNER, null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                //這裏調用
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    set(result);
            }
        }
        ........
   }複製代碼

而後將結果傳遞出去：

private Result postResult(Result result) {
        @SuppressWarnings("unchecked")
        Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
                new AsyncTaskResult<Result>(this, result));
        message.sendToTarget();
        return result;
    }複製代碼

這裏會建立一個Message，將消息發送到handler.這裏就解釋了爲何postResult裏面是主線程了。咱們看一下getHandler（）方法。

private static Handler getHandler() {
        synchronized (AsyncTask.class) {
            if (sHandler == null) {
                sHandler = new InternalHandler();
            }
            return sHandler;
        }
    }複製代碼

這個InternalHandler是繼承與Handler 的類，咱們都很清除handler，這裏就再也不詳細的描述了，接着咱們看下全局變量都是什麼意思：

//獲取CPU的數量，咱們能夠借鑑哦
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
   //這裏使用的仍是ThradPoolExecutor，這裏是核心線程的數量。
    private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
    //這是最大線程的數量，跟cpu的數量有關。
    private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
    //閒置線程的存活時間，我記得以前是60s。
    private static final int KEEP_ALIVE_SECONDS = 30;
    //本身定義的ThreadFactory咱們能夠借鑑。
    private static final ThreadFactory sThreadFactory = new ThreadFactory() {
        private final AtomicInteger mCount = new AtomicInteger(1);

        public Thread newThread(Runnable r) {
            return new Thread(r, "AsyncTask #" + mCount.getAndIncrement());
        }
    };
    //定義的任務隊列，這裏是LinkedBlockingQueue，詳情能夠百度一下阻塞隊列。大小是128.
    private static final BlockingQueue<Runnable> sPoolWorkQueue =
            new LinkedBlockingQueue<Runnable>(128);
   //AsyncTask.execute（）最終調用的是THREAD_POOL_EXECUTOR的execute（）方法。咱們能夠看到它是ThreadPoolExecutor 。
    public static final Executor THREAD_POOL_EXECUTOR;

    static {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
                sPoolWorkQueue, sThreadFactory);
        threadPoolExecutor.allowCoreThreadTimeOut(true);
        THREAD_POOL_EXECUTOR = threadPoolExecutor;
    }

    public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
    //Message發送消息的標記位。
    private static final int MESSAGE_POST_RESULT = 0x1;
     //Message發送消息的標記位。
    private static final int MESSAGE_POST_PROGRESS = 0x2;

    private static volatile Executor sDefaultExecutor = SERIAL_EXECUTOR;
    //自定義的Handler
    private static InternalHandler sHandler;

    private final WorkerRunnable<Params, Result> mWorker;
    private final FutureTask<Result> mFuture;

    private volatile Status mStatus = Status.PENDING;

    private final AtomicBoolean mCancelled = new AtomicBoolean();
    private final AtomicBoolean mTaskInvoked = new AtomicBoolean();複製代碼

這裏呢基本上咱們就瞭解AsyncTask 的內部構造了。Android 3.0 以上使用的話，SerialExecutor做爲默認的線程，它將任務串行的處理。咱們想要並行的處理的話須要這樣：

asynctask。executeOnExecutor（自定義線程,""）;複製代碼

三.總結

我相信線程池構造方法知道後咱們就能夠去使用了，使用起來也是很簡單的這裏就不在敘述了，不懂的話你們能夠評論，我會實時回覆的啊~

歡迎你們關注，點個贊吧~：
個人CSDN
個人掘金
個人簡書

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程