[Kotlin Coroutines] Channels

kotlinlang.org/docs/channels.html#channel-basics

Channels - Help | Kotlin

kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-channel/

Channel - kotlinx-coroutines-core

 

kotlinx-coroutines-core / kotlinx.coroutines.channels / Channel

interface Channel<E> : SendChannel<E>, ReceiveChannel<E> (source)

 

Channel은 값의 stream을 전달한다. sender과 reciver사이의 통신을 위한 non-blocking 원시타입.

 

Channel 기본

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!")

 

abstract suspend fun send(element: E): Unit

 

element를 채널로 보낸다.

이 channel의 버퍼가 가득차거나 존재하지 않으면 호출부를 일시중지시킨다(suspend).

channel이 close 됐을 경우에 호출되면  exception을 발생시킨다.

 

abstract val isClosedForSend: Boolean

 

close 함수의 실행에 의해 channel이 close 되면 true를 반환한다. true를 반환할 때 send 함수, offer 함수를 호출하면 exception이 발생한다.

 

abstract suspend fun receive() : E

 

channel이 비어있지 않다면 channel의 element 하나를 반환하고 제거한다. 

channel이 비어있으면 caller를 중단한다.

channel이 close 된 상태이면 ClosedReceiveChannelException을 발생시킨다.

다른 exception으로 인해 channel이 close 되면, 이 channel은 failed channel이라 부른다.

 

abstract val isClosedForReceive: Boolean

 

SendChannel 측에서 close 함수가 실행되어 channel이 close 되고 이전에 받은 모든 element들이 이미 다 보내지면 true를 반환한다. true를 반환할 때 receive 함수를 호출하면 ClosedReceiveChannelException이 발생한다. 다른 exception에 의해 channel이 close 될 경우도 channel이 닫힌 것으로 간주하나, 그 channel은 failed channel로 부른다. 

 

abstract fun offer(element: E) : Boolean

 

용량을 초과하지 않으면 element를 이 channel에 즉시 추가하고 true를 반환한다.

용량을 초과하면 false를 반환한다. 

send의 동기적인 변형이다. 

channel이 send에 닫혀져 있다면 (isClosedForSend 값이 true) exception을 발생시킨다.

offer로 false를 반환받으면, 이 element가 consumer에게 전달되지 않았다는 것을 보장하고 onUndeliveredElement를 호출하지 않는다.

channel이 닫혀있다면, exception을 호출하기 전에 onUndeliveredElement를 호출한다. 

Channel 닫기(close)와 반복(iteration)

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!")

 

abstract suspend fun close(cause: Throwable? = null): Boolean

 

channel에  더 이상 전송시킬(send) element가 없다고 알리기 위해 channel을 닫을(close) 수 있다.

close는 멱등 연산자라서 이 함수를 최초 실행한 이후의 실행 후에는 아무 변화도 없고 false를 리턴시킨다. 

개념적으로 이 함수는 특수 "close token"를 channel에 보낸다. 이 close token을 iteration은 받자마자 중단하기 때문에, close 전에 받은 모든 element들이 receive로 호출되어 나갔다는 것이 보장된다.

 

→멱등에 관해 정리한 내용 확인

 

* Channel의 함수

abstract operator fun iterator(): ChannelIterator<E>

 

for loop를 이용해 이 channel에서의 element들을 받는 새 iterator를 반환한다.

보통 exception cause 발생이 없이 isClosedForReceived 함수의 반환값이 true일 때 Iteration은 완료하고, channel이 fail될 경우 close를 발생시킨 exception을 throw 한다. 

 

-> Channel close() 호출 후 for 루프 돌릴 수 있다.

 

channel 생성자(producer)들 만들기

코루틴이 연속된 element들을 생산하는 패턴(producer-consumer 패턴의 일부)은 흔히 쓰인다. 이 producer는 channel을 매개변수로 하는 함수로 만들 수는 수 있지만, 이는 함수가 반드시 값을 반환해야 한다는 상식과 어긋난다.(... 그래서 그렇게 하면 안된다고 하는 거죠? (＠_＠;) 안되기 때문에 produce 함수를 소개해주는 거죠?)

 

-> producer-consumer 패턴 확인

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

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

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

 

@ExperimentalCoroutinesApi fun <E> CoroutineScope.produce(
    context: CoroutineContext = EmptyCoroutineContext,
    capacity: Int = 0,
    block: suspend ProducerScope<E>.() -> Unit
): ReceiveChannel<E>

 

