您可以捐助,支持我们的公益事业。

1元 10元 50元





认证码:  验证码,看不清楚?请点击刷新验证码 必填



  求知 文章 文库 Lib 视频 Code iProcess 课程 角色 咨询 工具 火云堂 讲座吧   建模者  
会员   
 
   
 
  
每天15篇文章
不仅获得谋生技能
更可以追随信仰
 
     
   
 订阅
  捐助
加强版异步任务框架
 
100 次浏览     评价:  
2019-5-16
 
编辑推荐:
本文来自于cnblogs,文章主要是作者以AsyncTask为范本,介绍了了一个“加强版的异步任务框架”。

一、前言

为了提高流畅性,耗时任务放后台线程运行,已是APP开发的常识了。

关于异步有很多方案,当前最流行的,莫过于RxJava了;

更早一些时候,还有AsyncTask(骨灰级的API)。

总的来说,AsyncTask构思精巧,代码简洁,使用方便,有不少地方值得借鉴。

当然问题也有不少,比如不能随Activity销毁而销毁导致的内存泄漏,还有不适合做长时间的任务等。

笔者以AsyncTask为范本,写了一个“加强版的异步任务框架”:

保留了AsyncTask的所有用法,解决了其中的一些问题,同时引入了一些新特性。

接下来给大家介绍一下这“加强版”的框架,希望对各位有所启发。

二、任务调度

2.1 AsyncTask的Executor

AsyncTask的任务调度主要依赖两个Executor:
ThreadPoolExecutor 和 SerialExecutor。

代码如下:

private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;

public static final Executor THREAD_POOL_EXECUTOR;
public static final Executor SERIAL_EXECUTOR = new SerialExecutor();

static {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, 30, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(128), sThreadFactory);
threadPoolExecutor.allowCoreThreadTimeOut(true);
THREAD_POOL_EXECUTOR = threadPoolExecutor;
}

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);
}
}
}

关于线程池,估计大家都很熟悉了,参数就不多作解释了。

上面代码中,通过巧用“装饰者模式”,增加“串行调度”的功能。

装饰者模式有以下特点:装饰对象和真实对象有相同的接口,这样客户端对象就能以和真实对象相同的方式和装饰对象交互。

装饰对象包含一个真实对象的引用。

装饰对象接受所有来自客户端的请求,它把这些请求转发给真实的对象。

装饰对象可以在转发这些请求以前或以后增加一些附加功能。SerialExecutor只有二十来行代码,却用了两次装饰者模式:Runnable和Executor。

Runnable部分,往队列添加的匿名Runnable对象(装饰对象),当被Executor调用run()方法时,先执行“真实对象”的run()方法,然后再调用scheduleNext();

Executor部分,通过增加一个任务队列,实现串行调度的功能,而具体的任务执行转发给“真实对象”THREAD_POOL_EXECUTOR。

想要串行调度,为什么不多加一个coreSize=1的
ThreadPoolExecutor呢?

两个ThreadPoolExecutor,彼此线程不可复用。

虽然SerialExecutor的方案很不错,但是THREAD_POOL_EXECUTOR的coreSize太小了(不超过4),

这导致AsyncTask不适合执行长时间运行的任务,否则多几个任务就会堵塞。

因此,如果要改进AsyncTask,首先要改进Executor。

2.2 通用版Executor

实现思路和 SerialExecutor 差不多,加一个队列, 实现另一层调度控制。

首先,把 Runnable 和 scheduleNext 两部分都抽象出来:

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()
}
}
}

接下来的实现和SerialExecutor类似:

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()
}
}

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++
TaskCenter.poolExecutor.execute(active)
}
}
}

解析一下代码中的参数和变量:tasks:任务缓冲区

count:正在执行的任务的数量

windowSize:并发窗口,控制Executor的并发

capacity:任务缓冲区容量,小于等于0时为不限容量,超过容量触发rejectedHandler

rejectedHandler:默认为AbortPolicy(抛出异常)

priority:调度优先级

当count>=windowSize时,priority高者先被调度;

优先级相同的任务,遵循先进先出(FIFO)的调度规则。

需要注意的是,调度优先级不同于线程优先级,线程优先级更底层一些。

