Kotlin's collaboration: Context

Kotlin The context of the collaboration is called CoroutineContext, Usually used to switch thread pools .

CoroutineContext Application

launch

launch The first parameter of context yes CoroutineContext, The default value is EmptyCoroutineContext.

If you need to specify launch Working thread pool , You need to specify CoroutineContext Parameters .

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job {
    
    val newContext = newCoroutineContext(context)
    val coroutine = if (start.isLazy)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)
    coroutine.start(start, coroutine, block)
    return coroutine
}

withContext

withContext Used to switch thread execution code . Its first parameter is CoroutineContext, Specify thread pool .

public suspend fun <T> withContext(
    context: CoroutineContext,
    block: suspend CoroutineScope.() -> T
): T {
    
}

stay getUserInfoIo It is specified in withContext The context is Dispatchers.IO.

fun main() = runBlocking {
    
    val user = getUserInfoIo()
    logX(user)
}

suspend fun getUserInfoIo(): String {
    
    logX("Before IO Context.")
    withContext(Dispatchers.IO) {
    
        logX("In IO Context.")
        delay(1000)
    }
    logX("After IO Context.")
    return "BoyCoder"
}

Output is as follows ：

================================
Before IO Context.
Thread:main @coroutine#1, time:1656560405319
================================
================================
In IO Context.
Thread:DefaultDispatcher-worker-1 @coroutine#1, time:1656560405351
================================
================================
After IO Context.
Thread:main @coroutine#1, time:1656560406360
================================
================================
BoyCoder
Thread:main @coroutine#1, time:1656560406361
================================

You can see in the Thread:DefaultDispatcher-worker-1 Thread execution coroutine 1. Other codes in main Threads .

suspend main

main Function has suspend The version of the keyword , You can directly execute the suspend function .

suspend fun main() {
    
    val user = getUserInfoIo()
    logX(user)
}

It and runBlocking The difference is that withContext After switching threads , All implemented in DefaultDispatcher-worker-1 Threads .

================================
Before IO Context.
Thread:main, time:1656569681374
================================
================================
In IO Context.
Thread:DefaultDispatcher-worker-1, time:1656569681440
================================
================================
After IO Context.
Thread:DefaultDispatcher-worker-1, time:1656569682449
================================
================================
BoyCoder
Thread:DefaultDispatcher-worker-1, time:1656569682449
================================

runBlocking

runBlocking The first parameter of CoroutineContext, The default is EmptyCoroutineContext.

public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
    
}

You can give runBlocking Add custom CoroutineContext.

fun main() = runBlocking(Dispatchers.IO) {
    
    val user = getUserInfoIo()
    logX(user)
}

Output is as follows ：

================================
Before IO Context.
Thread:DefaultDispatcher-worker-2 @coroutine#1, time:1656570311862
================================
================================
In IO Context.
Thread:DefaultDispatcher-worker-2 @coroutine#1, time:1656570311896
================================
================================
After IO Context.
Thread:DefaultDispatcher-worker-2 @coroutine#1, time:1656570312903
================================
================================
BoyCoder
Thread:DefaultDispatcher-worker-2 @coroutine#1, time:1656570312903
================================

You can see the increase Dispatchers.IO after , The coordination process has been implemented in DefaultDispatcher-worker-2 Threads .

Dispatchers Scheduler

Kotlin Built in Dispatchers Yes 4 Kind of , They all inherit essentially CoroutineContext.

• Default： Default scheduler , be used for CPU Intensive task , The number of threads is consistent with the number of cores .
• Main： Main thread scheduler , Only in Android、Swing etc. UI The platform is useful , Ordinary JVM The project cannot be used .
• UnConfined： Unlimited scheduler , A coroutine may run on any thread .
• IO：IO Scheduler , be used for IO Intensive task , There will be more threads , such as 64 Threads