producer를 쉽게 만드는 produce 라는 이름의 coroutine builder가 있다. 문서보기

produce - kotlinx-coroutines-core

데이터 스트림을 생성해 channel로 보내는 새 코루틴을 시작한다.

코루틴에 의해 생성된 element들을 receive하는 데 쓰이는 ReceiveChannel이 반환된다.

 

context : CoroutineScope.coroutineContext

capacity : channel 버퍼의 용량 - 결과 channel이 특정 capacity(용량) 값에 따라 달라진다. 

block : coroutine 코드 - ProducerScope라서 코루틴이 바로 send()를 호출할 수 있다. 코루틴이 완료되면 channel은 닫힌다. receive channel이 취소되면 실행 중이던 coroutine이 취소된다.

 

exception 발생 시 coroutine은 channel을 닫고 결과 channel은 fail 된다.

 

@ExperimentalCoroutinesApi interface ProducerScope<in E> : 
    CoroutineScope,
    SendChannel<E>

 

문서보기

produce coroutine builder를 위한 scope. 이 coroutine이 elemet들을 보내는(send) channel의 참조.

 

suspend fun <E> ReceiveChannel<E>.consueEach(action: (E) -> Unit) : Unit

 

cosumer 쪽인 channel에서 for 루프를 대체하는 extension function.

receive된 각 element를 대상으로 action을 수행하고 action 실행 후에 channel을 cancel한다. 

 

Pipelines

파이프라인 하나는 코루틴 하나가 무한대의 스트림 값을 생성하는 패턴이다.

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++) // 1부터 시작하는 정수의 무한대 스트림
}

또 다른 코루틴들은 그 스트림을 소비하고, 몇 가지 처리를 하고 다른 결과값들을 생성한다.

fun CoroutineScope.square(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (x in numbers) send(x * x) // 제곱
}

메인 코드는 전체 파이프라인을 시작하고 연결한다. 

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

fun main() = runBlocking {
    val numbers = produceNumbers() // 1 부터 증가해 생성되는 정수들
    val squares = square(numbers) // 정수들 제곱
    repeat(5) {
        println(squares.receive()) // 처음 5개만 프린트
    }
    println("Done!") // we are done
    coroutineContext.cancelChildren() // cancel children coroutines
}

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++) // infinite stream of integers starting from 1
}

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

결과

1
4
9
16
25
Done!

-> 한 코루틴이 채널 하나에 stream producting 하는 중에 다른 코루틴이 그 채널의 stream을 cosuming 해서 다른 channel을 생성했다.

 

소수와 파이프라인

코루틴들이 하나의 파이프라인을 이용해서 소수를 생성하는 예제를 통해 파이프라인들의 극한까지 살펴 보겠다. 

 

무한대의 숫자들로 시작해보자.

fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
    var x = start
    while (true) send(x++) // infinite stream of integers from start
}

아래의 파이프라인 단계는 들어오는 숫자 스트림을 필터링하는데, 주어진 소수로 나눠지는 모든 숫자들을 삭제한다.

fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
    for (x in numbers) if (x % prime != 0) send(x)
}

숫자 2부터 시작하는 숫자 스트림으로 시작하는 파이프라인을 생성하고, 현재 채널에서 소수를 가져다가, 찾은 각 소수마다 새로운 파이프 단계를 시작한다.

numbersFrom(2) -> filter(2) -> filter(3) -> filter(5) -> filter(7) ...

아래 예시는 첫 10개의 소수들을 프린트하고, 메인 쓰레드의 콘텍스트에서 전체 파이프라인을 실행한다. 모든 코루틴들이  메인 runBlocking 코루틴의 스코프 안에서 시작되기 때문에 우리는 시작된 모든 코루틴들의 명시적인 리스트를 유지할 필요가 없다.

 

우리는 cancelChildren 확장함수를 사용해 처음 10개의 소수를 프린트 한 후에 모든 자녀 코루틴들을 취소한다.

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

fun main() = runBlocking {
    var cur = numbersFrom(2)
    repeat(10) {
        val prime = cur.receive()
        println(prime)
        cur = filter(cur, prime)
    }
    coroutineContext.cancelChildren() // cancel all children to let main finish    
}

fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
    var x = start
    while (true) send(x++) // infinite stream of integers from start
}

fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
    for (x in numbers) if (x % prime != 0) send(x)
}

결과

2
3
5
7
11
13
17
19
23
29

 

우리는 똑같은 파이프라인을 standard library의 iterator 코루틴 빌더를 이용해 만들 수 있다.

 

produce -> iterator 교체

send -> yield 교체

receive -> next 교체

ReceiveChannel -> Iterator 교체

-> 코루틴 스코프 제거

runBlocking을 사용할 필요없다.

 

교체한 코드. 결과 값은 위의 결과 값과 같다.

fun main() {
    var cur = numbersFrom(2)
    repeat(10) {
        val prime = cur.next()
        println(prime)
        cur = filter(cur, prime)
    }
}

fun numbersFrom(start: Int) = iterator<Int> {
    var x = start
    while (true) yield(x++) // infinite stream of integers from start
}

fun filter(numbers: Iterator<Int>, prime: Int) = iterator<Int> {
    for (x in numbers) if (x % prime != 0) yield(x)
}

 

그러나 위와 같이 채널들을 이용한 파이프라인의 이점은 Dispatchers.Default context에서 코드를 실행할 때 multiple CPU 코어들을 실제로 사용할 수 있다는 것이다. 

 

어쨌든, 소수들을 찾는 극단적이고 실용적이지도 않는 예제였다. 실제로 파이프라인들은 정말로 몇가지 다른 suspending invocation(호출, 원격 서비스로 비동기 호출 같이)을 포함하고, 이 파이프라인들은 sequence/iterator를 사용해서 만들어질 수는 없다. 왜냐하면 seqeunce/iterator는 완전히 비동기인 produce와 다르게 임의적인 중단을 허용하지 않기 때문이다. 

 

fun <T> runBlocking(

    context: CoroutineContext  = EmptyCoroutineContext,

    block: suspend CoroutineScope.() -> T

): T

 

context: 코루틴의 콘텍스트. 기본값은 현재 쓰레드의 이벤트 루프이다.

block: 코루틴 코드 

 

새로운 코루틴 하나를 실행하고 이 코루틴이 완료될 때까지 현재 쓰레드를 인터럽트로 막는다. 이 함수를 코루틴에서 사용해서는 안된다. 이것은 일반 차단 코드와 일시중단 스타일로 작성된 라이브러리와 연결해서 주요 기능들과 테스트에서 사용되도록 설계되었다.

 

이 빌더의 기본 CoroutineDispatcher는 이 코루틴이 완성될 때까지 중단된 쓰레드에서 연속적으로 진행하는 이벤트 루프의 내부 구현이다. 

 

context 변수에 CoroutineDispatcher가 명시적으로 지정되면 현재 쓰레드가 막혔을 동안 지정된 dispatcher의 context가 새로운 코루틴을 실행한다. 만약 지정된 dispatcher가 또 다른 runBlocking의 이벤트 루프라면, 이 호출은 바깥 루프를 사용한다.

 

이 막힌 쓰레드가 인터럽트되면, 코루틴 job은 취소되고, runBlocking은 InterruptedException을 호출한다.

 

fun CoroutineConetext.cancelChildren(

    cause: CancellationException? = null

): Unit

 

선택적인 취소 원인을 가지고 job 스스로의 상태를 건드리지 않고, 이 context 안의 Job의 모든 자녀들을 취소한다.

 

*Standard library의 함수

fun <T> iterator(

    block: suspend SeqeunceScope<T>.() -> Unit

): Iterator<T>

 

값을 하나씩 느리게 생성하는 Iterator를 빌드한다.

예제

import kotlin.test.*

fun main(args: Array<String>) {
    val collection = listOf(1, 2, 3)
    val wrappedCollection = object : AbstractCollection<Any>() {
        override val size: Int = collection.size + 2

        override fun iterator(): Iterator<Any> = iterator {
            yield("first")
            yieldAll(collection)
            yield("last")
        }
    }

    println(wrappedCollection) // [first, 1, 2, 3, last]
}

예제

fun main(args: Array<String>) {
    val iterable = Iterable {
        iterator {
            yield(42)
            yieldAll(1..5 step 2)
        }
    }
    val result = iterable.mapIndexed { index, value -> "$index: $value" }
    println(result) // [0: 42, 1: 1, 2: 3, 3: 5]

    // can be iterated many times
    repeat(2) {
        val sum = iterable.sum()
        println(sum) // 51
    }
}

 

