$30 off During Our Annual Pro Sale. View Details »

Kotlin Coroutines

Jussi Pohjolainen
March 28, 2024
440

Kotlin Coroutines

Jussi Pohjolainen

March 28, 2024
Tweet

Transcript

  1. Process and Thread • Process • A process is an

    instance of a program running on a computer. • It is a self-contained execution environment and typically has its own memory space. • Processes are independent of each other • Thread • A thread is a smaller sequence of programmed instructions within a process • Threads within the same process share the same memory space but operate independently • Multithreading allows a process to perform multiple tasks concurrently
  2. What is a Thread? • A thread is a single

    sequential flow of execution of tasks of a process in an operating system (OS). • OS threads are at the core of Java’s concurrency model • Java threads are just a wrapper at OS thread • JVM must work with the underlying OS to create and manage that thread, which is quite expensive because the JVM must communicate with the OS back and forth throughout the thread’s lifetime • This switching is an expensive operation, which makes threads expensive
  3. Java public class Main { public static void main(final String[]

    args) { for(int i=0; i<10; i++) { generateThread().start(); } } public static Thread generateThread() { Thread t = new Thread(() -> { for(int i=0; i<10; i++) { System.out.println(Thread.currentThread().getName() + " i = " + i); try { Thread.sleep(1000); } catch(InterruptedException e) { e.printStackTrace(); } } }); return t; } }
  4. Thread 1 / CPU Thread 2 / CPU Thread 3

    / CPU Java supports multithreading at OS Level. True parallel execution
  5. Multithreaded environment THREAD 1 i = 1 i = 2

    i = 3 i = 4 i = 5 i = 6 i = 7 i = 8 THREAD 2 i = 1 i = 2 i = 3 i = 4 i = 5 i = 6 i = 7 i = 8
  6. One threaded environment THREAD 1 method1 i = 1 i

    = 2 i = 3 i = 4 THREAD 1 method2 i = 1 i = 2 i = 3 i = 4
  7. JavaScript const sleep = async (msecs) => new Promise((resolve) =>

    { setTimeout(() => { resolve(); }, msecs); }); const count = async () => { for (let i = 0; i <= 10; i++) { console.log(i); await sleep(1000); } }; const main = async () => { count(); count(); }; main();
  8. Kotlin import kotlin.concurrent.thread fun main() { val t1 = Thread({

    println("hello") }) t1.start() val t2 = Thread() { println("hello") } t2.start() Thread { println("hello") }.start() thread { println("hello") } }
  9. Kotlin import kotlin.concurrent.thread fun main() { repeat(10) { generateThread().start(); }

    } fun generateThread() = thread(false) { for(i in 1..10) { println("${Thread.currentThread().name} i = $i") Thread.sleep(1000); } }
  10. Kotlin coroutine • Kotlin coroutines are entirely managed by the

    language • Do not require communication with the underlying OS to be created and managed • Coroutines are lightweight and can be launched millions at once • Java 19 introduced the concept of virtual threads. Virtual threads are lightweight threads that are managed by the JVM rather than the OS
  11. Kotlin Coroutine • Coroutines are not bound to any particular

    thread • Coroutines can be • Creating async tasks only one thread (like js) – giving illusion of simultaneous execution. • Creating async tasks using thread pool • Dispatchers.IO, Dispatchers.Default • Default thread pool amount is the number of CPU cores • Runtime.getRuntime().availableProcessors()
  12. Two java threads import kotlin.concurrent.thread fun main() { thread {

    repeat(10) { println(it) Thread.sleep(1000) } } thread { repeat(10) { println(it) Thread.sleep(1000) } } } App creates two threads and when threads have finished app ends
  13. Two coroutines fun main() { runBlocking { launch { repeat(10)

    { println("A ${Thread.currentThread().name}: $it") delay(1000) } } launch { repeat(10) { println("B ${Thread.currentThread().name}: $it") delay(1000) } } } } Only one thread is used! Runblocking scope is for main method or testing, it waits that all coroutines are done and then ends the app
  14. Two coroutines fun main() { runBlocking { launch { repeat(10)

    { println("A ${Thread.currentThread().name}: $it") delay(1000) } } launch { repeat(10) { println("B ${Thread.currentThread().name}: $it") delay(1000) } } } } It is now synchronous
  15. Two coroutines fun main() { runBlocking { launch(Dispatchers.Default) { repeat(10)

    { println("A ${Thread.currentThread().name}: $it") delay(1000) } } launch(Dispatchers.Default) { repeat(10) { println("B ${Thread.currentThread().name}: $it") delay(1000) } } } } Dispatchers.Default => CPU intensive work, creates thread pool
  16. Two coroutines fun main() { runBlocking { launch(Dispatchers.Default) { repeat(100)

    { println("A ${Thread.currentThread().name}: $it") delay(1000) } } launch(Dispatchers.Default) { repeat(100) { println("B ${Thread.currentThread().name}: $it") delay(1000) } } } } It will do some multitasking now
  17. Job and cancel fun main() { runBlocking { val job1

    : Job = launch(Dispatchers.Default) { repeat(10) { println("A ${Thread.currentThread().name}: $it") delay(1000) } } job1.cancel() } } It can cancel the coroutine
  18. Job and join fun main() { runBlocking { val jobs

    = List(10) { launch(Dispatchers.Default) { repeat(10) { println("${Thread.currentThread().name}: $it") delay(200) } } } jobs.forEach { it.join() } // Wait for all jobs to complete println("done") } } Waits until the job is done
  19. Job and cancel fun main() { runBlocking { val job1

    : Job = launch(Dispatchers.Default) { repeat(10) { println("A ${Thread.currentThread().name}: $it") delay(200) } } job1.cancel() println("Stop") } } Cancels immediately
  20. async fun main() { runBlocking { val coroutine: Deferred<Int> =

    async(Dispatchers.Default) { var sum = 0 repeat(10) { sum += it delay(100) } sum } val result = coroutine.await() println(result) } } In lambda, the last expression value in the block is return value Will wait until it receives the result Will hold the sum
  21. async val coroutines : List<Deferred<Int>> = List(100) { async(Dispatchers.Default) {

    var sum = 0 println(Thread.currentThread().name) repeat(10) { sum += it delay(100) } sum } } val results : List<Int> = coroutines.awaitAll() println(results) Uses thread pool Will wait until it receives the result
  22. Suspend Function • Function that can be paused and resumed

    at a later time witohout blocking the thread • Non-blocking
  23. fun main() { // Create the HttpClient val client =

    HttpClient.newHttpClient() // Build the HttpRequest for the Chuck Norris API val request = HttpRequest.newBuilder() .uri(URI.create("https://api.chucknorris.io/jokes/random")) .GET() // Use the GET method .build() // Send the asynchronous request and process the response var completableFuture = client.sendAsync (request, HttpResponse.BodyHandlers.ofString()) println("Start") completableFuture.thenApply { it.body() // let's get the body of the http response }.thenAccept { println(it) // and then print it }.join() } callbacks
  24. fun fetchChuckNorrisJoke(): CompletableFuture<String> { val client = HttpClient.newHttpClient() val request

    = HttpRequest.newBuilder() .uri(URI.create("https://api.chucknorris.io/jokes/random")) .GET() .build() return client.sendAsync(request, HttpResponse.BodyHandlers.ofString()) .thenApply { it.body() } } fun main() { val futureJoke = fetchChuckNorrisJoke() futureJoke.thenAccept { response -> println("Received response: $response") }.join() // Wait for the asynchronous operation to complete } Helper method Callback
  25. suspend fun fetchChuckNorrisJoke(): String { val value = suspendCancellableCoroutine {

    continuation -> val client = HttpClient.newHttpClient() val request = HttpRequest.newBuilder() .uri(URI.create("https://api.chucknorris.io/jokes/random")) .GET() .build() client.sendAsync(request, HttpResponse.BodyHandlers.ofString()) .thenApply { it.body() }.thenAccept { continuation.resume(it) } } return value } Initializes suspend operation that can be cancelled Resume suspended coroutine with the result Asynchronously fetch data
  26. fun main() { runBlocking { try { val joke =

    fetchChuckNorrisJoke() println("Received joke: $joke") } catch (e: Exception) { println("Failed to fetch joke: ${e.message}") } } }
  27. fun main() { runBlocking { val jokeDeferred1 = async {

    fetchChuckNorrisJoke() } val jokeDeferred2 = async { fetchChuckNorrisJoke() } // Wait for both operations to complete and get their results. val joke1 = jokeDeferred1.await() val joke2 = jokeDeferred2.await() // Print the results. println("Joke 1: $joke1") println("Joke 2: $joke2") } }
  28. Turning sync to async • Last example had already threading

    and callbacks • With suspendCancellableCoroutine it is possible to turn the function into a suspend function • If function does not have threading built in, you can use withContext
  29. import kotlinx.coroutines.* import java.nio.file.Files import java.nio.file.Paths suspend fun read(path: String)

    : String { println("${Thread.currentThread().name} - read") return withContext(Dispatchers.IO) { println("${Thread.currentThread().name} - read") Files.readString(Paths.get(path)) } } fun main() { runBlocking { launch { println("A: ${Thread.currentThread().name} - launch") delay(1000) val content = read("./build.gradle.kts") println(content) } launch { println("B: ${Thread.currentThread().name} - launch") delay(1000) val content = read("./build.gradle.kts") println(content) } } } main - thread main - thread Changes the context to thread pool!
  30. suspend fun fetchRandomJoke(): String { val client = HttpClient.newHttpClient() val

    request = HttpRequest.newBuilder() .uri(URI.create("https://api.chucknorris.io/jokes/random")) .GET() .build() return withContext(Dispatchers.IO) { println(Thread.currentThread().name) val response = client.send(request, HttpResponse.BodyHandlers.ofString()) if(response.statusCode() != 200) { throw RuntimeException("Failed to fetch joke, status code: ${response.statusCode()}") } response.body() } } fun main() { runBlocking { println(Thread.currentThread().name) val deferredList = List(10) { async { fetchRandomJoke() } } val jokes = deferredList.awaitAll() println(jokes) } } main - thread Do asynchro nously 10 fetching Changes the context to thread pool! May wait if thread pool is full