android中的协程

参考

导入依赖

除了要导入kotlin协程依赖外，还需要导入Android主线程协程库：

implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.9'

Retrofit的支持

https://github.com/JakeWharton/retrofit2-kotlin-coroutines-adapter

~~implementation 'com.jakewharton.retrofit:retrofit2-kotlin-coroutines-adapter:0.9.2'~~

此项目已经被deprecated了，因为在Retrofit 2.6.0+已经支持kotlin的suspend了。

MainScope

public fun MainScope(): CoroutineScope = ContextScope(SupervisorJob() + Dispatchers.Main)

通过MainScope()方法可以创建一个主线程的CoroutineScope，而主线程的dispatcher在Android中是用handler实现的，那么就可以使用协程的相关方法了，比如launch等。

val mainScope = MainScope()
override fun onCreate(savedInstanceState: Bundle?) {
    super.onCreate(savedInstanceState)
    ...
    launchButton.setOnClickListener {
        mainScope.launch {
            log(1)
            textView.text = async(Dispatchers.IO) {
                log(2)
                delay(1000)
                log(3)
                "Hello1111"
            }.await()
            log(4)
        }
    }
}


override fun onDestroy() {
    super.onDestroy()
    mainScope.cancel()
}

记得在onDestory中对这个mainScope进行cancel，防止内存泄露。

通过委托来实现

让activity实现CoroutineScope 接口，再by MainScope()

abstract class ScopedActivity: Activity(), CoroutineScope by MainScope(){
    override fun onDestroy() {
        super.onDestroy()
        cancel()
    }
}

将 Kotlin 协程与Architecture组件一起使用

https://developer.android.google.cn/topic/libraries/architecture/coroutines

除了MainScope()外，Android官方还提供了几个Architecture组件的协程/suspend的支持。

lifecycleScope

lifecycle可能是最常用到的一个架构组件，android官方提供了个kotlin协程的支持库

    dependencies {
        implementation "androidx.lifecycle:lifecycle-runtime-ktx:2.2.0"
    }

然后我们就可以直接在activity/fragment中直接通过lifecycleScope来在合适的生命周期启动协程，并在activity/fragment销毁时也自动取消lifecycleScope中启动的协程。

lifecycleScope是LifecycleCoroutineScope的对象，在LifecycleCoroutineScope定义了几个便捷的方法，让我们在合适的生命周期时机运行代码：

abstract class LifecycleCoroutineScope internal constructor() : CoroutineScope {
    internal abstract val lifecycle: Lifecycle

    /**
     * Launches and runs the given block when the [Lifecycle] controlling this
     * [LifecycleCoroutineScope] is at least in [Lifecycle.State.CREATED] state.
     *
     * The returned [Job] will be cancelled when the [Lifecycle] is destroyed.
     * @see Lifecycle.whenCreated
     * @see Lifecycle.coroutineScope
     */
    fun launchWhenCreated(block: suspend CoroutineScope.() -> Unit): Job = launch {
        lifecycle.whenCreated(block)
    }

    /**
     * Launches and runs the given block when the [Lifecycle] controlling this
     * [LifecycleCoroutineScope] is at least in [Lifecycle.State.STARTED] state.
     *
     * The returned [Job] will be cancelled when the [Lifecycle] is destroyed.
     * @see Lifecycle.whenStarted
     * @see Lifecycle.coroutineScope
     */

    fun launchWhenStarted(block: suspend CoroutineScope.() -> Unit): Job = launch {
        lifecycle.whenStarted(block)
    }

    /**
     * Launches and runs the given block when the [Lifecycle] controlling this
     * [LifecycleCoroutineScope] is at least in [Lifecycle.State.RESUMED] state.
     *
     * The returned [Job] will be cancelled when the [Lifecycle] is destroyed.
     * @see Lifecycle.whenResumed
     * @see Lifecycle.coroutineScope
     */
    fun launchWhenResumed(block: suspend CoroutineScope.() -> Unit): Job = launch {
        lifecycle.whenResumed(block)
    }
}

来跟一下源码

val LifecycleOwner.lifecycleScope: LifecycleCoroutineScope
    get() = lifecycle.coroutineScope

