Kotlin Channel 의 기본 개념에 대해 알아보려고 합니다~

Channel

Kotlin Channel은 코루틴 간 통신을 위한 동시성 프리미티브로, 데이터를 안전하게 전송하고 공유할 수 있는 메커니즘을 제공합니다.

즉 비동기 데이터 스트림이라고 생각할 수 있습니다.

한쪽에서 데이터를 보내고(send), 다른 쪽에서는 데이터를 받는(receive) 파이프라인 처럼 작동합니다.

위처럼 한 코루틴은 생산자로서 데이터를 생성하여 채널에 보내고, 다른 코루틴은 소비자로서 채널에서 데이터를 받습니다. 생성자 - 소비자 패턴을 쉽게 구현할 수 있습니다.

기본 사용법

자세한 개념에 대해 알아보기 전 기본 사용 방법에 대해 알아보겠습니다.

fun main() = runBlocking<Unit> {
    val channel = Channel<Int>()
    launch {
        for (x in 1..5) { // 1부터 5까지 반복문
            println("send ${x * x}")
            channel.send(x * x) // send 시 채널에 데이터 전송
        }
    }
    // receive 시 채널로부터 데이터 받음
    repeat(5) { println("Receive = ${channel.receive()}") }
    channel.close()
    println("Done!")
}

• send 시 채널에 데이터를 전송합니다.
• receive 시 채널로부터 데이터를 받을 수 있습니다.
• close() 시 채널을 닫습니다. 닫힌 후에는 더 이상 데이터를 보낼 수 없지만, 이미 채널에 있는 데이터는 여전히 수신할 수 있습니다.

val channel = Channel<Int>()
launch {
    for (x in 1..5) channel.send(x * x)
    channel.close() // we're done sending
}
// here we print received values using `for` loop (until the channel is closed)
for (y in channel) println(y)
println("Done!")

위처럼 작성 시 채널이 닫힐 때까지 모든 값을 수신하는 코드가 됩니다.

Channel 구현부

public interface SendChannel<in E> 
public interface ReceiveChannel<out E>
public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> 

채널은 SendChannel , ReceiveChannel 인터페이스가 존재하고 Channel 인터페이스가 이 두개의 인터페이스를 따르고 있습니다.

SendChannel과 ReceiveChannel에 대해선 나중에 알아보고 우선 채널에 대해 자세히 알아보겠습니다.

public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> {
    public companion object Factory {
        public const val UNLIMITED: Int = Int.MAX_VALUE
        public const val RENDEZVOUS: Int = 0
        public const val CONFLATED: Int = -1
        public const val BUFFERED: Int = -2
        internal const val OPTIONAL_CHANNEL = -3
        public const val DEFAULT_BUFFER_PROPERTY_NAME: String = "kotlinx.coroutines.channels.defaultBuffer"
        internal val CHANNEL_DEFAULT_CAPACITY = systemProp(DEFAULT_BUFFER_PROPERTY_NAME,
            64, 1, UNLIMITED - 1
        )
    }
}

public fun <E> Channel(
    capacity: Int = RENDEZVOUS,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    onUndeliveredElement: ((E) -> Unit)? = null
): Channel<E> 

우선 Channel 인스턴스를 생성할 때 전달할 수 있는 파라미터에 대해 알아보겠습니다.

capacity(채널의 타입)

채널의 버퍼 용량을 결정합니다.

capacity 는 백프레셔 처리할 방법을 정합니다.

백프레셔란 생산자가 소비자보다 빠르게 데이터를 발행하는 상황에 있어 Channel은 여러 버퍼링 전략을 통해 해결할 수 있도록 하는 전략을 말합니다..

capacity로 여러 값들을 설정할 수 있는데 이는 위의 Channel 인터페이스에 정의된 Factory에서 확인할 수 있습니다. 각 채널 용량으로 사용할 수 있는 미리 정의된 상수들입니다.

• RENDEZVOUS
  • 버퍼가 없는 채널. 송신자와 수신자가 만날 때(rendezvous)만 데이터가 전송됩니다.
  • send 호출은 상대편에서 receive 를 호출할 때 까지 일시 중단됩니다(그 반대도 마찬가지). 즉, sendScope에서 send 후 receive가 불려야 다음 코드블럭으로 넘어갈 수 있습니다.
  • 생산자와 소비자가 동시에 작동해야 할 때, 각 항목이 즉시 처리되어야 할 때 적합합니다.
