本文主要使用kotlin,，討論Android開發(fā)中的線程池用法,。

我們想使用線程的時(shí)候，可以直接創(chuàng)建子線程并啟動(dòng)

Thread { Log.d("rfDev", "rustfisher said: hello") }.start()

不想每次都創(chuàng)建新的子線程
如果有大量的異步任務(wù)，不想每次都創(chuàng)建子線程。有沒有什么把子線程統(tǒng)一管理的方法,？

遇到這樣的情況，我們可以考慮線程池,。線程池解決兩個(gè)問題：需要執(zhí)行大量異步任務(wù)的時(shí)候,，減輕每個(gè)異步任務(wù)的調(diào)用開銷，提高性能,。另外它還能夠限制和管理子線程,。每個(gè)ThreadPoolExecutor都維護(hù)了一些統(tǒng)計(jì)數(shù)據(jù)，例如已執(zhí)行的任務(wù)數(shù)量,。

有大量異步任務(wù)的時(shí)候,，可以考慮使用線程池。

預(yù)置線程池#

代碼參考 Android API 29

ThreadPoolExecutor提供了很多參數(shù),，方便開發(fā)者調(diào)控,。線程池的設(shè)計(jì)者建議開發(fā)者使用以下幾個(gè)工廠方法，Android中主要有5種

• newCachedThreadPool() 不限制數(shù)量的線程池,，能自動(dòng)回收線程

• newFixedThreadPool(int nThreads) 固定數(shù)量的線程池

• newSingleThreadExecutor() 單一的子線程

• newScheduledThreadPool(int corePoolSize) 能執(zhí)行延時(shí)任務(wù)或者周期性任務(wù)

• newWorkStealingPool() 工作竊取線程池

實(shí)際上我們?cè)贏ndroid Studio里輸入Executors.new的時(shí)候,，會(huì)跳出很多個(gè)提示選項(xiàng)。

Executors.new 的智能提示

可緩存線程池#

用Executors.newCachedThreadPool獲得一個(gè)可緩存線程池對(duì)象,，然后讓它執(zhí)行任務(wù)。

val tp: ExecutorService = Executors.newCachedThreadPool()tp.submit { Log.d(TAG, "rustfisher: cached線程池執(zhí)行任務(wù) 3") }

可緩存線程池會(huì)在需要的時(shí)候創(chuàng)建新的子線程,。當(dāng)原有的線程可用的時(shí)候,，會(huì)復(fù)用現(xiàn)有線程。
這個(gè)機(jī)制適用于執(zhí)行多個(gè)短期異步任務(wù),。任務(wù)比較小,，但是數(shù)量大。

調(diào)用execute方法會(huì)先嘗試復(fù)用已有的可用線程,。如果當(dāng)前沒有線程,，會(huì)新建一個(gè)線程并把它添加到池里。
超過60秒沒有使用的線程會(huì)被停止并移除,。因此即便長(zhǎng)時(shí)間不用這個(gè)線程池,，也不會(huì)造成多大的開銷,。

定長(zhǎng)線程池#

使用newFixedThreadPool(int nThreads)示例

val fixedTp: ExecutorService = Executors.newFixedThreadPool(4)fixedTp.submit { Log.d(TAG, "rustfisher 定長(zhǎng)線程池執(zhí)行任務(wù)") }

靜態(tài)方法里傳入了一個(gè)int參數(shù)nThreads，表示最大線程數(shù)量,。
如果當(dāng)前所有線程都在忙,，又有新的任務(wù)添加進(jìn)來。那么任務(wù)會(huì)在隊(duì)列中等待,，直到有可用的線程來處理任務(wù),。

如果有的線程遇到錯(cuò)誤而停止了，要執(zhí)行任務(wù)的話,，會(huì)創(chuàng)建新的線程補(bǔ)上位置,。

池里的線程會(huì)一直存活，直到線程池停止（ExecutorService#shutdown）,。

單一線程池#

val singleTp: ExecutorService = Executors.newSingleThreadExecutor()singleTp.submit { Log.d(TAG, "單一線程池執(zhí)行任務(wù)") }

只擁有1個(gè)子線程,。任務(wù)隊(duì)列不限制任務(wù)數(shù)量。如果線程遇到問題停止了,，接下來又要處理任務(wù)時(shí),，會(huì)新建一個(gè)線程來處理。

它能保證任務(wù)會(huì)按順序處理,，同一時(shí)間只能處理1個(gè)任務(wù),。

