Processes, threads, and coroutines

To put it simply: threads are the smallest unit of execution in an operating system, and processes are the smallest unit of resource management.

Thread belongs to process and is the actual executor of the program. A process guarantees at least one main thread and can package many sub-threads. Both processes and threads are managed by the operating system.

In a typical consumer/producer pattern, if multiple producers write data to a queue, and several consumers consume data from the queue, producers block if the queue is full, and consumers block if the queue is full

This process involves synchronization, thread state switching, thread context switching, and each point is a performance loss operation.

coroutines

This brings us to the concept of coroutines (not the Ctrip who bought the ticket).

A coroutine is a lightweight thread, like a process can have multiple threads, and a thread can have multiple coroutines.

So the question is, why use coroutines? What advantages does it have over threads?

Take a look at how coroutines are defined:

Roman Elizarov, the developer of coroutines, describes them this way: coroutines are like very lightweight threads. Threads are scheduled by the system, and the overhead of thread switching or thread blocking is high. Coroutines depend on threads, but do not block threads when they are suspended, and are virtually costless. The coroutines are controlled by the developer. So coroutines are also like user threads, very lightweight, you can create any coroutines in one thread.

Coroutines in Kotlin

Introduction in official documentation:

Coroutines simplify asynchronous programming by putting complexity into libraries. The program logic can be expressed sequentially in coroutines, and the underlying library takes care of the asynchrony for us. The library can wrap the relevant parts of user code as callbacks, subscribing to related events, and scheduling execution on different threads (or even different machines), while keeping the code as simple as sequential execution

Coroutines are executed sequentially in threads, so how do you implement asynchronous operations?

Thread has the concepts of blocking and awakening, and coroutines also have similar concepts. Suspending is equivalent to blocking. When a Thread receives a suspend request from a coroutine, it will perform other computational tasks, such as other coroutines. Coroutines are used in this way to achieve multi-threaded, asynchronous effects, and Thread comparison will have some differences

Rely on

Add the following dependencies to your Module’s build.gradle. The latest version found in maven’s repository is V1.5.2

Implementation 'org. Jetbrains. Kotlinx: kotlinx coroutines -- core: 1.5.2'Copy the code

Create coroutines

There is no new in Kotlin, so instead of using a similar new method, coroutine creation is done using functions wrapped by Kotlin coroutines:

  • RunBlocking: Used independently, the delay operation in the code block blocks the thread. The other three are GlobalScope apis
  • Launch: Creates a thread
  • Async: Creates a coroutine with a return value of Deferred, a subclass of Job
  • WithContext: Instead of creating a new coroutine, a block of code is executed within the specified coroutine

Ex. :

Println (" coroutine start :${Date()} ") runBlocking {println(" coroutine start :${Date()} ") for (I in 1.. 5) {println (" coroutines perform [$I] : ${Date ()} ") delay (500 l)} println (" coroutines end: ${Date ()} ")} println (" coroutines is over: ${Date ()} ")Copy the code

The execution result

Fri Sep 17 05:04:36 GMT+08:00 2021 I/ system. out: the coroutine has not started :Fri Sep 17 05:04:36 GMT+08:00 2021 I/ system. out: Fri Sep 17 05:04:36 GMT+08:00 2021 I/ system. out: Fri Sep 17 05:04:36 GMT+08:00 2021 I/ system. out: [3]:Fri Sep 17 05:04:37 GMT+08:00 2021 I/ system. out: Fri Sep 17 05:04:37 GMT+08:00 2021 I/ system. out: [5]:Fri Sep 17 05:04:38 GMT+08:00 2021 I/ system. out: Fri Sep 17 05:04:38 GMT+08:00 2021 I/ system. out: End :Fri Sep 17 05:04:39 GMT+08:00 2021 I/ system. out: The coroutine is ended :Fri Sep 17 05:04:39 GMT+08:00 2021Copy the code

Launch function

The function definitions

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job
Copy the code

Parameters:

  • Dispatchers.IO, dispatchers. Main, dispatchers. Unconfined (not specified, use current thread)
  • Start: Start mode, DEFAULT, ATOMIC, UNDISPATCHED, LAZY (LAZY load, only when called)
  • Block: The body of the closure method, which is executed by the coroutine

Return value: Job, the operable method of the coroutine, including

  • Job.start (), which starts the coroutine manually during lazy loading
  • Job.join () : wait for the execution of another coroutine to complete with thread. join
  • Job.cancel () : Cancels a coroutine
  • Job.cancelandjoin () : waits for the coroutine to complete and then cancels

Ex. :