val Lifecycle.coroutineScope: LifecycleCoroutineScope
    get() {
        while (true) {
            val existing = mInternalScopeRef.get() as LifecycleCoroutineScopeImpl?
            if (existing != null) {
                return existing
            }
            val newScope = LifecycleCoroutineScopeImpl(
                this,
                SupervisorJob() + Dispatchers.Main.immediate
            )
            if (mInternalScopeRef.compareAndSet(null, newScope)) {
                newScope.register()
                return newScope
            }
        }
    }

当创建完newScope 后就会调用其register()进行添加生命周期的观察者。

internal class LifecycleCoroutineScopeImpl(
    override val lifecycle: Lifecycle,
    override val coroutineContext: CoroutineContext
) : LifecycleCoroutineScope(), LifecycleEventObserver {
    init {
        // in case we are initialized on a non-main thread, make a best effort check before
        // we return the scope. This is not sync but if developer is launching on a non-main
        // dispatcher, they cannot be 100% sure anyways.
        if (lifecycle.currentState == Lifecycle.State.DESTROYED) {
            coroutineContext.cancel()
        }
    }

    fun register() {
        launch(Dispatchers.Main.immediate) {
            if (lifecycle.currentState >= Lifecycle.State.INITIALIZED) {
                lifecycle.addObserver(this@LifecycleCoroutineScopeImpl)
            } else {
                coroutineContext.cancel()
            }
        }
    }

    override fun onStateChanged(source: LifecycleOwner, event: Lifecycle.Event) {
        if (lifecycle.currentState <= Lifecycle.State.DESTROYED) {
            lifecycle.removeObserver(this)
            coroutineContext.cancel()
        }
    }
}

当生命周期发生变化时会判断当前状态是否已经DESTROYED，是的话就把协程job取消。

扩展函数

除了上边的方法外，还有几个扩展函数能达到上边方法的效果。

suspend fun <T> LifecycleOwner.whenCreated(block: suspend CoroutineScope.() -> T): T =
    lifecycle.whenCreated(block)

/**
 * Runs the given block when the [Lifecycle] is at least in [Lifecycle.State.CREATED] state.
 *
 * @see Lifecycle.whenStateAtLeast for details
 */
suspend fun <T> Lifecycle.whenCreated(block: suspend CoroutineScope.() -> T): T {
    return whenStateAtLeast(Lifecycle.State.CREATED, block)
}

/**
 * Runs the given block when the [LifecycleOwner]'s [Lifecycle] is at least in
 * [Lifecycle.State.STARTED] state.
 *
 * @see Lifecycle.whenStateAtLeast for details
 */
suspend fun <T> LifecycleOwner.whenStarted(block: suspend CoroutineScope.() -> T): T =
    lifecycle.whenStarted(block)

/**
 * Runs the given block when the [Lifecycle] is at least in [Lifecycle.State.STARTED] state.
 *
 * @see Lifecycle.whenStateAtLeast for details
 */
suspend fun <T> Lifecycle.whenStarted(block: suspend CoroutineScope.() -> T): T {
    return whenStateAtLeast(Lifecycle.State.STARTED, block)
}

/**
 * Runs the given block when the [LifecycleOwner]'s [Lifecycle] is at least in
 * [Lifecycle.State.RESUMED] state.
 *
 * @see Lifecycle.whenStateAtLeast for details
 */
suspend fun <T> LifecycleOwner.whenResumed(block: suspend CoroutineScope.() -> T): T =
    lifecycle.whenResumed(block)

/**
 * Runs the given block when the [Lifecycle] is at least in [Lifecycle.State.RESUMED] state.
 *
 * @see Lifecycle.whenStateAtLeast for details
 */
suspend fun <T> Lifecycle.whenResumed(block: suspend CoroutineScope.() -> T): T {
    return whenStateAtLeast(Lifecycle.State.RESUMED, block)
}

上边的方法都调用到了一个方法whenStateAtLeast，看下这个方法的实现：

