10.2 Buffered Channel

Buffered Channel

val channel = Channel<Int>(5)

• 디폴트 생성자를 통해 생성된 채널은 송신(생산자) 후 수신(소비자)자가 받 을 때 까지 기본적으로 정지된다( 하나씩 송신/수신)
• Channel 에 데이터를 담을 수 있는 버퍼의 개수를 정의할 수 있다
• 버퍼의 개수를 많이 해놓으면 빠른 처리 가능
  • 그렇다고 비즈니스 로직보다 많이하면 ui가 업데이트 되지 않을 수 있음
• Buffered Constant
  • Channel.BUFFERED
    • 버퍼를 64로 고정하며 65개 이상은 suspend 상태가 됨
  • Channel.CONFLATED
    • 버퍼에서 처리 못한 item은 버리고 새것으로 대체
  • Channel.UNLIMITED
    • 버퍼 사이즈를 메모리가 허락하는 한 무제한(메모리 부족 시 Runtime Error 발생)

 

Buffered Channel Overflow

val channel = Channel<Int>(5,BufferOverflow.DROP_OLDEST)
launch {
    (1..40).forEach {
        println("send $it")
        channel.send(it)
    }
}

delay(2000)
val result = launch {
    val channelSize = channel.iterator()
    while (channelSize.hasNext()){
        println("receive ${channelSize.next()}")
    }
}

 

• 생산과 소비가 원활하지 않는다면 지연이 발생할 수 있는 부분은 Channel Buffer를 이용해 처리
• Coroutine은 버퍼가 초과되면 기본적으로 먼저 버퍼에 들어온 오래된 값은 버린다( Buffer Overflow 발생)
• Channel Overflow 전략
  • BufferOverflow.DROP_OLDEST
    • 버퍼에 들어온 오래된 아이템은 버리고 새로운 아이템을 취한다
  • BufferOverflow.DROP_LATEST
    • 버퍼가 차면 새로운 데이터를 버린다
  • BufferOverflow.SUSPEND : DEFAULT
    • 버퍼가 idle 상태(유휴상태, suspend)에 있다가 다시 시작

 

Unlimited Channel

val channel = Channel<Int>(Channel.UNLIMITED)

• 무한대의 생산과 소비를 이루고자 할 때
• 생산 형태를 쉽게 구현하도록 제공하는 Builder로 produce가 존재
• 소비는 consumeEach( forEach 로 대체 가능) 등이 존재