一、前言
前文链接：深度解读AsyncTask
上一篇文章我们介绍了AsyncTask的相关知识点，并就其存在的问题做了深入的探讨。
AsyncTask总的来说实现简单，构思精巧，使用方便；但因其保守的设计，在通用性方面有较大局限。
不过，有局限，才有突破。
接下来，我们将结合APP开发中的使用场景，探讨如何设计一个更强的异步任务框架。
二、任务调度
前文提到，AsyncTask主要做了两项工作：流程控制和任务调度。
其中任务调度方面主要依赖于两个Executor：ThreadPoolExecutor 和 SerialExecutor 。
对于ThreadPoolExecutor而言，如果workQueue是容量较大的的队列，则基本上coreSize就是并发窗口（可以并发执行的线程数量）。
coreSize太小，CPU利用率不高，吞吐率低，不适用于IO密集型任务（比如网络请求）；
coreSize太大，如果执行的是计算密集型任务，线程切换频繁，CPU计算饱和（影响UI线程运算）。
所以，我们先从任务调度入手，解决“利用率/吞吐率”两难的问题。
同时探索一下在任务调度方面还有什么可以完善的。
2.1 并发控制
如何做到既适用“IO密集型”任务，又适用“计算密集型”任务呢？
用两个线程池吗？
倒是可以，但两个线程池会各自维护线程，彼此不能复用。
说到复用，AsyncTask给我们提供了一种思路：
先定义一个线程池THREAD_POOL_EXECUTOR，并行任务可以调用此Executor来执行；
封装SerialExecutor，加了一个任务队列，控制加入的任务串行执行，但是最终任务还是运行在THREAD_POOL_EXECUTOR上。
于是，调用者可以选择串行或者并行，且是在同一个线程池中调度的，线程可以复用。
就此思路，我们可以仿照SerialExecutor，在线程池ThreadPoolExecutor之上，封装一个Executor来控制并发，同时创建不同并发窗口的Executor。
简单地说，就是给水池接不同的水管，不同的水管口径可以不一样。
在造管道之前，我们先准备一些其他工具。
前面的SerialExecutor的代码中，有两个重要的构成部分：
1、Runnable的包装器（虽然是匿名的）；
2、下一个任务的触发器（虽然只是个方法）。
我们把这两部分都抽象出来：
interface Trigger {
fun next()
}
class RunnableWrapper constructor(
private val r: Runnable,
private val trigger: Trigger) : Runnable {
override fun run() {
try {
r.run()
} finally {
trigger.next()
}
}
}
现在我们先看下如何建造这条“管道”，既然比作“管道”，且命名为PipeExecutor吧。
class PipeExecutor @JvmOverloads constructor(
windowSize: Int,
private val capacity: Int = -1,
private val rejectedHandler: RejectedExecutionHandler = defaultHandler) : TaskExecutor {
private val tasks = PriorityQueue<RunnableWrapper>()
private val windowSize: Int = if (windowSize > 0) windowSize else 1
private var count = 0
private val trigger : Trigger = object : Trigger {
override fun next() {
scheduleNext()
}
}
override fun execute(r: Runnable) {
schedule(RunnableWrapper(r, trigger), Priority.NORMAL)
}
fun execute(r: Runnable, priority: Int) {
schedule(RunnableWrapper(r, trigger), priority)
}
@Synchronized
internal fun scheduleNext() {
count--
if (count < windowSize) {
startTask(tasks.poll())
}
}
@Synchronized
internal fun schedule(r: RunnableWrapper, priority: Int) {
if (capacity > 0 && tasks.size() >= capacity) {
rejectedHandler.rejectedExecution(r, TaskCenter.poolExecutor)
}
if (count < windowSize || priority == Priority.IMMEDIATE) {
startTask(r)
} else {
tasks.offer(r, priority)
}
}
private fun startTask(active: Runnable?) {
if (active != null) {
count++
// poolExecutor 是 ThreadPoolExecutor
TaskCenter.poolExecutor.execute(active)
}
}
}
解析一下代码中的参数和变量：
- tasks：任务缓冲区
- count：正在执行的任务的数量
- windowSize：并发窗口，控制Executor的并发
- capacity：任务缓冲区容量，小于等于0时为不限容量，超过容量触发rejectedHandler
- rejectedHandler：默认为AbortPolicy(抛出异常）
- priority：调度优先级
object Priority {
const val IMMEDIATE = Integer.MAX_VALUE
const val HIGH = 1
const val NORMAL = 0
const val LOW = -1
}
当count>=windowSize时，priority高者先被调度；
优先级相同的任务，遵循先进先出（FIFO）的调度规则；
priority=IMMEDIATE会跳过缓冲区直接进入线程池。
需要注意的是，调度优先级不同于线程优先级，线程优先级更底层一些。
比如AsyncTask的doInBackground()中就调用了：
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND)
这可以使得后台线程的线程优先级低于UI线程。
以下是PipeExecutor的流程图：
使用时，可以实例化多个PipeExecutor，他们各自根据参数调度自己的任务队列，而所有任务最终都是在同一个线程池中运行。
比方说可以创建windowSize小一点的PipeExecutor，用于计算密集型任务；
也可以创建windowSize大一点的PipeExecutor，用于IO密集型任务；
还可以使windowSize=1，用于串行执行。
2.2 任务去重
这一节可能相对难理解一点，不过我会想办法尽量表述清楚一些的。
要说去重，首先要定义重复：
做任务其实就是做计算，其模式可以表示为 R=f (X)；
如果任务A和任务B都是计算R=f (X)， 且X相等，那么我们说A和B是“重复任务”。
两个任务的计算是等价的，我们可以通过“去重”来节约运算。
什么时候会出现“重复任务”呢？
举两个例子：
一、数据更新，刷新页面
假设一个页面的数据项(R)有多个数据源(X=[x1,x2…xn]), 即R=f(x1,x2…xn);
任何一个数据源更新， R都要重新计算。
通常“数据更新->计算数据->刷新界面” 会采用“发布订阅模式”：
数据源更新发送事件，接收到事件，启动任务计算R。
假如x1,x2…xn短时间内相继更新了数据，发送了事件e1,e2...en，然后会几乎同时启动N个任务计算R。
- 如果任务并行，N个线程并发计算R=f(x1,x2…xn)，浪费计算资源，更不用说多线程问题的复杂性了；
所以这样的任务最好不要并发执行。 - 如果任务串行，若R的计算比较耗时，第一个任务还没计算完，又来了n-1个任务。
其实所有的任务都是计算R=f(x1,x2…xn)，对于后面的n-1个任务而言，是等价的。
因为在第一个任务还在计算的时候，x1,x2…xn都更新完了，没有再更新了，
所以对于后面的任务X是一样的，当然R=f(X)也是一样的了，这就是“重复任务”。
当然，对应的方法就是：对于后面的任务，只保留一个就好了。
除了这个方案，我们来看下其他方案：
- 丢弃后面所有的任务。
如果第一个任务在读取数据X之后x2...xn才更新，那第一个任务的(x1,x2…xn)和后面的任务就不一样；
如果直接丢弃后面的任务，很显然，R得不到正确更新。 - 取消之前的任务，直接计算后面的任务。
首先，即使任务可以取消（马上停止），x2取消第一个任务，x3取消第二个任务……则界面迟迟得不到更新。
最重要的是，任务不是说取消就取消的。
上一篇有讨论，Thread.stop()不安全， interrupt()只能中断wait(), sleep()等，对计算型任务是中断不了的。
二、图片加载
这个例子比上一个简单很多：多个ImageView需要加载同一张图片。
这其实也出现了“重复任务”，因为是同一张图片，所以R=f(X)等价。
和那个例子不同的是，这里有多个目标(Target)。
如果说前面的例子是 R->T的话，这里则是{R->T1, R->T2 ... R->Tn}。
两种场景，去重策略有所相同，也有所不同。
相同之处是：串行；
不同之处是：不能丢弃任务（不然有的ImageView得不到更新）。
- 串行的话不会慢吗？
不会，因为图片加载通常都有缓存策略，第一个任务解码图片之后，放入缓存，后面的任务读取缓存即可。 - 如果是不同的图片呢？
嗯，关键就在这：相同的任务串行，不同的任务并行。 - 怎么看任务是不是“相同的任务”？
这就是需要给任务一个“标签(tag)”, tag相同则为相同的任务。
比方说下载任务的话可以将url作为tag。
任务去重的实现如下：
class LaneExecutor(private val executor: PipeExecutor,
private val discard: Boolean = false) : TaskExecutor {
// 正在调度的任务
private val scheduledTasks = HashMap<String, Runnable>()
// 丢弃模式下等待的任务
private val waitingTasks by lazy { HashMap<String, TaskWrapper>() }
// 非丢弃模式下等待的任务
private val waitingQueues by lazy { HashMap<String, CircularQueue<TaskWrapper>>() }
private class TaskWrapper(val r: Runnable, val priority: Int)
private inner class LaneTrigger(val tag : String) : Trigger{
override fun next() {
executor.scheduleNext()
scheduleNext(tag)
}
}
private fun start(r: Runnable, tag: String, priority: Int) {
scheduledTasks[tag] = r
executor.schedule(RunnableWrapper(r, LaneTrigger(tag)), priority)
}
@Synchronized
fun scheduleNext(tag: String) {
scheduledTasks.remove(tag)
if (discard) {
waitingTasks.remove(tag)?.let { start(it.r, tag, it.priority) }
} else {
waitingQueues[tag]?.let { queue ->
val wrapper = queue.poll()
if (wrapper == null) {
// 如果队列清空了，则顺便把队列从HashMap移除，不然HashMap只增不减
waitingQueues.remove(tag)
} else {
start(wrapper.r, tag, wrapper.priority)
}
}
}
}
@Synchronized
fun execute(tag: String, r: Runnable, priority: Int = Priority.NORMAL) {
if (tag.isEmpty()) {
executor.execute(r, priority)
} else if (!scheduledTasks.containsKey(tag)) {
start(r, tag, priority)
} else if (discard) {
if (waitingTasks.containsKey(tag)) {
// 丢弃模式下，如果有相同的任务在等待，则丢弃传进来的任务
// 而如果传进来的又是 Futures(实现了Runnable), 则顺便调用其cancel()方法
if (r is Future<*>) {
r.cancel(false)
}
} else {
waitingTasks[tag] = TaskWrapper(r, priority)
}
} else {
// 非丢弃模式下，每个tag都有一个无界队列可以缓存任务
val queue = waitingQueues[tag]
?: CircularQueue<TaskWrapper>().apply { waitingQueues[tag] = this }
queue.offer(TaskWrapper(r, priority))
}
}
}
构造函数有两个参数：
executor：PipeExecutor的实例。
discard：当discard=true时，只留一个等待的任务; 否则, 不丢弃任务 。
LaneExecutor的实现和PipeExecutor有些相似的，两者都有缓冲任务的容器，而执行流程上，
大体都是走了 “execute -> start -> 转发给另一个Executor -> 执行结束 -> scheduleNext ” 这样一个流程。
区别在于，LaneExecutor由于要“标识任务”，所以有一个tag参数贯穿整个流程，连容器都是以tag为key的HashMap。
关于组合和继承，普遍的观点是组合优先于继承。
所以在设计LaneExecutor时，用PipeExecutor作为成员而非继承于PipeExecutor。
LaneExecutor自己实现任务去重，然后将任务转发给PipeExecutor。
他们的关系示意图如下( 分两种模式，分别对应前面提到的两种场景）：
discard=true
discard=false
洋葱似地一层包一层，很明显，也是装饰者模式。
这样的实现估计大家在用各种 InputStream 和 OutputStream 时已经领略了。
职责分配：
LaneExecutor负责任务去重；
PipeExecutor负责任务并发控制和调度优先级；
ThreadPoolExecutor负责分配线程来执行任务。
还有就是，为什么命名为LaneExecutor呢？
Lane有车道的意思（泳道也是这个词），看示意图，是不是有点像车道？
总结一下LaneExecutor的特点：
- 相同的任务串行，不同的任务并行
- discard=true, 串行的任务，各自(按tag分组）最多只能有一个任务等待，再有提交会被丢弃；
- discard=false, 每个tag分组都有一个无界队列缓冲，不会丢弃任务。
2.3 统一定义Executor
当项目复杂度到了一定程度，如果没有明确的公共定义，可能会出现各种冗余对象，比如缓存和Executor。
分散的Executor无法较好地控制并发；
如果各自创建的是ThreadPoolExecutor，则还要加上一条：降低线程复用。
故此，可以集中定义Executor，各模块统一调用。
代码如下：
object TaskCenter {
private val cpuCount = Runtime.getRuntime().availableProcessors()
// ... 定义threadFactory， 代码省略
internal val poolExecutor: ThreadPoolExecutor = ThreadPoolExecutor(
0, 256,
60L, TimeUnit.SECONDS,
SynchronousQueue(),
threadFactory)
// 常规的任务调度器，可控制任务并发，支持任务优先级
val io = PipeExecutor(20, 512)
val computation = PipeExecutor(Math.min(Math.max(2, cpuCount), 4), 512)
// 带去重策略的 Executor，可用于数据刷新等任务
val laneIO = LaneExecutor(io, true)
val laneCP = LaneExecutor(computation, true)
// 相同的tag的任务会被串行执行，相当于串行的Executor
// 可用于写日志，上报app信息等任务
val serial = LaneExecutor(PipeExecutor(Math.min(Math.max(2, cpuCount), 4), 1024))
}
三、拓展AsyncTask
上一章我们花了大量的篇幅讲述任务调度的种种细节，构造了相对完善的Executor（系列）。
这一章我们将结合前面的工作，以AsyncTask为蓝本，实现一个更强大的异步任务框架。
通过继承AsyncTask无法做到我们预想的效果，所以没办法，只能重新写一个了，反正代码也不多。
虽然是重新写，但原来的绝大部分实现和API都会得到保留。
当然名字也要另起一个，不然就和真正的AsyncTask冲突了；
且命名为UITask，因为和纯粹的线程不同，这个异步框架还要和UI线程交互。
3.1 替换Executor
前面实现的PipeExecutor和LaneExecutor，可以用到UITask中。
实现如下：
abstract class UITask<Params, Progress, Result> {
private val mFullName: String = this.javaClass.name
private var mPriority = Priority.NORMAL
private val mTag: String by lazy { generateTag() }
protected open fun generateTag(): String {
return mFullName
}
protected open val executor: TaskExecutor
get() = TaskCenter.laneIO
fun execute(vararg params: Params) {
if (executor is LaneExecutor) {
(executor as LaneExecutor).execute(mTag, mFuture, mPriority)
} else {
(executor as PipeExecutor).execute(mFuture, mPriority)
}
}
fun priority(priority: Int): UITask<Params, Progress, Result> {
var p = priority
if (priority != Priority.IMMEDIATE) {
if (priority > Priority.HIGH) {
p = Priority.HIGH
} else if (priority < Priority.LOW) {
p = Priority.LOW
}
}
mPriority = p
return this
}
}
代码中，executor默认为TaskCenter.laneIO，因为日常使用中用于数据加载的比较多；
声明了"open", 也是就可以通过override来设定需要的TaskExecutor。
3.2 生命周期
上一篇文章我们提到AsyncTask的问题，其中一个就是在Activity销毁时不会自动取消，当然也可以做到，只是写起来麻烦。
那么我们就在UITask封装一些代码，使其可以观察Activity/Fragment的生命周期。
说到观察，很自然地就想到了“观察者模式”来实现。
关系图如下：
UITask为观察者，Activity/Fragment为被观察者。
因为是多对多的关系，所以需要两个数据结构：一个SparseArray（Map也可以）一个列表。
SparseArray的key为被观察者的identityHashCode， value为观察者（UITask）列表。
当被观察者需要通知事件的时候，再次获取被观察者的identityHashCode，索引到对应观察者列表，遍历之。
具体实现如下：
object LifeEvent {
const val DESTROY = 0
const val SHOW = 1
const val HIDE = 2
}
interface LifeListener {
fun onEvent(event: Int)
}
object LifecycleManager {
private val holders = SparseArray<Holder>()
fun register(hostHash: Int, listener: LifeListener?) {
var holder: Holder? = holders.get(hostHash)
if (holder == null) {
holder = Holder()
holders.put(hostHash, holder)
}
holder.add(listener)
}
fun unregister(hostHash: Int, listener: LifeListener?) {
holders.get(hostHash)?.remove(listener)
}
fun notify(host: Any?, event: Int) {
val hostHash = System.identityHashCode(host)
val index = holders.indexOfKey(hostHash)
if (index >= 0) {
val holder = holders.valueAt(index)
if (event == LifeEvent.DESTROY) {
holders.removeAt(index)
}
holder.notify(event)
}
}
}
LifecycleManager是一个事件枢纽：
一方面，Activity/Fragment生命周期回调时通过LifecycleManager.notify发布事件；
另一方面，UITask需要通过LifecycleManager注册，建立和Activity/Fragment的关系。
代码如下：
abstract class UITask<Params, Progress, Result> : LifeListener {
private var mHostHash = 0
fun host(host: Any): UITask<Params, Progress, Result> {
mHostHash = System.identityHashCode(host)
LifecycleManager.register(mHostHash, this)
return this
}
private fun detachHost() {
LifecycleManager.unregister(mHostHash, this)
}
override fun onEvent(event: Int) {
if (event == LifeEvent.DESTROY) {
if (!isCancelled && status != Status.FINISHED) {
cancel(true)
}
} else if (event == LifeEvent.SHOW) {
changePriority(+1)
} else if (event == LifeEvent.HIDE) {
changePriority(-1)
}
}
private fun changePriority(increment: Int) {
if (mPriority != Priority.IMMEDIATE) {
mPriority = executor.changePriority(mFuture, mPriority, increment)
}
}
}
- host(Any)方法用于注册观察者，也就是构建host和Task的关系。
为什么命名为host呢？因为Task通常是在Activity/Fragment中创建（不然我们也不用大费周章折腾生命周期了），
这时候我们称Activity/Fragment为“宿主(host)”。 - detachHost() 为私有方法，因为这个方法是在UITask执行完成的时候被调用的（内部调用）。
- onEvent(Int)函数关注三个事件，前面也提到，除了DESTROY之外，还关注SHOW和HIDE，
主要是在Activity/Fragment的可见状态改变时调整调度优先级。
调整优先级有什么用呢？ 下面先看两张图感受一下。
为了凸显效果，我们把加载任务的并发量控制为1（串行）。
第一张是不会自动调整优先级的，完全的先进先出：
不改变优先级
可以看到，切换到第二个页面，由于上一页的任务还没执行完，
所以要一直等到上一页的任务都完成了才轮到第二个页面加载。
很显然这样体验不太好。
接下来我们看下动态调整优先级是什么效果：
动态调整优先级
切换到第二个页面之后，第一个页面的任务的“调度优先级”被降低了，所以会优先加载第二个页面的图片；
再次切换回第一个页面，第二个页面的优先级被降低，第一个页面的优先级恢复，所以优先加载第一个页面的图片。
那可否进入第二个页面的时暂停第一个页面的任务？
暂停的方案不太友好，比方说用户在第二个页面停留很久，第二个页面的任务都完成了，然后切换回第一个页面，发现只有部分图片（其他被暂停了）。
而如果只是调整优先级，则第二个页面的任务都执行完之后，会接着执行第一个页面的任务，返回第一个页面时就能够看到所有图片了。
这就好比赶车，让其他人给插个队，OK，但是不能不给别人排队了吧~
四、用法
4.1 常规用法
override fun onCreate(savedInstanceState: Bundle?) {
// ...
TestTask()
.priority(Priority.IMMEDIATE)
.host(this)
.execute("hello")
}
private inner class TestTask: UITask<String, Integer, String>(){
override fun generateTag(): String {
// 一般情况下不需要重写这个函数，这里只是为了演示
return "custom tag"
}
override fun onPreExecute() {
result_tv.text = "running"
}
override fun doInBackground(vararg params: String): String? {
for (i in 0..100 step 2) {
// do something
publishProgress(Integer(i))
}
return "result is：" + params[0].toUpperCase()
}
override fun onProgressUpdate(vararg values: Integer) {
val progress = values[0]
progress_bar.progress = progress.toInt()
progress_tv.text = "$progress%"
}
override fun onPostExecute(result: String?) {
result_tv.text = result
}
override fun onCancelled() {
showTips("Task cancel ")
}
}
UITask和AsyncTask用法是类似的, 只是多了一些API：
- 因为生命需要观察Activity的生命周期，所以需要调用host()，传入当前Activity
- 可以设置任务优先级
- 有必要时可以重写generateTag来自定义任务的tag
UITask相比AsyncTask，虽然外表看起来区别不大，但内核却有质的改变：
1、更灵活的并发控制
2、支持调度优先级
3、支持任务去重
4、支持生命周期（onDestroy时取消任务，自动调整优先级）
这些特性，在笔者的另一个开源项目中都用上了。
文章链接：如何实现一个图片加载框架
4.2 Executor
当然，项目中不仅仅是UITask，TaskCenter，以及各种Executor, 都是可以单独使用的。
比方说只是想简单地执行任务，不需要和UI交互，也可以直接使用Executor：
TaskCenter.io.execute{
// do something
}
TaskCenter.laneIO.execute("laneIO", {
// do something
})
val serialExecutor = PipeExecutor(1)
serialExecutor.execute{
// do something
}
TaskCenter.serial.execute ("your tag", {
// do something
})
4.3 For RxJava
有的文章拿AsyncTask和RxJava做比较，一个300行代码的框架和一个2M的框架，其实也没有太多可比性。
如果说AsyncTask是自行车，“加强版”是摩托车，则RxJava就是汽车（RxJava除了异步还有更多的内涵）。
既然有汽车，为什么还要造摩托？
打造这个“加强版”的初衷是提供更好的异步任务框架，而不是替代RxJava。
摩托和汽车，各有各的灵魂。
虽然摩托车和汽车有较大差异，但取摩托车的汽油来跑汽车也是可以的：
object TaskSchedulers {
val io: Scheduler by lazy { Schedulers.from(TaskCenter.io) }
val computation: Scheduler by lazy { Schedulers.from(TaskCenter.computation) }
val single by lazy { Schedulers.from(PipeExecutor(1)) }
}
Observable.range(1, 8)
.subscribeOn(TaskSchedulers.computation)
.subscribe { Log.d(tag, "number:$it") }
很多开源项目都设计了API来使用外部的Executor，这样有一个好处：
各种任务都在一个线程池上执行任务，可复用彼此创建的线程。
4.4 彩蛋
喜欢冰糖葫芦一样的链式调用？
override fun onCreate(savedInstanceState: Bundle?) {
// ...
val task = ChainTask<Double, Int, String>()
task.tag("ChainTest")
.preExecute { result_tv.text = "running" }
.background { params ->
for (i in 0..100 step 2) {
Thread.sleep(10)
task.publishProgress(i)
}
"result is：" + (params[0] * 100)
}
.progressUpdate { values ->
val progress = values[0]
progress_bar.progress = progress
progress_tv.text = "$progress%"
}
.postExecute { result_tv.text = it }
.cancel { showTips("ChainTask cancel ") }
.priority(Priority.IMMEDIATE)
.host(this)
.execute(3.14)
}
至于ChainTask是怎么实现的，本文就不多做介绍了，且留给读者自行思考；
或者下载项目，里面有具体的实现。
五、下载
implementation 'com.horizon.task:task:1.0.4'
相关代码已上传GitHub,
项目地址：https://github.com/No89757/Task
网友评论