Channel

채널이란?

• Deferred values provide a convenient way to transfer a single value between coroutines. Channels provide a way to transfer a stream of values.
• 간단하게 두 코루틴의 사이를 연결해주는 스트림이다

채널을 왜 사용할까요?

two separate coroutines we cannot simply emit the elements by a function call. we need to establish some communication between two coroutines

쓰레드간에 공유할 자원이 필요할때 우리는 두개의 쓰레드가 동시에 그걸 쓰거나 읽게 하지 못하도록 자원을 lock하거나 메모리에 의존하게된다. 메모리 공유! 이것이 흔히 쓰레드가 커뮤니케이션하는 방식이다. 하지만 이렇게하면 데드록, 레이스 컨디션 같은 이슈가 발생할 수 있다.

사용하는 방법은 무엇일까요?

공식문서에서 말하는 방법

• A Channel is conceptually very similar to BlockingQueue. One key difference is that instead of a blocking put operation it has a suspending send, and instead of a blocking take operation it has a suspending receive.

val channel = Channel<Int>()
launch {
    // this might be heavy CPU-consuming computation or async logic, we'll just send five squares
    for (x in 1..5) channel.send(x * x)
}
// here we print five received integers:
repeat(5) { println(channel.receive()) }
println("Done!")

 

위 코드의 실행 결과

 

 

기본 사용법 - 다른 블로그 참고 자료

하나의 코루틴에서 여러 코루틴에 보내는 방법 사용

val ordersChannel = Channel<Menu>()
    launch { // casher coroutine
        for (order in orders) {
            ordersChannel.send(order)
        }
        ordersChannel.close()
    }
 
    val takenTime = measureTimeMillis {
        coroutineScope {
            // 바리스타 1, 2는 ordersChannel을 통해 내려온 주문들을 받아 커피를 만든다.
            launch(CoroutineName("barista-1")) { makeCoffee(ordersChannel) }
            launch(CoroutineName("barista-2")) { makeCoffee(ordersChannel) }
        }
    }

 

기본 사용법 - Kotlin 개발자 블로그 참고

만약 플로우를 사용하면 이런모습의 순서로 진행될것.

순차적으로, 더 빠르게 값을 불러오고 싶다면?

We need to decouple the emitter and the collector — run the emitter in a separate coroutine from the collector, so that they can be concurrent.

 

위 그림은 아래 코드를 시각화 한것입니다.

val ints: Flow<Int> = flow { 
    for (i in 1..10) {
        delay(100)
        emit(i)
    }
}

fun <T> Flow<T>.buffer(size: Int = 0): Flow<T> = flow {
    coroutineScope {
        val channel = produce(capacity = size) {
          collect { send(it) }
        }
        channel.consumeEach { emit(it) }
    }
}

suspend fun main() {
    val time = measureTimeMillis {
        ints.buffer().collect { 
            delay(100)
            println(it) 
        }
    }
    println("Collected in $time ms")
}