Fan-out

하나의 producer 코루틴의 channel 1개, 처리 코루틴 여러개.

여러 코루틴들은 동일한 채널을 수신(receive)해서 서로간에 작업을 분배할 수 있다.

매 초마다 값이 증가하는 정수를 생산하는 producer 코루틴으로 시작한다.

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1 // start from 1
    while (true) {
        send(x++) // produce next
        delay(100) // wait 0.1s
    }
}

여러 processor 코루틴들을 가질 수 있다. 예제 코드에서는 코루틴의 id와 수신한 숫자를 print한다.

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
    for (msg in channel) {
        println("Processor #$id received $msg")
    }
}

이제 5개의 프로세서들을 시작하고 약 1초 동안 동작시킨다. 어떤 일이 일어나는지 보자.

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

fun main() = runBlocking<Unit> {
    val producer = produceNumbers()
    repeat(5) { launchProcessor(it, producer) }
    delay(950)
    producer.cancel() // cancel producer coroutine and thus kill them all
}

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1 // start from 1
    while (true) {
        send(x++) // produce next
        delay(100) // wait 0.1s
    }
}

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
    for (msg in channel) {
        println("Processor #$id received $msg")
    }    
}

출력은 다음과 비슷할 것이다. 각 정수를 수신한 프로세서의 id는 다를 수 있지만 말이다.

Processor #2 received 1
Processor #4 received 2
Processor #0 received 3
Processor #1 received 4
Processor #3 received 5
Processor #2 received 6
Processor #4 received 7
Processor #0 received 8
Processor #1 received 9
Processor #3 received 10

producer 코루틴을 취소하면 해당 채널도 닫힌다. 따라서 결과적으로 프로세서 코루틴들이 작업하고 있는 채널의 iteration은 종료된다.

launchProcessor 코드에서 fan-out을 수행하기 위해 for 루프로 channel을 명시적으로 반복하는 방법에 주목해라. consumEach와 다르게, 이 for 루프 패턴은 여러 코루틴들을 사용하기에 완벽하게 안전하다. 만약 프로세서 코루틴들 중 하나가 실패한다면, 다른 코루틴들은 여전히 계속 채널을 처리할 것인 반면, consumeEach를 통해 작성된 프로세서는 항상 기본이 되는 채널을 정상적이거나 비정상적인 완료에 소비(취소)를 한다.

 

-> fan-out에서 여러 processor 코루틴들 중 하나가 실패해도 다른 코루틴들이 작업을 정상적으로 진행할 수 있으나, consumeEach로 작성된 processor 코루틴은 코루틴이 실패하면 작업 중인 channel이 닫히기 때문에 다른 코루틴들의 작업이 중단된다.

 

Fan-in

여러 코루틴들이 하나의 channel에 접근 가능.

여러 코루틴들이 동일한 하나의 channel에 값을 send할 수 있다. 
예를 들어, string값들의 channel 하나를 가지고 있고, 지연 시간을 가지고 특정 string을 이 channel에 반복적으로 send하는 suspending function이 있다고 하자.

suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) {
    while (true) {
        delay(time)
        channel.send(s)
    }
}

만약 string값들을 보내는 코루틴 두 개를 시작한다면 어떤 일이 일어나는 지 보자.(이 예제에서 메인 코루틴의 자녀들로서 메인 쓰레드의 context에서 이 코루틴들을 실행했다.)

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

fun main() = runBlocking {
    val channel = Channel<String>()
    launch { sendString(channel, "foo", 200L) }
    launch { sendString(channel, "BAR!", 500L) }
    repeat(6) { // receive first six
        println(channel.receive())
    }
    coroutineContext.cancelChildren() // cancel all children to let main finish
}

suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) {
    while (true) {
        delay(time)
        channel.send(s)
    }
}

출력은 다음과 같다.

foo
foo
BAR!
foo
foo
BAR!

Buffered channels

지금까지 살펴본 channel들은 버퍼가 없었다. 버퍼가 없는 채널들은 sender와 receiver가 만날 때 요소들을 전송한다.(일명 랑데뷰. rendezvous 채널이라 부르더라). send가 먼저 실행되면 receive는 호출될 때까지 기다리고, receive가 먼저 호출되면 send가 호출될 때까지 중단된다.

 