• UNLIMITED
  • 제한 없는 버퍼를 가진 채널. send 호출은 절대 일시 중단되지 않습니다.
  • 메모리 사용량에 주의해야 합니다.
  • 소비자가 모든 항목을 처리해야 하모 메모리가 충분할 때 적합합니다.
• CONFLATED
  • 병합 채널. 새 요소가 이전 요소를 덮어씁니다.
  • 버퍼 크기는 1이며, 가장 최근 값만 유지됩니다.
  • 최신 값만 중요하고 중간 값은 무시해도 될 때 사용합니다.
• BUFFERED
  • 기본값(64)을 버퍼 크기로 사용하는 채널입니다.

onBufferOverflow 정책

버퍼가 가득 찼을 때의 동작을 지정합니다.

이는 BufferOverFlow Enum Class로 이뤄져 있습니다.

public enum class BufferOverflow {
    SUSPEND,
    DROP_OLDEST,
    DROP_LATEST
}

각 정책에 대해 알아보겠습니다~

• SUSPEND
  • 버퍼가 가득 차면 send 작업이 일시 중단됩니다. 버퍼에 공간이 생길 때까지 송신자는 대기합니다.
  • 백프레셔를 제공하여 생산자가 소비자보다 빠르게 생성하지 못하도록 합니다.
  • 모든 항목이 처리되어야 하고 백프레셔가 필요할 때 적합합니다.
• DROP_OLDEST
  • 버퍼가 가득 차면 가장 오래된 요소(버퍼의 맨 앞 요소)가 삭제되고 새 요소가 추가됩니다.
  • 즉, 큐의 맨 앞 요소를 버리고 새 요소를 뒤에 추가합니다.
  • 처리 속도가 중요하고 오래된 데이터를 포기(최신 데이터가 중요함)할 수 있는 경우 유용합니다.
• DROP_LATEST
  • 버퍼가 가득 차면 가장 최근에 추가된 요소(버퍼의 맨 뒤 요소)가 삭제되고 새 요소가 추가됩니다.
  • 즉, 큐의 맨 뒤 용소를 버리고 새 요소를 추가합니다.
  • 가장 최근 데이터가 아닌 이전 데이터를 유지하는 것이 중요할 때 유용합니다.

onUndeliveredElement

전달되지 않은 요소를 처리하는 콜백 함수

다음과 같은 상황에서 호출됩니다.

• send 작업이 취소되었을 때
• 채널이 닫힌 후 남아있는 요소가 있을 때
• 요소 처리 중 예외가 발생했을 때

기본적으로 null이기 때문에 전달되지 않은 요소에 대한 특별한 처리는 이뤄지지 않습니다.

이 콜백은 전달되지 않은 요소의 리소스를 정리하는 데 유용합니다.

채널이 I/O fㅣ소스나 네트워크 연결과 같은 자원을 포함한 객체를 전송하는 경우 이러한 리소스를 적절히 해제할 수 있습니다.

언제 사용할까?

Channel 이 무슨 역활을 하는지, 파라미터에 어떤 것이 있고 동작이 어떻게 달라지는 지 가볍게 알아봤습니다.

그럼 이러한 특성을 고려했을 때 Channel은 언제, 어떤 상황에 사용했을 때 유용할까요?

• 코루틴 간의 통신이 필요할 때
• 백프레셔 관리가 필요할 때
• 비동기 작업 간 흐름 제어

등 이 있지만 안드로이드에서 이벤트 처리 에 매우 유용하다고 생각합니다.

이벤트 처리의 특성으로

1. 이벤트를 받아줄 수 있는 구독자가 생길 때 까지 없어지지 않음
2. 이벤트가 처리되면 해당 이벤트는 사라짐

이 있습니다. 자세한 내용으로는 (https://medium.com/prnd/viewmodel에서-더이상-eventflow를-사용하지-마세요-3974e8ddffed 의 블로그를 참조하시면 이벤트 처리에 대한 매우 좋은 내용을 보실 수 있습니다.

이 때 BUFFERED을 사용한 채널을 사용하면 네트워크 도중 홈버튼을 눌렀을 때 이벤트가 도착해서 이를 처리할 수 있고 처리되었다면 사라지기 때문에 Configuration Change가 일어나도 같은 이벤트가 발생되지 않습니다.