[XX캠퍼스] 18.Kotlin Coroutines & Flow ( 채널 파이프 라인 )

https://dalinaum.github.io/coroutines-example/18

채널 파이프라인

 

 파이프라인

파이프 라인은 일반적인 패턴입니다.

 

하나의 스트림을 프로듀서가 만들고,

다른 코루틴에서 그 스트림을 읽어 새로운 스트림을 만드는 패턴.

 

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

// 리시브 채널
fun CoroutineScope.produceNumbers() = produce<Int> {
   // CoroutineScope.XX 코루틴 블록에서 만드는거처럼 사용할 수 있다.
    var x = 1
    while (true) {
        send(x++)
    }
}

// 샌드체널
// 리시브 체널을 받으면서, 채널을 이용해서 또다른 채널을 만들어냄 
fun CoroutineScope.produceStringNumbers(numbers: ReceiveChannel<Int>): ReceiveChannel<String> = produce {
    for (i in numbers) {
        send("${i}!")
    }
}


fun main() = runBlocking<Unit> {
    val numbers = produceNumbers() // 1,2,3,4,5,6,7,8... //numbers : 리시브 채널
    val stringNumbers = produceStringNumbers(numbers)

    repeat(5) {
        // 두개의 채널이 close가 없음으로  for를 활요할 수가 없다. 
        println(stringNumbers.receive())
    }
    println("완료")
    
    //produceNumbers 와 produceStringNumbers 모두취소 
    coroutineContext.cancelChildren()
}

* 여러체널을 이용해서 데이터를 순차적으로 처리 할 수 있다. 

 

 

응용 예시 

홀수 필터

파이프라인을 응용해 홀수 필터를 만들어 봅시다.

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

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) {
        send(x++)
    }
}

fun CoroutineScope.filterOdd(numbers: ReceiveChannel<Int>): ReceiveChannel<String> = produce {
    for (i in numbers) {
        if (i % 2 == 1) {
            send("${i}!")
        }
    }
}


fun main() = runBlocking<Unit> {
    val numbers = produceNumbers()
    val oddNumbers = filterOdd(numbers)

    repeat(10) {
        println(oddNumbers.receive())
    }
    println("완료")
    coroutineContext.cancelChildren()
}

소수 필터

파이프라인을 연속으로 타면서 원하는 결과를 얻을 수 있습니다.

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

fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
    var x = start
    while (true) {
        send(x++)
    }
}

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


fun main() = runBlocking<Unit> {
    var numbers = numbersFrom(2)

    repeat(10) {
        val prime = numbers.receive()
        println(prime)
        numbers = filter(numbers, prime)
    }
    println("완료")
    coroutineContext.cancelChildren()
}

누가 이렇게 할까 싶은 예제입니다. 원한다면 디스패처를 이용해 CPU 자원을 효율적으로 이용하는 것이 가능합니다.