flow을 사용한 비동기 작업

Flow는 Sequence 처럼 laze처리되지만 어떻게 비동기 방식이기도 할까요?

 

 예시에서는 기본 스레드나 UI 스레드와 같은 다른 스레드에 있는 관찰자와 데이터베이스 스레드 풀에서 생성된 데이터를 조정해야 합니다.

또한 데이터 변화에 따라 반복적으로 결과를 내보낼 것이므로 이 시나리오는 비동기 시퀀스 패턴에 적합합니다.

 

Flow 용 Room 통합을 작성해야한다고 생각해보자

Room에서 지원되는 suspendQuery로 시작한 경우 다음과 같이 작성할 수 있습니다.

// This code is a simplified version of how Room implements flow
fun <T> createFlow(query: Query, tables: List<Tables>): Flow<T> = flow {
    val changeTracker = tableChangeTracker(tables)

    while(true) {
        emit(suspendQuery(query))
        changeTracker.suspendUntilChanged()
    }
}

 코드는 Flow를 생성하기 위해 두 가지 가상의 suspend 함수에 의존한다.

 

suspendQuery – 일반 Room suspend 쿼리를 실행하는 주요 안전 함수

suspendUntilChanged – 테이블 중 무언가 변경 될 때까지 코루틴을 일시 중지하는 함수

 

collect되면 flow은 처음에 쿼리의 첫 번째 값을 내 보냅니다.

해당 값이 처리되면 flow이 다시 시작되고 suspendUntilChanged를 호출합니다. 테이블 중 하나가 변경 될 때까지 flow을 일시 중단합니다.  이 시점에서 테이블 중 하나가 변경되고 flow이 다시 시작될 때까지 시스템에서 아무 일도 일어나지 않습니다.

 

flow가 다시 시작되면 또 다른 주요 안전 쿼리를 만들고 결과를 내 보냅니다. 이 프로세스는 무한 루프에서 영원히 계속됩니다.

 

Flow and structured concurrency

작업을 유출하지 않아야 합니다. 코루틴은 그 자체로 비용이 많이 들지 않지만 반복적으로 작동하여 데이터베이스 쿼리를 실행합니다. 상당한 비용이 드는 유출입니다. 무한 루프를 만들었더라도 Flow에서 지원하는 구조화된 동시 실행은 유용합니다.

 

값을 소비하거나 flow을 반복할 수 있는 유일한 방법은 터미널 연산자를 사용하는 것입니다.

모든 터미널 연산자는 정지 함수이므로 작업은 이 연산자를 호출하는 scope의 전체 기간에 바인딩됩니다. scope가 취소되면 flow은 일반 코루틴 협력적 취소 규칙을 사용하여 자동으로 취소됩니다. 따라서 flow 빌더에서 무한 루프를 작성했더라도 구조화된 동시 실행으로 인해 유출 없이 안전하게 소비할 수 있습니다.