比如AsyncTask的doInBackground()中就调用了:

Process.setThreadPriority(Process. THREAD_PRIORITY_BACKGROUND)

这可以使得后台线程的线程优先级低于UI线程。

以下是PipeExecutor的流程图:

定义了PipeExecutor了之后,我们可以实现多个实例。

例如,可以仿照 RxJava 的 Schedulers,定义适用于“IO密集型”任务和“计算密集型”任务的Executor。

val io = PipeExecutor(20, 512)
val computation = PipeExecutor(Math.min(Math.max(2, cpuCount), 4), 512)

也可以定义串行调度的Executor:

val single = PipeExecutor(1)

不过我们不建议定义全局的串行调度Executor,因为会有相互阻塞的风险。

但是可以根据场景定义专属的串行调度Executor,比如给日志收集创建一个,给数据上报创建一个……

不同实例,犹如不同的水管,往同一个池子进水,故而命名为PipeExecutor。

2.3 去重版Executor

我们项目中,页面更新用的是“发布订阅模式”:

数据层有变更,发布更新消息;

上层收到消息,异步加载数据,刷新页面。

然后就碰到一个问题:若短时间内有多次数据更新,就会有多个消息发往上层。

不做特殊处理,就会几乎同时启动多个异步任务,浪费计算资源;

多个线程对并发读取同一数据,多线程问题也随之而来,若处理不好,结果不可预知。

用串行执行器?所有任务串行的话,无法利用任务并发的优势。

所以经过比较多种方案,最终的结论是:

1、任务分组,不同组并行,同组串行

2、同组的任务,如果有任务在执行,最多只能有一个在等待,丢弃后面的任务

所谓分组,就是给任务打tag, 比如刷新A数据的任务叫ATask, 刷新B任务的叫BTask。

关于第2点,其实有考虑过其他一些方案,比如下面两个:

取消正在执行的任务

首先不是所有任务都可以中断的,可以不接收其结果,但是不一定能中断其执行

即使能取消(比如中断网络请求),也不是最佳方案。

比方说当前线程或许已经快要下载完了,在等一会后面的任务就可以读缓存去结果了;

任务2取消任务1,任务3取消任务2……等到最后一个任务执行,用户可能已经不耐烦了。如果有任务在执行,丢弃后面的任务

比方说任务1读取了数据,在计算的时候,数据源变更,然后发送事件,启动任务2……

直接丢弃后面的任务,最终页面显示的是旧的数据。

我们定义了一个LaneExecutor来实现这个方案,示意图如下:

discard=true

各组任务就像一个个车道(Lane), 故而命名为LaneExecutor。

洋葱似地一层包一层,很明显,也是装饰者模式。

职责分配:

LaneExecutor负责任务去重;

PipeExecutor负责任务并发控制和调度优先级;

ThreadPoolExecutor负责分配线程来执行任务。

但后来又遇到另一个问题:

有多个控件要加载同一个URL的数据,然后很自然地我们就以 URL作为tag了,以避免重复下载(做有缓存,第一任务下载完成之后,后面的任务可以读取缓存)。

但是用LaneExecutor来执行时,只保留一个任务在等待,然后最终只有两个控件能显示数据。

查到问题后,笔者给LaneExecutor加了一种模式,该模式下,不丢弃任务。

discard=false

如此,所有任务都会被执行,但是只有第一个需要下载数据,后面任务读缓存就好了。

2.4 统一管理Executor

当项目复杂度到了一定程度,如果没有统一的公共定义,可能会出现各种冗余实例。

分散的Executor无法较好地控制并发;

如果各自创建的是ThreadPoolExecutor,则还要加上一条:降低线程复用。

故此,可以集中定义Executor,各模块统一调用。

代码如下:

object TaskCenter {
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
// 可用于写日志,上报统计信息等任务
val serial = LaneExecutor(PipeExecutor(Math.min(Math.max(2, cpuCount), 4), 1024))
}

2.5 Executor的使用

TaskCenter.io.execute{
// do something
}

TaskCenter.laneIO.execute("laneIO", {
// do something
}, Priority.HIGH)

