Coroutines Flow 8편 –

by

실행 프로세스

하나의 소스에서 오는 비동기 이벤트를 표현하기 위해 Flow를 사용하는 것은 쉽습니다. 이 경우 들어오는 이벤트에 대한 응답을 코드로 등록하고 후속 작업을 계속 수행해야 합니다. addEventListener() 함수와 비슷한 역할을 하는 무언가가 필요합니다. 이 역할 각각에 운영자가 할 수 있습니다. 하지만, 각각에중간 연산자입니다. Flow를 수집하려면 터미널 운영자도 필요합니다. 그렇지 않으면 onEach만 호출해도 아무 효과가 없습니다.

onEach 이후에 수집 터미널 연산자를 사용하는 경우 다음 코드는 흐름이 수집될 때까지 기다립니다.

// Imitate a flow of events
fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }

fun main() = runBlocking<Unit> {
    events()
        .onEach { event -> println("Event: $event") }
        .collect() // <--- Collecting the flow waits
    println("Done")
}

완전한 코드는 여기에서 찾을 수 있습니다

보시다시피 출력은 다음과 같습니다.

Event: 1
Event: 2
Event: 3
Done

에서 시작 여기에서 터미널 운영자를 편리하게 사용할 수 있습니다. 모으다붓다 에서 시작로 교체하면 흐름 캡처가 별도의 코루틴에서 실행되어 후속 코드가 즉시 지속적으로 실행될 수 있습니다.

fun main() = runBlocking<Unit> {
    events()
        .onEach { event -> println("Event: $event") }
        .launchIn(this) // <--- Launching the flow in a separate coroutine
    println("Done")
}

완전한 코드는 여기에서 찾을 수 있습니다

다음과 같이 출력됩니다.

Done
Event: 1
Event: 2
Event: 3

에서 시작필요한 CoroutineScope 매개변수는 CoroutineScope를 지정하고 흐름이 실행될 때 캡처할 코루틴을 결정합니다. 위의 예에서 이 영역은 실행 차단 흐름이 실행되는 동안 코루틴 빌더에서 가져옵니다. 실행 차단 자식 코루틴이 완료될 때까지 스코프를 기다리게 하고 주로 예제가 종료되지 않도록 함수가 반환되지 않도록 합니다.

유한 수명 개체의 실제 응용 프로그램 범위 가져오다*1. 엔터티의 수명이 끝나는 순간 해당 범위가 삭제되고 해당 데이터 흐름의 수집이 중지됩니다. 이런 식으로 onEach { … }.launchIn(영역) 부부 addEventListener그러나 취소 및 구조화된 동시성 이벤트 리스너 제거 기능*2에 해당하는 역할을 수행하므로 필요하지 않습니다.

에서 시작 또한 전체 영역을 취소하거나 연결하다없이 흐름을 수집하는 코 루틴 취소작업을 수행하는 데 사용할 수 있는 작업을 반환한다는 점을 기억하세요.

흐름 취소 확인

당신의 편안함을 위해 흐름 빌더는 반환된 값을 지우는 방법을 추가로 제공합니다. 활성 상태인지 확인 확인을 수행 흐름 { … } 부지런히 방출된 것이 반복될 때 꺼질 수 있음을 의미합니다. 

fun foo(): Flow<Int> = flow { 
    for (i in 1..5) {
        println("Emitting $i") 
        emit(i) 
    }
}

fun main() = runBlocking<Unit> {
    foo().collect { value -> 
        if (value == 3) cancel()  
        println(value)
    } 
}

완전한 코드는 여기에서 찾을 수 있습니다

최대 3개를 소비하고 4개를 소비한 후 CancellationException이 발생합니다.

Emitting 1
1
Emitting 2
2
Emitting 3
3
Emitting 4
Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@6d7b4f4c

그러나 대부분의 다른 흐름 연산자는 성능상의 이유로 추가 취소 확인을 수행하지 않습니다. 예를 들어 IntRange.asFlow 확장 기능을 사용하여 동일한 사용 중인 루프를 작성하고 아무데도 멈추지 않는 경우 중단 검사가 발생하지 않습니다.

