异步、非阻塞式 Android 启动任务调度库

1、背景

节前面试的时候被问到 Android 启动任务依赖怎么做调度。当时随口给了一个方案，后来想想觉得有意思就自己花了一天的时间写了一个。这个库已经开源到 Github 上面：

https://github.com/Shouheng88/AndroidStartup

在写这个库之前只是看了下 Jetpack 的 Startup. 毕竟，如果这个库已经非常完善了，那么我就没必要自己再搞一个了。截止目前，在我看来，这个库最大的缺点是，这个库所有的任务都在主线程中触发并执行，而我们为了优化启动的性能通常会将任务放到异步线程中执行。所以，Jetpack 的库充其量只能解决你的任务的依赖关系。

如果要支持异步任务执行，首先要解决的是如何保证任务的先后顺序。最初我也想到了使用并发包里的闭锁的方案，但是这种方案有个问题。即，闭锁执行的时候使用 CAS 以阻塞的方式进行等待，这样会白白浪费线程资源。如果因此占用了 CPU，将会影响到我们其他线程的执行。所以，在我的库中，我使用了非阻塞的事件通知机制。这样在某个任务结束之后会通知所有依赖于它的任务。当一个任务的所有依赖都执行完毕，再执行自己的任务。当然，对于不同参数的线程池，异步任务执行的表现也是不一样的，所以，我也提供了方法用来自定义线程池。

此外，我在开发这个库的时候还用到了注解处理器。你可以通过注解声明自己的任务，然后编译期间会自动发现并拼接你的任务。相对于其他的框架，又多了一个初始化的选择。

2、结构

在开发的时候我将任务调度和启动工具分成了两个独立的模块，这样任务调度工具也可以单独拿出来使用。后来，增加了注解驱动相关的逻辑，就又增加了两个模块。所以，现在各模块及功能如下：

scheduler:           任务调度工具
startup:             启动工具，任务调度工具包装
startup-annotation:  注解定义
startup-compiler:    注解编译器

3、调度器

先来看任务调度器的工作原理。

3.1 任务封装

首先是任务的定义。在我的项目中我使用 ISchedulerJob 定义任务。

interface ISchedulerJob {

    fun threadMode(): ThreadMode

    fun dependencies(): List<Class<out ISchedulerJob>>

    fun run(context: Context)
}

它定义了三个方法，分别是：

• threadMode() 用来指定执行任务的线程
• dependencies() 用来指定当前任务依赖的任务
• run() 你的初始化的任务执行的方法

其次，真实的任务分发的逻辑是通过 Dispatcher 来完成的。根据任务之间的依赖关系，我们可以构建成一拓扑结构。执行任务的时候先执行的任务就是这个拓扑结构的根结点。放在这里就是 dependencies() 方法为空的结点。所以，这里，我们首先要监测拓扑结构是否存在环。然后，只需要找到根结点并从根结点执行任务即可。

3.2 环检测

对于环监测，如果不考虑空间复杂度，我们可以使用 Set 来发现循环依赖：

private fun checkDependencies() {
    val checking = mutableSetOf<Class<out ISchedulerJob>>()
    val checked = mutableSetOf<Class<out ISchedulerJob>>()
    val schedulerMap = mutableMapOf<Class<ISchedulerJob>, ISchedulerJob>()
    schedulerJobs.forEach { schedulerMap[it.javaClass] = it }
    schedulerJobs.forEach { schedulerJob ->
        checkDependenciesReal(schedulerJob, schedulerMap, checking, checked)
    }
}

private fun checkDependenciesReal(
    schedulerJob: ISchedulerJob,
    map: Map<Class<ISchedulerJob>, ISchedulerJob>,
    checking: MutableSet<Class<out ISchedulerJob>>,
    checked: MutableSet<Class<out ISchedulerJob>>
) {
    if (checking.contains(schedulerJob.javaClass)) {
        // Cycle detected.
        throw SchedulerException("Cycle detected for ${schedulerJob.javaClass.name}.")
    }
    if (!checked.contains(schedulerJob.javaClass)) {
        checking.add(schedulerJob.javaClass)
        if (schedulerJob.dependencies().isNotEmpty()) {
            schedulerJob.dependencies().forEach {
                if (!checked.contains(it)) {
                    val job = map[it]
                        ?: throw SchedulerException(String.format("dependency [%s] not found", it.name))
                    checkDependenciesReal(job, map, checking, checked)
                }
            }
        }
        checking.remove(schedulerJob.javaClass)
        checked.add(schedulerJob.javaClass)
    }
}

这里的逻辑和 Jetpack 中的环监测的逻辑差不多。这里用了两个多余的数据结构，分别记录已经检测的和检测中的任务结点，如果发现了一个需要检测的任务正在检测中，则说明存在环。

3.3 任务启动

任务启动之前需要先根据任务间的依赖关系建立数据结构，简单说就是需要知道当前任务有哪些依赖任务和哪些依赖于它的任务。

