Coroutine 을 알아보자

2024. 7. 19. 10:26

1. 코루틴 개념

모든 코루틴은 CouroutineScope 안에서 돌아감
코루틴의 핵심은 suspend
- 함수의 동작을 일시중단이 가능
코루틴 내에서 thread는 놀지않게함. 처리되는 쓰레드가 일시중단 될경우 다른 것들을 처리함

1.1 Builder

코루틴 함수들을 시작하는 지점이라고 보는 것이 옳음.
즉, 코루틴 함수 스택이 시작되는 지점.

1.1.1 launch

//구현부분
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
}

독립적으로 실행되는 새로운 coroutineScope를 생성
Thread를 생성하는 것과 비슷함.
- 즉, 코루틴을 실행한 thread를 멈추지 않음.

fun main()
{
  GlobalScope.launch {
      delay(1000L)
      println("aa")
  }
}

//위와 같이 할 경우 "aa"는 실행되지 않음. 왜냐면 프로그램이 끝나버리니까.

CoroutineScope의 확장함수

CoroutineScope.launch()

1.1.2 runBlocking

코루틴을 실행한 thread를 중단(blocking) 시킴
- 새로운 코루틴을 시작후 완료될 때까지 현재 스레드를 중단 가능한 상태로 블로킹
- dispatcher를 쓰면 다른 스레드에서 실행시킬수 있지만, 어쨌든 해당 스레드는 블로킹됨
프로그램이 끝나지 않도록 하는 목적으로만 쓰임(현재는 suspend와 runTest를 사용함…)
- Main함수 → suspend
- Unit test 함수 → runTest

<aside> 💡 runBlocking은 다른 코루틴의 자식이 될 수 없음.

</aside>

1.1.3 async

launch와 동일하지만 값을 반환함.
CoroutineScope의 확장함수

CoroutineScope.async()

1.2 CouroutineScope

runBlocking과 같이 중단시킴
- 아직은 잘 모르겠으나 runBlocking보다 좋다고 함
- runBLocking은 실행한 현재 thead를 block 시키지만 얘는 어쨌든 코루틴의 블럭을 써서 메인 스레드를 중지시키지 않는 느낌

suspend fun main(): Unit = coroutineScope {
  launch {
    delpy(1000L)
    println("world")
  }
  println("hello")
}

1.3 CoroutineContext

코루틴의 데이터를 저장하고 전달하는 객체
- 컬렉션으로 처리됨
- data object
코루틴의 부모자식 관게를 Context를 상속받는것임
Job, CorutineName, CoroutineDispatcher와 같은 객체들의 모음임

1.3.1 컨텍스트 합치기

- 연산자를 지원하며 같은 내용은 덮어쓰기, 다른 내용은 추가됨

data class CoroutineName(
    val name: String
) : AbstractCoroutineContextElement(CoroutineName)
{
    override fun toString(): String = "CoroutineName($name)"

    companion object Key : CoroutineContext.Key<CoroutineName>
}

fun main() {
    val ctx1: CoroutineContext = CoroutineName("Name1")
    println(ctx1[CoroutineName]?.name)

    val ctx2: CoroutineContext = CoroutineName("Name2")
    println(ctx2[CoroutineName]?.name)

    val ctx3 = ctx1 + ctx2
    println(ctx3[CoroutineName]?.name)
}

1.4 Job

CoroutineContext에 포함된 인터페이스로써 state를 가짐
수명을 가지고 있으며 취소가능
모든 코루틴은 자신만의 job을 가지고 있으며, CoroutineContext에서 상속되지 않는 유일한 값임

기본적으로 active한 상태에서 시작하지만 옵셔널하게 lazy 시작을 위해 new 상태로 시작할 수 있음
active 상태에서 job이 실행되고 코루틴은 Job을 실행할 수 있음

구분 내용

new	시작 가능한 상태
active	JOB을 실행할 수 있는 상태
completing	잡을 완료하고 자식들을 기다리는 상태
completed	자식들도 완료된 상태
cancelling	에러 후처리 상태
canceled	에러 후처리 완료

1.4.1 Job Factory 함수

아래와 같이 Job을 코루틴 없이 생성후 할당할 수 있음
팩토리함수로 만들경우 자식들이 완료되도 completed 상태로 가지않고 active상태로 존재
- 계속 다른 코루틴에서 사용될 수 있기 때문에
따라서 아래와 같이 해줘야 됨

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        delay(1000)
        println("Text 1")
    }
    launch(job) {
        delay(1000)
        println("Text 2")
    }

    job.children.forEach { it.join() }
}

1.4.2 Job의 취소

취소될경우 cancelling 상태로 변경되고 exception이 떨어짐
그래서 try catch로 잡기보단 finally로 후처리 되어야 할것들을 처리해주는게 좋음

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        try {
            delay(1000)
            println("Text 1")
        }
        finally {
            println("finally")

            launch {
                println("will not be print")
            }

            withContext(NonCancellable) {
                println("will be print")
            }
        }
    }

    job.cancelAndJoin()
    println("done")
}

취소된 context 내에서는 코루틴을 추가로 실행될 수 없음… 단 withContext를 통해 처리가능

1.4.3 invokeOnCompletion

completed 또는 canceled 되기전에 실행되어야 할 내용을 정의

job.invokeOnCompletion { cause: Throwable? ->
        println("job completed with $cause")
    }

1.5 예외처리