fun main() = runBlocking<Unit> {
    (1..5).asFlow().collect { value -> 
        if (value == 3) cancel()  
        println(value)
    } 
}

완전한 코드는 여기에서 찾을 수 있습니다

1에서 5까지의 모든 숫자가 캡처되고 중단은 runBlocking이 반환되기 전에만 인식됩니다.

1
2
3
4
5
Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@3327bd23

바쁜 흐름을 취소 가능하게 만들기

코루틴에 사용 중인 루프가 있는 경우 명시적으로 취소를 확인해야 합니다. .onEach {currentCoroutineContext().ensureActive() } 또한 추가 할 수 있습니다 취소 가능 이 역할에 대해 운영자가 이미 정의되었습니다.

fun main() = runBlocking<Unit> {
    (1..5).asFlow().cancellable().collect { value -> 
        if (value == 3) cancel()  
        println(value)
    } 
}

완전한 코드는 여기에서 찾을 수 있습니다

취소 가능 연산자에 대해 1에서 3까지의 숫자만 기록됩니다. 

1
2
3
Exception in thread "main" kotlinx.coroutines.JobCancellationException: BlockingCoroutine was cancelled; job="coroutine#1":BlockingCoroutine{Cancelled}@5ec0a365

다음 내용은 독자의 이해를 돕기 위해 번역자가 추가한 내용입니다.

*하나. Android Activity의 LifecycleScope 또는 ViewModel의 viewModelScope가 이 역할을 수행합니다.

*2. 코루틴은 구조적 병렬성 원칙을 따르기 때문에 범위가 중단되면 해당 범위에 속한 코루틴은 자동으로 중단됩니다. 리스너는 별도로 제거할 필요가 없습니다.

이 문서는 공식 코루틴 문서를 번역한 것입니다.

원래의: 비동기 흐름 – 흐름 시작

원본 텍스트의 최종 편집: 2022년 9월 28일

흐름 및 무효 전류

무효 전류그러나 Rxjava 또는 Project Reactor와 같은 반응형 프레임워크에 익숙한 사람들은 Flow의 디자인에 매우 익숙할 것입니다.

실제로 Flow의 디자인은 Reactive Streams와 다양한 구현에서 영감을 받았습니다. 그러나 Flow의 주요 목표는 Kotlin의 중단 친화적이고 구조화된 동시성을 존중하면서 디자인을 가능한 한 단순하게 유지하는 것입니다. Reactive Prophets와 그들의 엄청난 노력 없이는 이러한 목표를 달성할 수 없었을 것입니다. 반응형 스트림 및 Kotlin 흐름 문서에서 전체 내용을 읽을 수 있습니다.

개념적으로는 다르지만 흐름은 반응성 스트림이며 흐름은 반응성(사양 및 TCK*1 준수) 방출기로 또는 그 반대로 변환될 수 있습니다. 이러한 변환기는 기본적으로 kotlinx.coroutines 패키지에서 제공되며 다른 반응 모듈용 변환기는 여기에서 찾을 수 있습니다(Reactive Streams의 경우 kotlinx-coroutines-reactive, Project Reactor의 경우 kotlinx-coroutines-). 리액터용 coroutines-rx3 및 RxJava2/RxJava3).

다음 내용은 독자의 이해를 돕기 위해 번역자가 추가한 내용입니다.

*하나. TCK는 Technology Compatibility Kit의 약자입니다. Reactive Stream TCK는 Reactive Stream을 구현하기 위한 규칙을 정의하고 정의된 규칙에 따라 구현자가 잘 구현했는지 검증하는 데 사용됩니다.

이 문서는 공식 코루틴 문서를 번역한 것입니다.

원래의: 비동기 흐름 – 흐름 및 반응 흐름

원본 텍스트의 최종 편집: 2022년 9월 28일