android中的協程

參考

https://developer.android.com/kotlin/coroutines

https://www.bennyhuo.com/2019/05/27/coroutine-android/

https://juejin.im/post/6854573211418361864

導入依賴

除了要導入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?
}