[Coroutines] Channel, Pipeline

✅ Coroutine Channel

기본적으로 Deffered는 Coroutine간의 단일 값을 전달하는 방법을 제공하고, Channel은 Stream의 값을 전달하는 방법을 제공한다.

 

Channel은 BlockingQueue와 비슷하다. 한가지 차이점은 BlockingQueue의 put 함수 대신 일시중단 가능한 send()함수가 존재하고, take 함수 대신 receive() 함수가 존재한다.

 

channel은 더이상 사용하지 않을 때 close()를 이용하여 원소가 없음을 나타낼 수 있다. 이 덕분에 receiver side에서는 for loop를 사용하여 채널로부터 element를 편하게 수신할 수 있다. close()가 호출되는 순간 반복이 중단되지만, close 되기 전까지 보낸 모든 요소는 수신될 수 있도록 보장한다. 만약 cancel() 함수를 호출한다면 채널을 닫고 전송한 요소들도 모두 제거하기 때문에 마지막 전송요소에 대해서도 수신을 할 수 없게 된다.

 

식당에서 요리를 주문받는 채널을 예시로 들어보자.

 

fun setRestaurantOrderChannel() {  
    runBlocking {  
        val cuisineList = listOf<String>("스테이크", "파스타", "피자", "샐러드")  
        val kitchenChannel = Channel<String>()  
        launch {  
            cuisineList.forEach {  
                kitchenChannel.send(it)  
            }  
            kitchenChannel.close()  
        } 

        for(cuisine in kitchenChannel) Log.d("cuisine", "${cuisine}")  
        Log.d("Close", "Restaurant")  
    }  
}

 

이렇게 Channel을 만드는데 producer라는 편리한 빌더가 있다.

그리고 consumer입장에서 for loop문을 대신해서 값을 receive할 수 있는 consumeEach() 함수가 존재한다.

 

fun getSquares() {  
    runBlocking {  
        val squares = produceSquares()  
        squares.consumeEach { println(it) }  
        println("Done!")  
    }  
}  

fun CoroutineScope.produceSquares(): ReceiveChannel<Int> = produce {  
    for (x in 1..5) send(x * x)  
}

 

Channel의 Pipeline은 Stream값을 무한대로 생성하는 패턴이다.

하단의 예시는 Pipeline에서 만들어낸 스트림을 소비하면서 가공하여 다른 값을 만들어내고 있다.

 

fun main() {  
    runBlocking {  
        val squares = produceSquares(produceNumber())  
        squares.consumeEach { println(it) }  
        println("Done!") 
        coroutineContext.cancelChildren() // 하위coroutine 취소
    }  
}  

fun CoroutineScope.produceNumber(): ReceiveChannel<Int> = produce {  
     var x = 1  
     while (true) send(x++)  
}  

fun CoroutineScope.produceSquares(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {  
    for (x in 1..5) send(x * x)  
}

 

Channel에서 값을 전송하는 메서드에는 send()와 offer()가 있다. 둘의 차이점은 suspend function의 여부와 반환값이다.

 

send()는 일시중단이 가능하다.

또한 Channel의 Buffer가 가득찼거나 존재하지 않는다면 호출자를 중단하고, Channel이 닫힌 경우에는 Exception이 발생한다.

 

offer()의 경우에는 채널의 용량이 초과하였다면 값을 전송하고 true를 반환하며,

초과하지 않았다면 요소를 전송하지 못하고 false를 반환한다.

 

수신하는 메서드에도 비슷하게 poll() 함수와 receive()함수가 있다.

 

receive의 경우에는 채널이 비어있으면 대기하고, poll의 경우에는 channel이 비어있거나 닫혀있으면 close cause exception이 발생한다.