val serialExecutor = PipeExecutor(1)
serialExecutor.execute{
// do something
}

TaskCenter.serial.execute ("your tag", {
// do something
})

PipeExecutor的使用和常规的Executor是一样的,execute中传入Runnable即可,

然后由于Runnable只有一个方法,也没有参数,lambda的形式就显得更加简洁了。

LaneExecutor由于要给任务打tag, 所以要传入tag参数;

如果不传,则没有分组的效果,也就是回退到PipeExecutor的特性;

两种Executor都可以传入优先级。

很多开源项目都设计了API来使用外部的Executor,比如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") }

这样有一个好处,各种任务都在一个线程池上执行任务,可复用彼此创建的线程。

三、流程控制

3.1 AsyncTask的执行流

上一章我们分析了任务调度,构造了一系列Executor,增强任务处理方面的通用性。

不过任务调度只是AsyncTask的一部分,AsyncTask的精髓其实在于流程控制:在任务执行的不同阶段,回调相应的方法。

下面是AsyncTask的流程图:

通过使用FutureTask和Callable,使得AsyncTask具备对任务执行更强的控制力,比如cancel任务。

有的文章说cancel()不一定的立即中断任务,但其实Futuret.cancel()确实已经是最好的方案了,

如果强行调用Thread.stop(),则犹如关掉空中飞机的引擎,后果不堪设想。

通过与Handler的配合,AsyncTask可以在任务执行过程中和执行结束后发布数据到UI线程,

这使得AsyncTask尤其适用于“数据加载+界面刷新”的场景。

而这类场景在APP开发中较为常见,这也是AsyncTask一度被广泛使用的原因之一。

3.2 生命周期

AsyncTask其中一个广为诟病的问题就是内存泄漏:

若AsyncTask持有Activity引用,且生命周期比Activity的长,则Activity无法被及时回收。

这个问题其实不是AsyncTask独有,Handler,RxJava等都存在类似问题。

解决方案有多种,静态类、弱引用、Activity销毁时取消等。

RxJava提供了dispose方法来取消任务,同时也有很多集成生命周期的开源方案,比如RxLifecycle、AutoDispose等。

AsyncTask也提供了cancel方法,但是比较命苦,吐槽者众,助力者寡。

其实要实现自动cancel不难,建立和Activity/Fragment的关系即可,可通过观察者模式来实现。

UITask是参考AsyncTask写的一个类, 使用了上一章介绍的Executor。

结构上,UITask为观察者,Activity/Fragment为被观察者,LifecycleManager为 UITask 和 Activity/Fragment 构建关系的桥梁。

实现上需要两个数据结构:一个SparseArray,一个List。

SparseArray的key为被观察者的identityHashCode, value为观察者列表。

UITask提供了host()方法,方法中获取宿主(也就是Activity/Fragment)的identityHashCode,

通过register()方法,添加 “Activity->UITask” 到SparseArray中。

abstract class UITask<Params, Progress, Result> : LifeListener {
fun host(host: Any): UITask<Params, Progress, Result> {
LifecycleManager.register(System.identityHashCode(host), this)
return this
}

override fun onEvent(event: Int) {
if (event == LifeEvent.DESTROY) {
cancel(true)
} else if (event == LifeEvent.SHOW) {
changePriority(+1)
} else if (event == LifeEvent.HIDE) {
changePriority(-1)
}
}
}

override fun onCreate(savedInstanceState: Bundle?) {
TestTask().host(this).execute("hello")
}

 

需要在BaseActivity中通知事件:

abstract class BaseActivity : Activity() {
override fun onDestroy() {
super.onDestroy()
LifecycleManager.notify(this, LifeEvent.DESTROY)
}

override fun onPause() {
super.onPause()
LifecycleManager.notify(this, LifeEvent.HIDE)
}

override fun onResume() {
super.onResume()
LifecycleManager.notify(this, LifeEvent.SHOW)
}
}

调用notify()方法时,会根据Activity索引到对应观察者列表,然后遍历列表,回调观察者onEvent()方法。

其中,当通知的事件为DESTROY时,UITask执行cancel()方法,从而取消任务。

3.3 动态调整优先级

