12장 디스패처1

디스패처를 이용해 코루틴이 실행되어야 할 스레드 또는 스레드풀 결정할수 있다.

어떤 스레드에서 실행될지 정하는것은 코루틴 컨텍스트이다.

기본 디스패처 Dispatcher.Default

- 설정하지 않으면, 기본적으로 설정됨.

- CPU집약적인 연산을 하도록 설계

- 코드가 실행되는 컴퓨터의 CPU 개수와 동일한 수의 스레드풀을 가진다.

suspend fun main(): Unit = coroutineScope {
    repeat(1000) {
        launch {
            List(1000) { Random.nextLong() }.maxOrNull()

            val threadName = Thread.currentThread().name
            println("Run on thread: $threadName")
        }
    }
}

 

위 코드를 실행한 컴퓨터는 8개의 스레드를 가지고 있다.

Run on thread: DefaultDispatcher-worker-4
Run on thread: DefaultDispatcher-worker-5
Run on thread: DefaultDispatcher-worker-8
Run on thread: DefaultDispatcher-worker-5
Run on thread: DefaultDispatcher-worker-7

기본 디스패처 제한

Dispatchers.Default의 limitedParallelism을 사용하면 디스패처가 같은 스레드 풀을 사용하지만 같은 시간에 특정 수 이상의 스레드를 사용하지 못하도록 제한이 가능하다

 

*Dispatchers.IO 에서도 사용 가능하며 더 중요하다.

private val dispatcher = Dispatchers.Default.limitedParallelism(5)

 

메인 디스패처 

안드로이드에서 메인 스레드는 UI와 상호작용하는데 사용하는 유일한 스레드이다. 메인 스레드는 자주 사용되어야 하지만, 아주 조심스럽게 다뤄야 한다. 블로킹 대신 중단하는 라이브러리를 사용하고, 복잡한 연산을 하지 않는다면, Dispatcher.Main으로 충분하다. 

IO 디스패처 

IO연산으로 스레드를 블로킹할 때 사용하기 위해 설계되었다.

 

suspend fun main(): Unit = coroutineScope {
    val time = measureTimeMillis {
        coroutineScope {
            repeat(50) {
                launch(Dispatchers.IO) {
                    Thread.sleep(1000)
                }
            }
        }
    }
    println("time = $time")
}

//time = 1043

 

위 코드는 왜 1초밖에 걸리지 않을까?

처음에는 스레드풀이 비어 있지만, 더 많은 스레드가 필요해지면, 스레드가 생성되고 작업이 끝날 때까지 활성화된 상태로 유지됩니다.

활성화된 스레드수가 너무 많다면, 성능이 점점 떨어지게 되고, 결국 메모리가 부족하게 된다.

 

같은 시간에 사용할 수 있는 스레드 수를 제한한 디스패처가 필요하다.

 

Dispatchers.Default 프로세서가 가지고 있는 코어의 개수로 제한

Dispatchers.IO 64개로 제한

suspend fun main(): Unit = coroutineScope {
    repeat(1000) {
        launch (Dispatchers.IO) {
            Thread.sleep(200)

            val threadName = Thread.currentThread().name
            println("Running on thread: $threadName")
        }
    }
}

//Running on thread: DefaultDispatcher-worker-17
//Running on thread: DefaultDispatcher-worker-55
//Running on thread: DefaultDispatcher-worker-11
//Running on thread: DefaultDispatcher-worker-58
//Running on thread: DefaultDispatcher-worker-43

 

 *Dispatchers.Default과 Dispatchers.IO은 같은 스레드 풀을 공유한다. 스레드는 재사용되고 다시 분배될 필요가 없다.

 

suspend fun main(): Unit = coroutineScope {
    launch(Dispatchers.Default) {
        println("Dispatchers.Default: ${Thread.currentThread().name}")
        withContext(Dispatchers.IO) {
            println("Dispatchers.IO: ${Thread.currentThread().name}")
        }
    }
}