public actual object Dispatchers {
    
    /** * The default [CoroutineDispatcher] that is used by all standard builders like * [launch][CoroutineScope.launch], [async][CoroutineScope.async], etc * if no dispatcher nor any other [ContinuationInterceptor] is specified in their context. * * It is backed by a shared pool of threads on JVM. By default, the maximal level of parallelism used * by this dispatcher is equal to the number of CPU cores, but is at least two. * Level of parallelism X guarantees that no more than X tasks can be executed in this dispatcher in parallel. */
    @JvmStatic
    public actual val Default: CoroutineDispatcher = createDefaultDispatcher()

    /** * A coroutine dispatcher that is confined to the Main thread operating with UI objects. * This dispatcher can be used either directly or via [MainScope] factory. * Usually such dispatcher is single-threaded. * * Access to this property may throw [IllegalStateException] if no main thread dispatchers are present in the classpath. * * Depending on platform and classpath it can be mapped to different dispatchers: * - On JS and Native it is equivalent of [Default] dispatcher. * - On JVM it is either Android main thread dispatcher, JavaFx or Swing EDT dispatcher. It is chosen by * [`ServiceLoader`](https://docs.oracle.com/javase/8/docs/api/java/util/ServiceLoader.html). * * In order to work with `Main` dispatcher, the following artifacts should be added to project runtime dependencies: * - `kotlinx-coroutines-android` for Android Main thread dispatcher * - `kotlinx-coroutines-javafx` for JavaFx Application thread dispatcher * - `kotlinx-coroutines-swing` for Swing EDT dispatcher * * In order to set a custom `Main` dispatcher for testing purposes, add the `kotlinx-coroutines-test` artifact to * project test dependencies. * * Implementation note: [MainCoroutineDispatcher.immediate] is not supported on Native and JS platforms. */
    @JvmStatic
    public actual val Main: MainCoroutineDispatcher get() = MainDispatcherLoader.dispatcher

    /** * A coroutine dispatcher that is not confined to any specific thread. * It executes initial continuation of the coroutine in the current call-frame * and lets the coroutine resume in whatever thread that is used by the corresponding suspending function, without * mandating any specific threading policy. Nested coroutines launched in this dispatcher form an event-loop to avoid * stack overflows. * * ### Event loop * Event loop semantics is a purely internal concept and have no guarantees on the order of execution * except that all queued coroutines will be executed on the current thread in the lexical scope of the outermost * unconfined coroutine. * * For example, the following code: * ``` * withContext(Dispatchers.Unconfined) { * println(1) * withContext(Dispatchers.Unconfined) { // Nested unconfined * println(2) * } * println(3) * } * println("Done") * ``` * Can print both "1 2 3" and "1 3 2", this is an implementation detail that can be changed. * But it is guaranteed that "Done" will be printed only when both `withContext` are completed. * * * Note that if you need your coroutine to be confined to a particular thread or a thread-pool after resumption, * but still want to execute it in the current call-frame until its first suspension, then you can use * an optional [CoroutineStart] parameter in coroutine builders like * [launch][CoroutineScope.launch] and [async][CoroutineScope.async] setting it to the * the value of [CoroutineStart.UNDISPATCHED]. */
    @JvmStatic
    public actual val Unconfined: CoroutineDispatcher = kotlinx.coroutines.Unconfined

    /** * The [CoroutineDispatcher] that is designed for offloading blocking IO tasks to a shared pool of threads. * * Additional threads in this pool are created and are shutdown on demand. * The number of threads used by this dispatcher is limited by the value of * "`kotlinx.coroutines.io.parallelism`" ([IO_PARALLELISM_PROPERTY_NAME]) system property. * It defaults to the limit of 64 threads or the number of cores (whichever is larger). * * Moreover, the maximum configurable number of threads is capped by the * `kotlinx.coroutines.scheduler.max.pool.size` system property. * If you need a higher number of parallel threads, * you should use a custom dispatcher backed by your own thread pool. * * This dispatcher shares threads with a [Default][Dispatchers.Default] dispatcher, so using * `withContext(Dispatchers.IO) { ... }` does not lead to an actual switching to another thread — * typically execution continues in the same thread. */
    @JvmStatic
    public val IO: CoroutineDispatcher = DefaultScheduler.IO
}

Dispatchers.IO

Dispatchers.IO May reuse Dispatchers.Default The thread of . As can be seen from the above example , Although the setting is Dispatchers.IO, the truth is that DefaultDispatcher-worker Threads .

take runBlocking Of CoroutineContext Change it to Dispatchers.Default

fun main() = runBlocking(Dispatchers.Default) {
    
    val user = getUserInfoIo()
    logX(user)
}

The output is as follows ：

================================
Before IO Context.
Thread:DefaultDispatcher-worker-1 @coroutine#1, time:1656575072324
================================
================================
In IO Context.
Thread:DefaultDispatcher-worker-1 @coroutine#1, time:1656575072358
================================
================================
After IO Context.
Thread:DefaultDispatcher-worker-1 @coroutine#1, time:1656575073366
================================
================================
BoyCoder
Thread:DefaultDispatcher-worker-1 @coroutine#1, time:1656575073367
================================