suspend fun <T> Lifecycle.whenStateAtLeast(
    minState: Lifecycle.State,
    block: suspend CoroutineScope.() -> T
) = withContext(Dispatchers.Main.immediate) {
    val job = coroutineContext[Job] ?: error("when[State] methods should have a parent job")
    val dispatcher = PausingDispatcher()
    val controller =
        LifecycleController(this@whenStateAtLeast, minState, dispatcher.dispatchQueue, job)
    try {
        withContext(dispatcher, block)
    } finally {
        controller.finish()
    }
}

internal class LifecycleController(
    private val lifecycle: Lifecycle,
    private val minState: Lifecycle.State,
    private val dispatchQueue: DispatchQueue,
    parentJob: Job
) {
    private val observer = LifecycleEventObserver { source, _ ->
        if (source.lifecycle.currentState == Lifecycle.State.DESTROYED) {
            // cancel job before resuming remaining coroutines so that they run in cancelled
            // state
            handleDestroy(parentJob)
        } else if (source.lifecycle.currentState < minState) {
            dispatchQueue.pause()
        } else {
            dispatchQueue.resume()
        }
    }

    init {
        // If Lifecycle is already destroyed (e.g. developer leaked the lifecycle), we won't get
        // an event callback so we need to check for it before registering
        // see: b/128749497 for details.
        if (lifecycle.currentState == Lifecycle.State.DESTROYED) {
            handleDestroy(parentJob)
        } else {
            lifecycle.addObserver(observer)
        }
    }

    @Suppress("NOTHING_TO_INLINE") // avoid unnecessary method
    private inline fun handleDestroy(parentJob: Job) {
        parentJob.cancel()
        finish()
    }

    /**
     * Removes the observer and also marks the [DispatchQueue] as finished so that any remaining
     * runnables can be executed.
     */
    @MainThread
    fun finish() {
        lifecycle.removeObserver(observer)
        dispatchQueue.finish()
    }
}

可以看到当执行一次完成后，就直接结束。也就是说当在onResume时执行一段代码，然后onPause再onResume时是不会再执行一次的。

viewModelScope

在viewModel和Activity/fragment的生命周期是不一致的，所以是不能直接使用lifecycleScope的，其实是可以使用MainScope的，只不过要进行手动管理取消。

所以Android官方又贴心的为我们开发了一个viewModel的协程支持库。

    dependencies {
        implementation "androidx.lifecycle:lifecycle-viewmodel-ktx:2.2.0"
    }

然后在viewModel中就可以通过viewModelScope调用协程相关方法来启动协程，比如launch。

例如，以下 viewModelScope会启动一个协程，用于在后台线程中发出网络请求。该库会处理所有设置和相应的范围清除：

class MainViewModel : ViewModel() {
    // Make a network request without blocking the UI thread
    private fun makeNetworkRequest() {
        // launch a coroutine in viewModelScope
        viewModelScope.launch  {
            remoteApi.slowFetch()
            ...
        }
    }

    // No need to override onCleared()
}

源码

val ViewModel.viewModelScope: CoroutineScope
        get() {
            val scope: CoroutineScope? = this.getTag(JOB_KEY)
            if (scope != null) {
                return scope
            }
            return setTagIfAbsent(JOB_KEY,
                CloseableCoroutineScope(SupervisorJob() + Dispatchers.Main.immediate))
        }

internal class CloseableCoroutineScope(context: CoroutineContext) : Closeable, CoroutineScope {
    override val coroutineContext: CoroutineContext = context

    override fun close() {
        coroutineContext.cancel()
    }
}

官方说viewModelScope会自动的取消viewModelScope中的协程，那么官方是怎么实现的呢？

还得回到viewModel中：

private final Map<String, Object> mBagOfTags = new HashMap<>();
private volatile boolean mCleared = false;

<T> T setTagIfAbsent(String key, T newValue) {
    T previous;
    synchronized (mBagOfTags) {
        previous = (T) mBagOfTags.get(key);
        if (previous == null) {
            mBagOfTags.put(key, newValue);
        }
    }
    T result = previous == null ? newValue : previous;
    if (mCleared) {
        // It is possible that we'll call close() multiple times on the same object, but
        // Closeable interface requires close method to be idempotent:
        // "if the stream is already closed then invoking this method has no effect." (c)
        closeWithRuntimeException(result);
    }
    return result;
}

