Channels

val channel = Channel<Int>() launch { // 이 부분은 CPU를 많이 소모하는 계산 또는 비동기 논리가 될 수 있습니다. // 우리는 다섯 개의 제곱을 전송할 것입니다. for (x in 1..5) channel.send(x * x) } // 여기서 다섯 개의 정수를 수신하여 출력합니다: repeat(5) { println(channel.receive()) } println("Done!") // 출력: // 1 // 4 // 9 // 16 // 25 // Done!

val channel = Channel<Int>() launch { for (x in 1..5) channel.send(x * x) channel.close() // 전송이 완료되었습니다. } // 여기서 `for` 루프를 사용하여 수신된 값을 출력합니다 (채널이 닫힐 때까지) for (y in channel) println(y) println("Done!") // 출력: // 1 // 4 // 9 // 16 // 25 // Done!

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!") } // 출력: // 1 // 4 // 9 // 16 // 25 // Done!

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) }

val numbers = produceNumbers() // 1부터 시작하는 정수 생성 val squares = square(numbers) // 정수들을 제곱 repeat(5) { println(squares.receive()) // 처음 다섯 개 출력 } println("Done!") // 작업 완료 coroutineContext.cancelChildren() // 자식 코루틴 취소

fun CoroutineScope.numbersFrom(start: Int) = produce<Int> { var x = start while (true) send(x++) // 시작 숫자부터 시작하는 무한한 정수 스트림 }

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

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

var cur = numbersFrom(2) repeat(10) { val prime = cur.receive() println(prime) cur = filter(cur, prime) } coroutineContext.cancelChildren() // 메인 종료를 위해 모든 자식 코루틴 취소 // 출력: // 2 // 3 // 5 // 7 // 11 // 13 // 17 // 19 // 23 // 29

fun CoroutineScope.produceNumbers() = produce<Int> { var x = 1 // 1부터 시작 while (true) { send(x++) // 다음 숫자 전송 delay(100) // 0.1초 대기 } }

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

val producer = produceNumbers() repeat(5) { launchProcessor(it, producer) } delay(950) producer.cancel() // 생산자 코루틴을 취소하여 모두 종료 // 출력: // 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

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

val channel = Channel<String>() launch { sendString(channel, "foo", 200L) } launch { sendString(channel, "BAR!", 500L) } repeat(6) { // 처음 6개 수신 println(channel.receive()) } coroutineContext.cancelChildren() // 모든 자식을 취소하여 메인이 끝나도록 함 // 출력: // foo // foo // BAR! // foo // foo // BAR!

val channel = Channel<Int>(4) // 버퍼링된 채널 생성 val sender = launch { // 송신자 코루틴 시작 repeat(10) { println("Sending $it") // 각 요소를 보내기 전에 출력 channel.send(it) // 버퍼가 가득 차면 중단됨 } } // 아무것도 수신하지 않고 그냥 기다림... delay(1000) sender.cancel() // 송신자 코루틴 취소 // 출력: // Sending 0 // Sending 1 // Sending 2 // Sending 3 // Sending 4

data class Ball(var hits: Int) fun main() = runBlocking { val table = Channel<Ball>() // 공유 테이블 launch { player("ping", table) } launch { player("pong", table) } table.send(Ball(0)) // 공을 서비스함 delay(1000) // 1초 대기 coroutineContext.cancelChildren() // 게임 종료, 취소 } suspend fun player(name: String, table: Channel<Ball>) { for (ball in table) { // 공을 루프에서 수신 ball.hits++ println("$name $ball") delay(300) // 잠시 대기 table.send(ball) // 공을 다시 보냄 } } // 출력: // ping Ball(hits=1) // pong Ball(hits=2) // ping Ball(hits=3) // pong Ball(hits=4)

fun main() = runBlocking<Unit> { val tickerChannel = ticker(delayMillis = 200, initialDelayMillis = 0) // 틱커 채널 생성 var nextElement = withTimeoutOrNull(1) { tickerChannel.receive() } println("초기 요소는 즉시 사용 가능합니다: $nextElement") // 초기 지연 없음 nextElement = withTimeoutOrNull(100) { tickerChannel.receive() } // 이후 모든 요소는 200ms 지연 println("다음 요소는 100ms에 준비되지 않음: $nextElement") nextElement = withTimeoutOrNull(120) { tickerChannel.receive() } println("다음 요소는 200ms에 준비됨: $nextElement") // 대규모 소비 지연 모방 println("소비자가 300ms 동안 일시 중지합니다") delay(300) // 다음 요소는 즉시 사용 가능 nextElement = withTimeoutOrNull(1) { tickerChannel.receive() } println("대규모 소비자 지연 후 다음 요소는 즉시 사용 가능합니다: $nextElement") // `receive` 호출 사이의 일시 중지가 고려되어 다음 요소의 도착이 더 빨라짐 nextElement = withTimeoutOrNull(120) { tickerChannel.receive() } println("소비자 300ms 일시 중지 후 100ms에 다음 요소가 준비됨: $nextElement") tickerChannel.cancel() // 더 이상 요소가 필요하지 않음을 표시 } // 출력: // 초기 요소는 즉시 사용 가능합니다: kotlin.Unit // 다음 요소는 100ms에 준비되지 않음: null // 다음 요소는 200ms에 준비됨: kotlin.Unit // 소비자가 300ms 동안 일시 중지합니다 // 대규모 소비자 지연 후 다음 요소는 즉시 사용 가능합니다: kotlin.Unit // 소비자 300ms 일시 중지 후 100ms에 다음 요소가 준비됨: kotlin.Unit