Dispatchers.Default: DefaultDispatcher-worker-1
Dispatchers.IO: DefaultDispatcher-worker-1

 

 Dispather.IO 가장 흔한 경우는 라이브러리에서 블로킹 함수를 호출해야 하는 경우

class DiscUseRepository(
    private val discReader: DiscReader,
): UserRepository {
    override suspend fun getUser(): UserData = 
        withContext(Dispatchers.IO) {
            UserData(discReader.read("userName"))
        }
}

 * 너무 많은 스레드를 블로킹하는 서비스를 이용할 경우 limitedParallelism을 이용한다

커스텀 스레드 풀을 사용하는 IO 디스패처

Dispatchers.IO에는  limitedParallelism 함수를 위해 정의된 특별한 작동 방식이 있다.

limitedParallelism 독립적인 스레드풀을 가진 새로운 디스패처를 생성한다.

 

limitedParallelism 100개의 스레드를 사용하는 Dispatchers.IO에서 실행하면 1초

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

@OptIn(ExperimentalCoroutinesApi::class)
suspend fun main(): Unit = coroutineScope {
    launch(Dispatchers.Default) {
        printCoroutinesTime(Dispatchers.IO)
    }

    launch {
        val dispatcher = Dispatchers.IO.limitedParallelism(100)
        printCoroutinesTime(dispatcher)
    }
}


suspend fun printCoroutinesTime(
    dispatcher: CoroutineDispatcher
) {
    val test = measureTimeMillis {
        coroutineScope {
            repeat(100) {
                launch(dispatcher) {
                    Thread.sleep(1000)
                }
            }
        }
    }
    println("$dispatcher took $test")
}

//LimitedDispatcher@180d7435 took 1022
//Dispatchers.IO took 2024

 

갯수 제한이 없는 스레드 풀이 있다고 가정하면

Dispatcher.IO = pool.limitedParallelism(64)

Dispatcher.IO.limitedParallelism(x) = pool.limitedParallelism(x)

 

limitedParallelism는 독립적인 디스패처를 생성한다.

 

정해진 수의 스레드 풀을 가진 디스패처

Executors 클래스를 스레드의 수가 정해져 있는 스레드풀이나 캐싱된 스레드 풀을 만들수 있다.

ExecutorService, Executor 인터페이스를 구현한 클래스는 asCoroutineDispatcher로 변경 가능하다.

 

* ExecutorService.asCoroutineDispatcher로 만들어진 디스패처는 close로 닫아야한다.

=>안하면,  사용하지 않는 스레드가 다른 서비스와도 공유되지 않으면서 살아있는 상태로 유지됨.

 

싱글스레드로 제한된 디스패처

다수의 스레드를 사용하는 디스패처에서는 공유상태로 인한 문제점을 고려해야된다.

import kotlinx.coroutines.*


suspend fun main(): Unit = coroutineScope {
    var i = 0
    repeat(10_000){
        launch(Dispatchers.IO) {
            i++
        }
    }
    delay(1000)
    println(i)
}

//9789

 

10,000 보다 적은 값이 출력되는 이유는 동일한 시간에 다수의 스레드가 공유된 상태를 변경했기 때문이다.

싱글스레드를 사용하면 특별한 동기화 조치 없이 해결가능하다.

import kotlinx.coroutines.*


suspend fun main(): Unit = coroutineScope {
    var i = 0

    val dispatcher = Dispatchers.Default.limitedParallelism(1)
    repeat(10_000){
        launch(dispatcher) {
            i++
        }
    }
    delay(1000)
    println(i)
}

//10000

 

* 단 하나의 스레드만 사용하기 때문에, 해당 스레드가 블로킹 당하면, 작업이 순차적 실행으로 되어 처리 속도가 느려진다.

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis


suspend fun main(): Unit = coroutineScope {
    var i = 0

    val dispatcher = Dispatchers.Default.limitedParallelism(1)
    val job = Job()

    repeat(5){
        launch(dispatcher+job) {
            Thread.sleep(1000)
        }
    }
    job.complete()
    val time = measureTimeMillis { job.join() }
    println("Took $time ms")
}

//Took 5034 ms