기본적으로 모든 예외는 부모-자식간으로 다 전파됨.
중단점이 없을경우 모든 코루틴에 에러가 전파되어 모든 코루틴이 취소됨.

1.5.1 supervisorJob()

기본적으로 자식에서 발생하는 예외를 무시시킴

suspend fun main(): Unit = runBlocking {

    launch(SupervisorJob()) {
        try {
            delay(1000)
            println("Text 1")

            throw Exception("Exception 1")
        }
        finally {
            println("finally")

            launch {
                println("will not be print")
            }
        }
    }

    launch(SupervisorJob()) {
        delay(1000)
        println("Text 2")
    }
    println("done")
}

//done
//Text 2

supervisorJob은 단 하나의 자식만 가짐
- 따라서 아래와 같이 쓸경우 exception이 전파됨

suspend fun main(): Unit = runBlocking {

    launch(SupervisorJob()) {
        launch {
            delay(1000)
            throw Exception("exception")
        }

        launch() {
            delay(2000)
            println("출력 안됨")
        }
    }
    delay(3000)
    println("done")
}
// Exception.....
//done

코루틴 내에서 try - catch는 Exception을 핸들링 할 수 없음

(2) Flow

stream 처리와 비슷
생성자 → 중간 → 소비자(collect)
중간
- filter
- map

fun main(): Unit = runBlocking(Dispatchers.Default) {
    flow { // producer
        repeat(times = 3) { data ->
            delay(300L)
            emit(data)
            println("produced $data")
        }
    }.collect { data -> // consumer
        delay(700L)
        println("consumed $data")
    }
}

(3) channel

val channel = lifecycleScope.produce<Int>{
	repeat(5){
		channel.send(20 + it)
	}
}

lifecycleScope.launch {
	for(temperature in channel) {
		Log.d("TEST", "$temperature")
	}
}

vs Flow

val flow: Flow<Int> = flow {
	repeat(5){
		emit(20 + it)
	}
}

lifecycleScope.launch {
	flow.collect { temperature ->
		Log.d("TEST", "$temperature")
	}
}

→ Chanel은 수신하기도 전에 보내기 시작. hot

→ Flow는 수신쪽에서 collect를 호출해야 보내기 시작. cold

(4) dispatcher

실제 thread에 코루틴을 실행시키는 것
- thread pool 같은 느낌이 없지는 않음.
각 플랫폼에 따라 정해진 값이 있음
- android에 경우 main, io, default 등
- 커스텀한 이름을 지정할 수 있음

(5) 에러처리

전파
- 자동으로 에러를 전파함.
- launch, actor
노출
- await()를 호출하기 전까지는 전파안됨
- async, produce

→ 위 두개의 차이로 인해 잘 핸들링 해야함.

→ 코루틴의 에러는 try - catch로 처리되지 않음. → 이미 전파됨.

supervisorJon

상위로의 전파를 막음

suspend fun main() {
    // Child Job #2
    CoroutineScope(Dispatchers.Default + coroutineExceptionHandler).launch {
        // Child Job #5
        launch(SupervisorJob()) {
            delay(300)
            throw Exception("first coroutine Failed")
        }.join()
        // Child Job #6
        launch(SupervisorJob()) {
            delay(400)
            println("second coroutine succeed")
        }.join()
    }.join()
}

Caught: first coroutine Failed
second coroutine succeed

Process finished with exit code 0

supervisorScope
- 범위를 지정해서 쓸수도 있음

suspend fun main() {
    // Child Job #2
    CoroutineScope(Dispatchers.Default + coroutineExceptionHandler).launch {
        supervisorScope {
            // Child Job #5
            launch {
                delay(300)
                throw Exception("first coroutine Failed")
            }.join()
            // Child Job #6
            launch {
                delay(400)
                println("second coroutine succeed")
            }.join()
        }
    }.join()
}

2. Coroutine in Vertx

(1) CoroutineVerticle()

Verticle의 start, stop을 suspend 함수로 시작하게 해줌
- 또한 내부적으로 coroutine context를 vertx eventloop로 처리될 수 있도록 기본 dispatcher를 처리
- verticle 이외의 코루틴에 영향이 없도록 supervisorJob으로 error handling 처리함
아래는 코드 일부

abstract class CoroutineVerticle : Verticle, CoroutineScope {

  private lateinit var vertxInstance: Vertx
  private lateinit var context: Context

  override val coroutineContext: CoroutineContext by lazy 
      { context.dispatcher() + SupervisorJob() }

  override fun init(vertx: Vertx, context: Context) {
    this.vertxInstance = vertx
    this.context = context
  }

  override fun getVertx(): Vertx = vertxInstance

  override fun start(startFuture: Promise<Void>?) {
    launch {
      try {
        start()
        startFuture?.complete()
      } catch (t: Throwable) {
        startFuture?.fail(t)
      }
    }
  }

  override fun stop(stopFuture: Promise<Void>?) {
    val job = coroutineContext.job
    launch {
      try {
        stop()
        stopFuture?.complete()
      } catch (t: Throwable) {
        stopFuture?.fail(t)
      } finally {
        job.cancel()
      }
    }
  }

vertx에서는 runBlocking을 쓰면 안됨.

대안

기본적으로 coroutineVerticle로 시작하면 모든 function을 suspend로 만들수 있음
그 외에 새로운 블럭을 만들어야 한다면

val result: Future<List<CardInfo>> = vertxFuture {
      service.getCardRecommList(pageNo, pageLimit, cate_grp)
    }

S'story