單一線程池創(chuàng)建后，不能動(dòng)態(tài)修改線程數(shù)量,。不像newFixedThreadPool(1)的定長(zhǎng)線程池可以修改線程數(shù),。

計(jì)劃任務(wù)線程池#

val scheduleTp: ScheduledExecutorService = Executors.newScheduledThreadPool(3)

計(jì)劃任務(wù)線程池能夠執(zhí)行延遲任務(wù)和周期任務(wù)。

延遲任務(wù)#

需要設(shè)定延時(shí)與時(shí)間單位

scheduleTp.schedule({ Log.d(TAG, "計(jì)劃任務(wù)1 runnable") }, 300, TimeUnit.MILLISECONDS)scheduleTp.schedule(Callable { Log.d(TAG, "計(jì)劃任務(wù)2 callable") }, 400, TimeUnit.MILLISECONDS)

周期任務(wù)#

主要涉及到2個(gè)方法scheduleAtFixedRate和scheduleWithFixedDelay,。

假設(shè)任務(wù)時(shí)間小于周期時(shí)間,，則按給定周期時(shí)間來進(jìn)行。這兩個(gè)方法表現(xiàn)一致,。

假設(shè)任務(wù)執(zhí)行時(shí)間大于周期時(shí)間,，這兩個(gè)方法有點(diǎn)不同

• scheduleAtFixedRate執(zhí)行完上一個(gè)任務(wù)后，用時(shí)超過了周期時(shí)間,，會(huì)立刻執(zhí)行下一個(gè)任務(wù),。

• scheduleWithFixedDelay在上一個(gè)任務(wù)執(zhí)行完畢后，還會(huì)等待周期時(shí)間,，再去執(zhí)行下一個(gè)任務(wù),。

工作竊取線程池#

Android SDK 大于等于24，有一種新的線程池,，暫且稱為“工作竊取線程池”,，或者叫“靈活調(diào)度線程池”。

if (android.os.Build.VERSION.SDK_INT >= android.os.Build.VERSION_CODES.N) {    Executors.newWorkStealingPool()}

線程池維護(hù)足夠的線程來支持給定的并行度（parallelism level）,，可能會(huì)用多個(gè)隊(duì)列來減少爭(zhēng)用,。
并行度對(duì)應(yīng)的是活躍的線程最大數(shù),，或者能處理任務(wù)的線程最大數(shù)。

線程的實(shí)際數(shù)量可能會(huì)動(dòng)態(tài)增減,。工作竊取線程池不保證按提交順序來處理任務(wù),。

執(zhí)行任務(wù)#

執(zhí)行任務(wù)的時(shí)候可以傳入Runnable和Callable，前面用的都是Runnable,。

用Callable的例子

tp.submit(Callable { "OK" })

無返回值任務(wù)的調(diào)用#

無返回值任務(wù)用Callable和Runnable都行,。

val tp: ExecutorService = Executors.newCachedThreadPool()tp.submit { Log.d(TAG, "rustfisher: cached線程池submit runnable") }tp.execute { Log.d(TAG, "rustfisher: cached線程池execute runnable") }tp.submit(Callable { Log.d(TAG, "rustfisher: cached線程池submit callable") })tp.shutdown() // 最后記得用完后停掉線程池

有返回值任務(wù)的調(diào)用#

有返回值的任務(wù)需要Callable接口。

submit#

調(diào)用submit方法時(shí)會(huì)返回一個(gè)Future對(duì)象,。通過Future的get()方法可拿到返回值,。這里需要注意get()是阻塞的，完成任務(wù)后,，能拿到返回值,。

val tp: ExecutorService = Executors.newCachedThreadPool()val future = tp.submit(Callable {    return@Callable "callable的返回值"})Log.d(TAG, "future get之前 isDone: ${future.isDone}, isCancelled: ${future.isCancelled}")val res = future.get()Log.d(TAG, "future get之后 isDone: ${future.isDone}, isCancelled: ${future.isCancelled}")Log.d(TAG, "future get: $res")

運(yùn)行l(wèi)og

future get之前 isDone: false, isCancelled: falsefuture get之后 isDone: true, isCancelled: falsefuture get: callable的返回值

invokeAll#

對(duì)于列表里的任務(wù)，可以使用invokeAll(Collection<? extends Callable<T>> tasks),，返回一個(gè)Future的列表,。
作為對(duì)比，給其中一個(gè)任務(wù)加上延時(shí),。