Channel 팩토리 함수와 produce 빌더 둘다 capacity 매개변수를 버퍼사이즈로 사용할 수 있다. 버퍼는 중단 전에 여러 요소들을 sender가 전송하게 하고, 버퍼가 가득찰 때 차단되는 것은 지정된 capacity를 가진 BlockingQueue와 비슷하다. 

 

fun <E> Channel(

    capacity: Int = RENDEZVOUS,

    onBufferedOverflow: BufferOverflow = BufferOverflow.SUSPEND,

    pnUndeliveredElement: (E) -> Unit = null

): Channel<E>

 

capacity - 양수의 채널 용량 또는 Channel.Facotry에서 정의된 상수 중 하나

onBufferOverflow - 버퍼 오버플로우에서의 액션. 선택사항. 기본값은 전송 시도를 일시중단한다. capacity >=0이거나 capacity == Channel.BUFFERED 일 때만 지원된다. 최소 하나의 버퍼링된 요소가 있으면 채널을 암시적으로 생성한다.

onUndeliveredElement - 선택사항인 함수. 요소가 전송되었지만 소비자로 요소가 전송되지 못할 때 호출된다.

capacity < -2 경우에 IllegalArgumentException 을 호출한다.

 

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

fun main() = runBlocking<Unit> {
    val channel = Channel<Int>(4) // create buffered channel
    val sender = launch { // launch sender coroutine
        repeat(10) {
            println("Sending $it") // print before sending each element
            channel.send(it) // will suspend when buffer is full
            println("Sending $it done")
        }
    }
    // don't receive anything... just wait....
    delay(1000)
    sender.cancel() // cancel sender coroutine    
}

결과

Sending 0
Sending 0 done
Sending 1
Sending 1 done
Sending 2
Sending 2 done
Sending 3
Sending 3 done
Sending 4

처음 4개의 요소들은 버퍼에 추가되었고, sender는 다섯번째 요소를 send하기 위해 중단된다.

 

Channel들은 공평하다

여러 코루틴들에서 호출 순서에 관련해 채널의 전송(send)과 수신(receive) 연산은 공평하다. FIFO로 진행된다. receive를 호출한 첫번째 코루틴이 요소를 가져간다. 

아래 예시에서 두 코루틴 "ping"과 "pong"은 "ball" 객체를 공용 "table" 채널에서 수신하고 있다.

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

data class Ball(var hits: Int)

fun main() = runBlocking {
    val table = Channel<Ball>() // a shared table
    launch { player("ping", table) }
    launch { player("pong", table) }
    table.send(Ball(0)) // serve the ball
    delay(1000) // delay 1 second
    coroutineContext.cancelChildren() // game over, cancel them
}

suspend fun player(name: String, table: Channel<Ball>) {
    for (ball in table) { // receive the ball in a loop
        ball.hits++
        println("$name $ball")
        delay(300) // wait a bit
        table.send(ball) // send the ball back
    }
}

결과

ping Ball(hits=1)
pong Ball(hits=2)
ping Ball(hits=3)
pong Ball(hits=4)

"ping" 코루틴이 처음 시작되었고, 공을 수신하는 첫 코루틴이다. 비록 "ping" 코루틴이 테이블에 공을 전송한 후에 바로 공을 받는 것을 시작했으나, 공은 "pong" 코루틴에게 수신되었다. 왜냐하면 이미 "pong"이 공을 기다리고 있었기 때문이다.

 

종종 channel들의 실행이 실행 프로그램의 특성에 따라 unfair한 경우도 있다. 

 

Ticker channels

Ticker channel은 이 채널의 마지막 요소의 소비 이후에 delay가 발생할 때마다 Unit을 생성하는 특별한 랑데뷰 채널(버퍼링없는 채널)이다.

쓸데없는 독립형 채널로 보이지만, 복잡한 시간 기반 produce 파이프라인을 생성하거나 windowing(? 이해 못했습니다.), 시간종속 처리를 하는 연산들을 하는데 유용하다.

Ticker channel을 생성하려면 팩토리 함수 ticker를 사용한다. 더이상의 요소들이 필요하지 않다면 ReceiveChannel.cancel 함수를 사용한다.

 

@ObsoleteCoroutinesApi fun ticker(

    delayMillis: Long,

    initialDelayMillis: Long = delayMillis,

    context: CoroutineContext = EmptyCoroutineContext,

    mode: TickerMode = TickerMode.FIXED_PERIOD

): ReceiveChannel<Unit>

 