Globalscope.launch (dispatchers.io) {println(" Coroutine start :${Date()} ") for (I in 1.. 3) {println (" coroutines mission 1 print ($I) : ${Date ()} ")} delay (500) for (I in 1.. Var job = globalscope.launch (start = coroutinestart.lazy){$job = globalscope.launch (start = coroutinestart.lazy){$job = globalscope.launch (start = coroutinestart.lazy) Println (" Coroutine start :${Date()} ")} delay(500) job.start(Copy the code

Async function

Like launch, async returns a value

Globalscope.launch (dispatchers.unconfined) {val deferred = Globalscope.async {delay(5000L) println("async execution ") return @async "king"} println(" coroutine start :${Date()}") val result = deferred.await() println(" coroutine return value :$result") Println (" Coroutine end :${Date()}")}Copy the code

Async returns a value of type Deferred, inherits from Job and extends an await method to receive the return value from block closures. Async does not block the current thread, but suspends the coroutine (blocking the coroutine).

cancel

Aunch {} returns Job, async{} returns Deffer. Both Job and Deffer have cancel() methods, which are used to retrieve coroutines.

abnormal

There are two kinds of exceptions to Kotlin coroutines:

  • A CancellationException thrown by the suspend method inside the coroutine because the coroutine is canceled
  • Normal exceptions, this type of exception, have two mechanisms for exception propagation
    • Launch: An exception is automatically thrown to the parent coroutine, which will cause the parent coroutine to exit
    • Async: Exposes exceptions to the user (by catching exceptions thrown by deffer.await()))
Fun main() = runBlocking {val job = globalscope.launch {// root coroutine with launch println("Throwing" exception from launch") throw IndexOutOfBoundsException() } job.join() println("Joined failed job") val deferred = GlobalScope.async { // root coroutine with async println("Throwing exception from async") throw ArithmeticException() // } try {deferred.await() println("Unreached")} catch (e: ArithmeticException) { println("Caught ArithmeticException") } }Copy the code

Output to view the delivery of exceptions

Throwing exception from launch
Exception in thread "DefaultDispatcher-worker-2 @coroutine#2" java.lang.IndexOutOfBoundsException
Joined failed job
Throwing exception from async
Caught ArithmeticException
Copy the code

Global handling exception

/ / handler global val handler = CoroutineExceptionHandler {_, The exception - > println (" CoroutineExceptionHandler got $exception ")} val job. = GlobalScope launch (handler) {/ / root coroutines, Throw AssertionError()} val Deferred = GlobalScope.async(handler) {// Also root coroutine, But using async instead of launch throw ArithmeticException() without printing anything relies on the user to call deferred.await()} joinAll(job, deferred)Copy the code

Suspension and recovery of coroutines

Functions in coroutines are called sequentially by default

Suspend fun doSomethingUsefulOne(): Int {delay(1000L) return 13} suspend fun doSomethingUsefulTwo(): Int {delay(1000L) return 29} val time = measureTimeMillis {val one = doSomethingUsefulOne() val two = doSomethingUsefulTwo() println("The answer is ${one + two}") } println("Completed in $time ms")Copy the code

The output

The answer is 42
Completed in 2017 ms
Copy the code

suspend

Suspend refers to suspending and is used to modify methods. Code is run sequentially when multiple methods suspend are written in a coroutine

The context of coroutines

As mentioned, the context can be specified when the coroutine is started,

The coroutine context is divided into four types dispatchers. Default, dispatchers. IO, dispatchers. Main, dispatchers. Unconfined

Dispatchers.Default

Default scheduler. It uses the JVM’s shared thread pool, and the scheduler’s maximum concurrency is the number of CPU cores, which defaults to 2

Dispatchers.IO

IO scheduler, which streams blocked IO tasks into a shared thread pool so that the current thread is not blocked. The thread pool size for environment variables kotlinx. Coroutines. IO. The value of parallelism, the default is 64 or the number of core highly actie.

The scheduler shares threads with dispatchers.default, so creating a new coroutine with withContext(dispatchers.io) does not necessarily cause a thread switch.

Dispatchers.Main

This scheduler restricts all execution to the UI main thread, which is dedicated to the UI and is represented in Android as the UI main thread

Dispatchers.Unconfined

Unconstrained scheduler, which does not restrict operations to execution on any thread — operations prior to the first suspension point are performed on the thread that initiated the coroutine, and the corresponding suspension function determines which thread to execute next after the suspension point resumes.

Simply put: from where to hang from where to recover

Use async concurrency

The two coroutine methods are directly independent of each other and can use concurrency to get results faster

val time = measureTimeMillis {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")
Copy the code

Structured concurrency using Async

suspend fun concurrentSum(): Int = coroutineScope { val one = async { doSomethingUsefulOne() } val two = async { doSomethingUsefulTwo() } one.await()  + two.await() }Copy the code

In this case, if an error occurs inside the concurrentSum function, and it throws an exception, all coroutines started in scope will be cancelled.

scope

CoroutineScope – CoroutineScope, used to manage coroutines, manage the content of

  • How coroutines are launched – it defines launch, async, withContext, and other coroutine launch methods (in extention mode), and within these methods defines how the context of the child coroutine is inherited.
  • Manages the coroutine life cycle – this defines the cancel() method, which cancels the current scope and all coroutines in the scope.

Distinguish between scope and context

By definition, CoroutineScope is similar to CoroutineContext. The ultimate purpose of CoroutineContext is both CoroutineContext, but Roman Elizarov, head of Kotlin CoroutineContext and Scope, says in CoroutineContext and Scope The only difference is the purpose of use – scope for managing coroutines; The context is simply a collection of records of the environment in which the coroutine runs. Their relationship is as follows

Asynchronous flow

Asynchronous flows in Kotlin are conceptually very similar to flows in RxJava and can be classified as reactive flows. And Kotlin also provides a library of responses to transform into other reactive flows

  • Kotlinx-coroutines -reactive Used to reactive Streams
  • Kotlinx-coroutines -reactor Is used for Project reactor
  • Kotlinx coroutines — rx2 for RxJava2

Using asynchronous streams

Val flow = flow {1 delay(1000L) emit(12) // 2 delay(1000L) emit(13)} runBlocking {flow.collect { println(it) } }Copy the code

channel

Use the Channel construct directly

val channel = Channel<Int>() launch { for (x in 1.. 5) channel.send(x * x) channel.close() } for (y in channel) println(y) println("Done!" )Copy the code

The use of produce

val channel = produce {
    send(12)
    send(13)
}

for (value in channel) {
    println(value)
}
Copy the code

\