上一节,我们看到UITask除了关注DESTROY事件,还关注 Activity/Fragment 的HIDE和SHOW,

并根据可见状态调整优先级。

调整优先级有什么用呢? 下面先看两张图感受一下。

为了凸显效果,我们把加载任务的并发量控制为1(串行)。

第一张是不会自动调整优先级的,完全的先进先出:

不改变优先级

可以看到,切换到第二个页面,由于上一页的任务还没执行完,

所以要一直等到上一页的任务都完成了才轮到第二个页面加载。

很显然这样体验不太好。

接下来我们看下动态调整优先级是什么效果:

动态调整优先级

切换到第二个页面之后,第一个页面的任务的“调度优先级”被降低了,所以会优先加载第二个页面的图片;

再次切换回第一个页面,第二个页面的优先级被降低,第一个页面的优先级恢复,所以优先加载第一个页面的图片。

那可否进入第二个页面的时暂停第一个页面的任务?

暂停的方案不太友好,比方说用户在第二个页面停留很久,第二个页面的任务都完成了,然后切换回第一个页面,发现只有部分图片(其他被暂停了)。

而如果只是调整优先级,则第二个页面的任务都执行完之后,会接着执行第一个页面的任务,返回第一个页面时就能够看到所有图片了。

这就好比赶车,让其他人给插个队,没有问题,但是不能不给别人排队了吧。

3.4 链式调用

UITask的用法和AsyncTask大同小异,回调方法和参数泛型都是一样的,所以就不多作介绍了。

如今很多开源库都提供了链式API,使用起来确实灵活方便,视觉上也比较连贯。

喜欢冰糖葫芦一样的链式调用?

我们提供了一个ChainTask类,拓展了UITask,提供链式调用的API。

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) {
// do something
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)
}

四、总结

所以,这个“加强版异步任务框架”,都做哪些方面的加强呢?

1、更灵活的并发控制

2、支持调度优先级

3、支持任务去重

4、支持生命周期(onDestroy时取消任务,自动调整优先级)

5、支持链式调用

最后,可能会这样的疑问:

既然已经有 RxJava 这样好用的开源库来实现异步了, 为什么还要写这个项目呢?

首先,RxJava 不仅仅是异步而已:“ReactiveX是一个通过使用可观察序列来编写异步和基于事件的程序的库。”

“可观察序列 - 事件 - 异步”加起来才使得 RxJava 如此富有魅力。

有所得,必有所付出,为了实现这些丰富的特性,代码量也是比较可观的(当前版本jar包约2.2M)。

AsyncTask则比较简单,除去注释只有三百多行代码;

功能也比较纯粹:执行异步任务,在任务执行的不同阶段,回调相应的方法。

Task参考了AsyncTask,功能类似,只是做了一些完善;

jar包大小45K,也算是比较轻量的。

这个年头,apk动辄几十M甚至上百M,2.2M的库并非不可接受。

但是也有一些场景,比方说给第三方写SDK的时候,对包大小和依赖比较敏感,而且也不需要这么大而全的特性,这时一些轻量级的方案就比较合适了。

而且,除了包大小之外,Task所实现的功能和RxJava也不尽相同。

如果说AsyncTask是自行车,RxJava是汽车,则Task是摩托车。

各有各的用途,各有各的灵魂。

五、下载

项目已经上传到maven和github, 欢迎大家下载 & star

dependencies {
implementation 'com.horizon.task:task:1.0.4'
}

项目地址:

https://github.com/No89757/Task

 
   
100 次浏览  评价: 差  订阅 捐助
相关文章

深度解析:清理烂代码
如何编写出拥抱变化的代码
重构-使代码更简洁优美
团队项目开发"编码规范"系列文章
相关文档

重构-改善既有代码的设计
软件重构v2
代码整洁之道
高质量编程规范
相关课程

基于HTML5客户端、Web端的应用开发
HTML 5+CSS 开发
嵌入式C高质量编程
C++高级编程
每天2个文档/视频
扫描微信二维码订阅
订阅技术月刊
获得每月300个技术资源
 
 

关于我们 | 联系我们 | 京ICP备10020922号 京公海网安备110108001071号