@MainThread
final void clear() {
    mCleared = true;
    // Since clear() is final, this method is still called on mock objects
    // and in those cases, mBagOfTags is null. It'll always be empty though
    // because setTagIfAbsent and getTag are not final so we can skip
    // clearing it
    if (mBagOfTags != null) {
        synchronized (mBagOfTags) {
            for (Object value : mBagOfTags.values()) {
                // see comment for the similar call in setTagIfAbsent
                closeWithRuntimeException(value);
            }
        }
    }
    onCleared();
}

protected void onCleared() {
}

private static void closeWithRuntimeException(Object obj) {
    if (obj instanceof Closeable) {
        try {
            ((Closeable) obj).close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

可以看到当viewmodel被clear时，会先把mBagOfTags 这个map中的值一个个的取出来去close(如果值是Closeable的话)，

而在viewModelScope的定义处会创建一个CloseableCoroutineScope加入到mBagOfTags 中，而CloseableCoroutineScope是实现了Closeable接口的，

所以当viewmodel被clear时，viewModelScope启动的协程会自动cancel。

将协程与 LiveData 一起使用

这个不常使用。

使用 LiveData 时，您可能需要异步计算值。例如，您可能需要检索用户的偏好设置并将其传送给界面。在这些情况下，LiveData KTX 可提供一个 liveData 构建器函数，该函数会调用 suspend 函数，并将结果作为 LiveData 对象传送。

要使用此模块，请将以下内容添加到应用的 build.gradle 文件中：

    dependencies {
        implementation "androidx.lifecycle:lifecycle-livedata-ktx:2.2.0"
    }

然后就可以调用liveData函数来异步计算值。

当 LiveData 变为非活动状态onInactive()时，代码块会在可配置的超时过后自动取消。如果代码块在完成前取消，则会在 LiveData 再次变为活动状态onActive()后重启；如果在上次运行中成功完成，则不会重启。请注意，代码块只有在自动取消的情况下才会重启。如果代码块由于任何其他原因（例如，抛出 CancelationException）而取消，则不会重启。

在以下示例中，loadUser() 是在其他地方声明的 suspend 函数。您可以使用 liveData 构建器函数异步调用 loadUser()，然后使用 emit() 来发出结果：

val user: LiveData<User> = liveData {
    val data = database.loadUser() // loadUser is a suspend function.
    emit(data)
}

源码

fun <T> liveData(
    context: CoroutineContext = EmptyCoroutineContext,
    timeoutInMs: Long = DEFAULT_TIMEOUT,
    @BuilderInference block: suspend LiveDataScope<T>.() -> Unit
): LiveData<T> = CoroutineLiveData(context, timeoutInMs, block)

其中的lambda会有一个接收者是LiveDataScope，

同时可以指定一个timeoutInMs，超过指定时间后

interface LiveDataScope<T> {
    /**
     * Set's the [LiveData]'s value to the given [value]. If you've called [emitSource] previously,
     * calling [emit] will remove that source.
     *
     * Note that this function suspends until the value is set on the [LiveData].
     *
     * @param value The new value for the [LiveData]
     *
     * @see emitSource
     */
    suspend fun emit(value: T)

    /**
     * Add the given [LiveData] as a source, similar to [MediatorLiveData.addSource]. Calling this
     * method will remove any source that was yielded before via [emitSource].
     *
     * @param source The [LiveData] instance whose values will be dispatched from the current
     * [LiveData].
     *
     * @see emit
     * @see MediatorLiveData.addSource
     * @see MediatorLiveData.removeSource
     */
    suspend fun emitSource(source: LiveData<T>): DisposableHandle

    /**
     * References the current value of the [LiveData].
     *
     * If the block never `emit`ed a value, [latestValue] will be `null`. You can use this
     * value to check what was then latest value `emit`ed by your `block` before it got cancelled.
     *
     * Note that if the block called [emitSource], then `latestValue` will be last value
     * dispatched by the `source` [LiveData].
     */
    val latestValue: T?
}