You can see withContext Switch to IO after , Also used. DefaultDispatcher-worker-1. This is because Dispatchers.Default By Dispatchers.IO Reuse threads .

Customize Dispatchers

Use custom executor And then convert to Dispatcher As CoroutineContext.

val mySingleDispatcher = Executors.newSingleThreadExecutor {
    
    Thread(it, "mySingleThread").apply {
    
        isDaemon = true
    }
}
    .asCoroutineDispatcher()

fun main() = runBlocking(mySingleDispatcher) {
    
    val user = getUserInfoIo()
    logX(user)
}

Output is as follows ：

================================
Before IO Context.
Thread:mySingleThread @coroutine#1, time:1656575809476
================================
================================
In IO Context.
Thread:DefaultDispatcher-worker-1 @coroutine#1, time:1656575809510
================================
================================
After IO Context.
Thread:mySingleThread @coroutine#1, time:1656575810521
================================
================================
BoyCoder
Thread:mySingleThread @coroutine#1, time:1656575810521
================================

Only In IO Context Running on the DefaultDispatcher-worker-1, Other code runs in custom dispatcher.

Dispatchers.Unconfined

If launch Use the default context, The order of execution is 1、4、2、3.

fun main() = runBlocking {
    
    logX("Before launch") // 1
    launch {
    
        logX("In launch") // 2
        delay(1000)
        logX("End launch") // 3
    }
    logX("After launch") // 4
}

Output is as follows

================================
Before launch
Thread:main @coroutine#1, time:1656576229645
================================
================================
After launch
Thread:main @coroutine#1, time:1656576229677
================================
================================
In launch
Thread:main @coroutine#2, time:1656576229679
================================
================================
End launch
Thread:main @coroutine#2, time:1656576230686
================================

If launch Use Unconfined, The order of execution is uncertain .

fun main() = runBlocking {
    
    logX("Before launch") // 1
    launch(Dispatchers.Unconfined) {
    
        logX("In launch") // 2
        delay(1000)
        logX("End launch") // 3
    }
    logX("After launch") // 4
}

Output is as follows

================================
Before launch
Thread:main @coroutine#1, time:1656576759031
================================
================================
In launch
Thread:main @coroutine#2, time:1656576759055
================================
================================
After launch
Thread:main @coroutine#1, time:1656576759060
================================
================================
End launch
Thread:kotlinx.coroutines.DefaultExecutor @coroutine#2, time:1656576760059
================================

The order of execution changes to 1、2、4、3. And the mark 3 Execute on DefaultExecutor.

Dispatchers.Unconfined May execute on any thread , It should not be used casually Dispatchers.Unconfined.

CoroutineScope Process scope

Use launch when , There has to be CoroutineScope Process scope ,launch yes CoroutineScope The extension function of .

CoroutineScope Is very simple to implement , It is an interface class , Only coroutineContext member .

public interface CoroutineScope {
    
    /** * The context of this scope. * Context is encapsulated by the scope and used for implementation of coroutine builders that are extensions on the scope. * Accessing this property in general code is not recommended for any purposes except accessing the [Job] instance for advanced usages. * * By convention, should contain an instance of a [job][Job] to enforce structured concurrency. */
    public val coroutineContext: CoroutineContext
}

Specify custom CoroutineScope, Use scope start-up 3 Collaborators cheng , Achieve structured concurrency .

fun main() = runBlocking {
    
    val scope = CoroutineScope(Job())

    scope.launch {
    
        logX("first start")
        delay(1000)
        logX("first end")
    }

    scope.launch {
    
        logX("second start")
        delay(1000)
        logX("second end")
    }

    scope.launch {
    
        logX("third start")
        delay(1000)
        logX("third end")
    }

    delay(500)
    scope.cancel()
    delay(1000)
}

Output is as follows

================================
first start
Thread:DefaultDispatcher-worker-2 @coroutine#2, time:1656577491658
================================
================================
third start
Thread:DefaultDispatcher-worker-3 @coroutine#4, time:1656577491663
================================
================================
second start
Thread:DefaultDispatcher-worker-1 @coroutine#3, time:1656577491660
================================

all scope All started processes are canceled , No further execution .

Job and Dispatcher

Job

Job Inherited CoroutineContext.Element.

public interface Job : CoroutineContext.Element {
    
}

Element Inherit CoroutineContext

@kotlin.SinceKotlin public interface CoroutineContext {
    
    public interface Element : kotlin.coroutines.CoroutineContext {
    
    }
}

therefore Job Itself is CoroutineContext.