처음 지연 시간 후에 첫 아이템(Unit)을 생산하고 지연시간이 있는 후속 아이템들을 생산하는 채널을 생성한다. 

결과 채널은 랑데뷰 채널이다. 이 채널의 receiver가 채널의 element들의 수신을 따라가지 못하면, element들은 backpressure 때문에 send되지 못한다. 이러한 경우의 ticker의 실제 타이밍 행동은 기본값이 TickerMode.FIXED_PERIOD인 mode 매개변수에 의해 통제된다. 

cancel 호출 후에 바로 element 생산을 중단한다.

ticker는 consumer의 가능한 일시중지를 인지하고, 기본적으로 일시중지가 발생하면 다음에 생성되는 요소의 지연 시간을 조정해서 생성된 요소들의 고정된 비율을 유지하려고 한다. 

 

delayMillis - 각 요소들 사이의 밀리초 지연 시간

initialDelayMillis - 밀리 초의 지연 시간 후에 첫 요소가 생성될 것이다. 기본 값은 delayMillis와 같다.

context - producing 코루틴의 context

mode - 요소들이 수신되지 않을 때의 특정 행동. 기본 값이 TickerMode.FIXED_PERIOD

 

-> Backpressure 확인하기

 

@ObsoleteCoroutinesApi enum classTickerMode

 

ticker 함수를 위한 모드.

참고: ticker 채널은 현재 구조화된 동시성과 통합되지 않았고, 이 api는 향후에 변경될 것이다.

 

Enum value

FIXED_PERIOD - cosumer가 따라잡지 못하거나 느리다면 고정된 기간을 유지하기 위해 지연 시간을 조정한다.

FIXED_DELAY -  cosumer가 따라잡지 못하거나 느리면 생성된 element들 사이의 고정된 지연 시간을 유지한다.

 

-> @ObsoleteCoroutinesApi 확인하기

 

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

fun main() = runBlocking<Unit> {
    //ticker 채널 생성. 처음 0초 딜레이로 Unit 아이템을 produce하고
    //그 후로는 모든 Unit 아이템이 100초 간격으로(지연 시간으로) produce한다.
    val tickerChannel = ticker(delayMillis = 100, initialDelayMillis = 0)
    //1초 타임아웃 receive -> 반환 값 Unit. 생성하자마다 아이템을 produce했기에
    var nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
    println("Initial element is available immediately: $nextElement") // no initial delay

    //50초 타임아웃 receive -> 반환 값 null. 대략 51초까지 produce된 아이템이 없음
    nextElement = withTimeoutOrNull(50) { tickerChannel.receive() }
    println("Next element is not ready in 50 ms: $nextElement")

    //60초 타임아웃 receive -> 반환 값 Unit. 100초 경에 아이템이 produce 됐을 것.
    nextElement = withTimeoutOrNull(60) { tickerChannel.receive() }
    println("Next element is ready in 100 ms: $nextElement")

    // Emulate large consumption delays
    // cosumer의 150초 딜레이.
    println("Consumer pauses for 150ms")
    delay(150)
    // Next element is available immediately
    // 1초 타임아웃 receive -> 반환 값 Unit.200초에 이미 아이템이 produce 됐을 것.
    // 250초 경에 receive.
    nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
    println("Next element is available immediately after large consumer delay: $nextElement")
    // Note that the pause between `receive` calls is taken into account and next element arrives faster
    // 60초 타임아웃 receiver -> 반환 값 Unit. 
    // receive 호출 사이의 일시정지(delay)를 고려해서 Unit 요소가 빨리 produce 되었다.
    // 50초만에 아이템이 produce 됐다.  
    nextElement = withTimeoutOrNull(60) { tickerChannel.receive() }
    println("Next element is ready in 50ms after consumer pause in 150ms: $nextElement")

    tickerChannel.cancel() // indicate that no more elements are needed
}

print 결과

Initial element is available immediately: kotlin.Unit
Next element is not ready in 50 ms: null
Next element is ready in 100 ms: kotlin.Unit
Consumer pauses for 150ms
Next element is available immediately after large consumer delay: kotlin.Unit
Next element is ready in 50ms after consumer pause in 150ms: kotlin.Unit

 

---

발번역으로 그래도 여기까지 이해했다.