invokeAll示例

    val tp: ExecutorService = Executors.newFixedThreadPool(5)    val callList = arrayListOf<Callable<String>>(            Callable {                Log.d(TAG, "task1 ${Thread.currentThread()}")                return@Callable "rust"            },            Callable {                Log.d(TAG, "task2 ${Thread.currentThread()}")                Thread.sleep(1500) // 加上延時(shí)                return@Callable "fisher"            },            Callable {                Log.d(TAG, "task3 ${Thread.currentThread()}")                return@Callable "列表里面的任務(wù)"            },    )    Log.d(TAG, "invokeAll 準(zhǔn)備提交任務(wù)")    val futureList = tp.invokeAll(callList)    Log.d(TAG, "invokeAll 已提交任務(wù)")    futureList.forEach { f ->        Log.d(TAG, "任務(wù)列表執(zhí)行結(jié)果 ${f.get()}") // 這里會(huì)阻塞 別在ui線程里get    }

運(yùn)行l(wèi)og,，可以看到提交任務(wù)后，經(jīng)過延時(shí),，拿到了運(yùn)行結(jié)果,。注意看invokeAll前后的時(shí)間。invokeAll會(huì)阻塞當(dāng)前線程,。使用的時(shí)候必須小心,，不要在ui線程中調(diào)用。

    2021-09-11 14:40:07.062 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: invokeAll 準(zhǔn)備提交任務(wù)    2021-09-11 14:40:07.063 16914-19230/com.rustfisher.tutorial2020 D/rfDevTp: task1 Thread[pool-4-thread-1,5,main]    2021-09-11 14:40:07.063 16914-19231/com.rustfisher.tutorial2020 D/rfDevTp: task2 Thread[pool-4-thread-2,5,main]    2021-09-11 14:40:07.063 16914-19232/com.rustfisher.tutorial2020 D/rfDevTp: task3 Thread[pool-4-thread-3,5,main]    2021-09-11 14:40:08.563 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: invokeAll 已提交任務(wù)    2021-09-11 14:40:08.563 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: 任務(wù)列表執(zhí)行結(jié)果 rust    2021-09-11 14:40:08.563 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: 任務(wù)列表執(zhí)行結(jié)果 fisher    2021-09-11 14:40:08.563 16914-16914/com.rustfisher.tutorial2020 D/rfDevTp: 任務(wù)列表執(zhí)行結(jié)果 列表里面的任務(wù)

提交了3個(gè)任務(wù),，在3個(gè)不同的子線程中執(zhí)行,。

invokeAny#

invokeAny(Collection<? extends Callable<T>> tasks)也是接收Callable集合。
然后返回最先執(zhí)行結(jié)束的任務(wù)的值,，其它未完成的任務(wù)將被正常取消掉不會(huì)有異常,。

invokeAny示例

    val tp: ExecutorService = Executors.newCachedThreadPool()    val callList = arrayListOf<Callable<String>>(            Callable {                Thread.sleep(1000) // 設(shè)計(jì)延時(shí)                return@Callable "rust"            },            Callable {                Thread.sleep(400)                return@Callable "fisher"            },            Callable {                Thread.sleep(2000)                return@Callable "列表里面的任務(wù)"            },    )    Log.d(TAG, "invokeAny 提交任務(wù)")    val res = tp.invokeAny(callList)    Log.d(TAG, "執(zhí)行結(jié)果 $res")

    2021-09-11 14:04:55.253 14066-14066/com.rustfisher.tutorial2020 D/rfDevTp: invokeAny 提交任務(wù)    2021-09-11 14:04:55.654 14066-14066/com.rustfisher.tutorial2020 D/rfDevTp: 執(zhí)行結(jié)果 fisher

觀察log可以看到，最后執(zhí)行的是“fisher”這個(gè)任務(wù),。

停止線程池#

使用完畢后，記得終止線程池

/*ExecutorService*/ shutdown()shutdownNow()

shutdown()在已提交的任務(wù)后面創(chuàng)建一個(gè)停止命令,，并且不再接受新的任務(wù),。如果線程池已經(jīng)停止了，調(diào)用這個(gè)方法將不生效,。

shutdownNow()方法嘗試停止所有執(zhí)行中的任務(wù),，停下等待中的任務(wù),。并且返回等待執(zhí)行的任務(wù)列表List<Runnable>。