Dispatchers

Dispatchers It's a singleton class , There are 4 Presets Dispatchers.

public actual object Dispatchers {
    
}

With Default For example , It is CoroutineDispatcher class , The inheritance relationship is as follows ：
CoroutineDispatcher -> ContinuationInterceptor -> CoroutineContext.Element -> CoroutineContext

therefore Dispatchers It's also CoroutineContext.

CoroutineContext The design of the

CoroutineContext Design method and Map The design method of is very similar .

public interface CoroutineContext {
    
    public operator fun <E : Element> get(key: Key<E>): E?

    public fun <R> fold(initial: R, operation: (R, Element) -> R): R

    public operator fun plus(context: CoroutineContext): CoroutineContext =

    public fun minusKey(key: Key<*>): CoroutineContext

• get amount to map Of get
• fold amount to map Of foreach
• plus amount to map Of put
• minusKey amount to map Of remove

utilize CoroutineContext The interface of , You can write + 、[] This set like operation . because CoroutineContext Reload the plus The operator , It can be used + Instead of plus; heavy load get The operator , It can be used [] Instead of get.

@OptIn(ExperimentalStdlibApi::class)
fun main() = runBlocking {
    
    val scope = CoroutineScope(Job() + mySingleDispatcher)

    scope.launch {
    
        logX(coroutineContext[CoroutineDispatcher] == mySingleDispatcher)
        delay(500)
        logX("first end")
    }

    delay(500)
    scope.cancel()
    delay(1000)
}

Output is as follows

================================
true
Thread:mySingleThread @coroutine#2, time:1656584502696
================================

It can be seen that CoroutineContext Inside dispatchers Namely mySingleDispatcher.

CoroutineName The name of the project

CoroutineName You can specify the process name , Its essence is also CoroutineContext.

@OptIn(ExperimentalStdlibApi::class)
fun main() = runBlocking {
    
    val scope = CoroutineScope(Job() + mySingleDispatcher)

    scope.launch(CoroutineName("MyFirstCoroutine")) {
    
        logX(coroutineContext[CoroutineDispatcher] == mySingleDispatcher)
        delay(500)
        logX("first end")
    }

    delay(500)
    scope.cancel()
    delay(1000)
}

Output is as follows

================================
true
Thread:mySingleThread @MyFirstCoroutine#2, time:1656584886943
================================

The name of the collaboration process changes to CoroutineName Defined MyFirstCoroutine,#2 It's a journey id.

CoroutineExceptionHandler Exception handling of coroutine

CoroutineExceptionHandler Used to catch exceptions in the process , So is it CoroutineContext.

@OptIn(ExperimentalStdlibApi::class)
suspend fun main() {
    
    val scope = CoroutineScope(Job() + mySingleDispatcher)
    val handler = CoroutineExceptionHandler {
     _, throwable ->
        println("catch exception $throwable")
    }
    val job = scope.launch(handler) {
    
        logX(coroutineContext[CoroutineDispatcher] == mySingleDispatcher)
        val str: String? = null
        str!!.length
        logX("first end")
    }
    job.join()
}

Output is as follows

================================
true
Thread:mySingleThread @coroutine#1, time:1656585762125
================================
catch exception java.lang.NullPointerException

Deliberately throw a null pointer exception , And then to CoroutineExceptionHandler Capture .

Suspend functions and CoroutineContext

suspend The suspend function can directly access coroutineContext.

suspend fun testCoroutineContext() = coroutineContext

fun main() = runBlocking {
    
    print(testCoroutineContext())
}

Output is as follows

[CoroutineId(1), "coroutine#1":BlockingCoroutine{
    Active}@63e2203c, [email protected]1efed156]

The suspended function needs to be run in a coroutine or another suspended function , Therefore, it can also access the cooperative process CoroutineContext.

coroutineContext yes Continuation.kt Internal members of .

public suspend inline val coroutineContext: CoroutineContext
    get() {
    
        throw NotImplementedError("Implemented as intrinsic")
    }

summary

• CoroutineContext Process context and map Is very similar , and map equally , It can be added or obtained Element.
• Job、Dispatchers、CoroutineName、CoroutineExceptionHandler Nature is CoroutineContext.
• CoroutineScope Encapsulates the CoroutineContext,CoroutineContext yes CoroutineScope Members of .
• suspend The suspend function is also similar to CoroutineContext of . The pending function should be run in the coroutine , Xiecheng has its CoroutineContext, Therefore, the suspended function can also access CoroutineContext.