[XX캠퍼스] 17.Kotlin Coroutines & Flow ( 채널 기초 )

https://dalinaum.github.io/coroutines-example/17

채널 기초

 

채널

채널은 일종의 파이프입니다.

송신측에서 채널에 send로 데이터를 전달하고

수신 측에서 채널을 통해 receive 받습니다.

(trySend와 tryReceive도 있습니다. 과거에는 null을 반환하는 offer와 poll가 있었습니다.)

 

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*


fun main() = runBlocking<Unit> {
    val channel = Channel<Int>()
    launch {
        for (x in 1..10) {
            channel.send(x) // suspension point
        }
    }

    repeat(10) {
            println(channel.receive()) // suspension point
            //체널에 데이터가 없으면 잠들었다가, 데이터가 생기면 작업 수행. 
    }
    println("완료")
}

실행결과
1
2
3
4
5
6
7
8
9
10
완료

* 기본적으로 send와 receive는 suspension Point 로 본다. 

체널에 데이터가 없는경우 잠들었다가, 데이터가 있으면 깨어나서 작업을 수행 

 

* trySend와 tryReceive 는 suspension Point가 아니기때문에 기다리지 않는다. 

* 기본적으로 send와 receive를 주로 쓰고, 특별한 경우 trySend와 tryReceive 사용. 

 

같은 코루틴에서 채널을 읽고 쓰면?

send나 receive가 suspension point이고 서로에게 의존적이기 때문에

같은 코루틴에서 사용하는 것은 위험할 수 있습니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*


fun main() = runBlocking<Unit> {
    val channel = Channel<Int>()
    launch {
        for (x in 1..10) {
            channel.send(x)  // send와 동시에 런치 블록 자체가 잠이 들어버려
                             // receive를 수행하지 못함.. 
        }

        repeat(10) {
            println(channel.receive())
        }
        println("완료")
    }
}

무한으로 대기하는 것을 볼 수 있습니다.

 

 

채널 close

채널에서 더 이상 보낼 자료가 없으면 close 메서드를 이용해 채널을 닫을 수 있습니다.

채널은 for in 을 이용해서 반복적으로 receive할 수 있고 close되면 for in은 자동으로 종료됩니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*


fun main() = runBlocking<Unit> {
    val channel = Channel<Int>()
    launch {
        for (x in 1..10) {
            channel.send(x)
        }
        channel.close()
    }

    for (x in channel) {
        println(x)
    }
    println("완료")
}

실행결과 
1
2
3
4
5
6
7
8
9
10
완료

for 문을 통해서 channel을 가지고 사용할때는 반드시 close가 필요하다 안그러면 무한으로 대기. 

 

 

---

채널 프로듀서

생산자(producer)와 소비자(consumer)는 굉장히 일반적인 패턴입니다.

채널을 이용해서 한 쪽에서 데이터를 만들고 다른 쪽에서 받는 것을 도와주는 확장 함수들이 있습니다.

1. produce 코루틴을 만들고 채널을 재공합니다.
2. consumeEach 채널에서 반복해서 데이터를 제공합니다.

ProducerScope는 CoroutineScope 인터페이스와 SendChannel 인터페이스를 함께 상속받습니다.

그래서 코루틴 컨텍스트와 몇가지 채널 인터페이스를 같이 사용할 수 있는 특이한 스코프입니다.

 

produce를 사용하면 

ProducerScope를 상속받은 ProducerCoroutine 코루틴을 얻게 됩니다.

 

참고:

우리가 흔히 쓰는 runBlocking은 BlockingCoroutine을 쓰는데

이는 AbstractCoroutine를 상속받고 있어요.

 

결국 코루틴 빌더는 코루틴을 만드는데 이들이 코루틴 스코프이기도 한거죠.

AbstractCoroutine은 

JobSupport, Job(인터페이스), Continuation(인터페이스), CoroutineScope(인터페이스)을 상속받고 있고요.

 

Continuation은 다음에 무엇을 할지, Job은 제어를 위한 정보와 제어,

CoroutineScope는 컨텍스트 제공의 역할을 합니다. 

JobSupport는 잡의 실무(?)를 한다고 봐야죠.

 

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*


fun main() = runBlocking<Unit> {
    val oneToTen = produce { 
    // 스스로 체널을 만들고 채널을 만든다. 
    // 코루틴 스코프를 만든다 ( ProducerScope = CoroutineScope + SendChannel
        for (x in 1..10) {
            channel.send(x)
        }
    }
    

    oneToTen.consumeEach {
        println(it)
    }
    println("완료")
}

• produce 파트를 함수로 분리해봅시다.
• suspend 함수와 CoroutineScope의 확장 함수의 방식을 해봅시다. (produce는 CoroutineScope의 확장 함수)