private fun buildDispatcherJobs() {
    roots.clear()

    // Build the map from scheduler class type to dispatcher job.
    val map =  mutableMapOf<Class<ISchedulerJob>, DispatcherJob>()
    schedulerJobs.forEach {
        val dispatcherJob = DispatcherJob(this.globalContext, executor, it)
        map[it.javaClass] = dispatcherJob
    }

    // Fill the parent field for dispatcher job.
    schedulerJobs.forEach { schedulerJob ->
        val dispatcherJob = map[schedulerJob.javaClass]!!
        schedulerJob.dependencies().forEach {
            dispatcherJob.addParent(map[it]!!)
        }
    }

    // Fill the children field for dispatcher job.
    schedulerJobs.forEach { schedulerJob ->
        val dispatcherJob = map[schedulerJob.javaClass]!!
        dispatcherJob.parents().forEach {
            it.addChild(dispatcherJob)
        }
    }

    // Find roots.
    schedulerJobs.filter {
        it.dependencies().isEmpty()
    }.forEach {
        val dispatcherJob = map[it.javaClass]!!
        roots.add(dispatcherJob)
    }
}

这里先对任务做一个封装，将所有的任务包装成 DispatcherJob，然后根据任务的依赖关系找到各任务的父任务，并调用其 addParent() 方法，这里在 DispatcherJob 中会使用一个 AtomicInteger 进行计数，统计其父任务的数量。然后，再通过各任务的父任务维护子任务关系。最后，再根据任务的依赖找到拓扑的根结点。这样，我们就可以从根结点开始执行整个拓扑结构。

3.4 任务通知机制

上面我们提到了 DispatcherJob，启动一个 DispatcherJob 只需要调用它的 execute() 方法，该方法中会根据线程模型做判断，从而选择执行的线程执行任务：

override fun execute() {
    val realJob = {
        // Run the task.
        job.run(context)
        // Handle for children.
        children.forEach { it.notifyJobFinished(this) }
    }

    try {
        if (job.threadMode() == ThreadMode.MAIN) {
            // Cases for main thread.
            if (Thread.currentThread() == Looper.getMainLooper().thread) {
                realJob()
            } else {
                Handler(Looper.getMainLooper()).post { realJob() }
            }
        } else {
            // Cases for background thread.
            executor.execute { realJob() }
        }
    } catch (e: Throwable) {
        throw SchedulerException(e)
    }
}

这里任务的执行逻辑被包装到了一个 lambda 方法中。如果是主线程，可以根据当前线程状态执行执行或者 post 到主线程中执行。如果是异步任务，则将其丢到线程池当中执行。

当一个任务的工作结束之后会获取所有的子任务进行通知，这里用到了 notifyJobFinished() 方法。这个方法也很简单，就是没当一个任务执行完毕，则计数器减 1，当所有的依赖任务都执行完毕的时候，它才开始执行自己的任务，以此来通过事件而不是阻塞的方式进行任务调度：

override fun notifyJobFinished(job: IDispatcherJob) {
    if (waiting.decrementAndGet() == 0) {
        // All dependencies finished, commit the job.
        execute()
    }
}

4、启动器

对于启动器，你有三种选择。使用类似于 Jetpack 的 ContentProvider、自己声明任务或者使用注解 @StartupJob 进行任务声明。

对于内容提供器，原理比较简单，就是再自定义 ContentProvider 的 onCreate() 方法中扫描自定义的 meta-data. ContentProvider 的声明方式有几个问题，第一，ContentProvider 作为四大组件之一，建立过程需要消耗一定性能。此外，默认 ContentProvider 运行在主进程当中，所以，如果你的应用中如果用到了多进程，那么默认的 ContentProvider 不会为你的子进程做初始化，除非你明确指定它的进程。

所以，除了 ContentProvider，你还可以使用手动声明的方式，

AndroidStartup.newInstance(this).jobs(
   CrashHelperInitializeJob(),
   ThirdPartLibrariesInitializeJob(),
   DependentBlockingBackgroundJob(),
   BlockingBackgroundJob()
).launch()

此外，我还特意增加了注解的方式声明任务。使用起来很简单，你只需要在自己的任务上面使用注解声明即可，如：

@StartupJob
class BlockingBackgroundJob : ISchedulerJob {

    override fun threadMode(): ThreadMode = ThreadMode.BACKGROUND

    override fun dependencies(): List<Class<out ISchedulerJob>> = emptyList()

    override fun run(context: Context) {
        Thread.sleep(5_000L) // 5 seconds
        L.d("BlockingBackgroundJob done! ${Thread.currentThread()}")
    }
}

它的工作原理也比较简单，就是当你调用 AndroidStartup 的 scanAnnotations() 方法的时候，它会通过反射调用 JobHunter 的方法获取所有的任务。在编译期间，我们会为这个接口提供实现，并会对所有扫描到的任务进行初始化并在该实现中返回。

总结

以上就是这个库的实现原理。