Flow — khái niệm, luồng cold và hot trong coroutine Kotlin

Flow trong Kotlin là gì?

Flow là một kiểu từ gói kotlinx.coroutines.flow, đại diện cho một luồng dữ liệu bất đồng bộ cold. Về bản chất, Flow là một chuỗi coroutine phát ra các giá trị qua hàm emit() và kết thúc thành công hoặc với một ngoại lệ. Việc thu thập luồng được thực hiện qua toán tử đầu cuối collect(), là một hàm suspend.

Ngữ nghĩa cold

Luồng cold có nghĩa là mã bên trong trình xây dựng flow được thực thi lại cho mỗi người đăng ký. Observable.fromIterable trong RxJava hoạt động tương tự: người đăng ký mới nhận được tất cả các giá trị từ đầu. Trong Flow, điều này được triển khai qua hàm suspend collect, chặn coroutine trong suốt thời gian thu thập dữ liệu.

Trình xây dựng Flow

Kotlin cung cấp một số cách để tạo Flow: flow { } — cấu trúc cơ bản với emit(), flowOf(vararg values) — cho một tập giá trị cố định, .asFlow() — một phần mở rộng cho collection và Sequence. Tất cả trình xây dựng đều cold — dữ liệu chỉ được tạo khi toán tử đầu cuối được gọi.

Luồng cold và hot

Sự phân chia giữa luồng cold và hot là một khái niệm chính của lập trình phản ứng. Luồng cold (Flow, Observable) bắt đầu tạo dữ liệu khi đăng ký. Luồng hot (Channel, SharedFlow) phát dữ liệu độc lập — người đăng ký chỉ nhận được những gì xảy ra sau khi đăng ký, mà không có phần đầu của chuỗi.

SharedFlow là một Flow hot có thể có nhiều người đăng ký và phát lại các giá trị gần đây khi replay được cấu hình. SharedFlow phù hợp cho các sự kiện (thông báo một lần). StateFlow là biến thể của nó với một giá trị trạng thái cố định, lưu trữ giá trị cuối cùng cho người đăng ký mới.

ChannelFlow sử dụng Channel bên trong, kết hợp các thuộc tính của Flow và Channel. Nó hỗ trợ đệm và backpressure qua capacity. ChannelFlow hữu ích khi chuyển đổi API callback thành luồng phản ứng, nơi các giá trị được phát ra từ các coroutine khác nhau.

Chuyển đổi giữa cold và hot

Để chuyển đổi cold Flow thành hot SharedFlow, sử dụng toán tử shareIn(scope, started, replay). Tham số started kiểm soát thời điểm bắt đầu: SharingStarted.WhileSubscribed() — hoạt động khi còn người đăng ký, Lazily — bắt đầu ở người đăng ký đầu tiên, Eagerly — bắt đầu ngay lập tức. Chuyển đổi ngược lại — hot thành cold: StateFlow.asFlow() trả về một Flow cold phát ra giá trị hiện tại của StateFlow khi collect. Điều này thuận tiện cho kiểm thử.

Toán tử Flow

Flow cung cấp một bộ toán tử phong phú hoạt động như các hàm suspend bên trong một coroutine. Các toán tử không có trạng thái và trả về một Flow mới — luồng gốc không thay đổi. Điều này cho phép xây dựng chuỗi biến đổi an toàn mà không có tác dụng phụ.

Toán tử map biến đổi mỗi giá trị luồng qua một biến đổi bất đồng bộ hoặc đồng bộ. filter chỉ cho qua các giá trị thỏa mãn điều kiện. catch bắt các ngoại lệ trước toán tử đầu cuối và cho phép khôi phục luồng. flatMapLatest hủy lần phát trước đó khi có giá trị mới đến — tương tự switchMap trong Rx.

Toán tử debounce trong Flow trì hoãn việc công bố giá trị trong một thời gian chờ xác định. Nếu có giá trị mới đến trong thời gian này, bộ đếm thời gian được đặt lại. Trong Android, debounce được sử dụng cho tìm kiếm: yêu cầu chỉ được gửi sau khi tạm dừng 300-400 ms, giảm số lượng cuộc gọi API xuống 3-5 lần.

Toán tử đầu cuối

Ngoài collect(), Flow hỗ trợ các toán tử đầu cuối khác: toList() thu thập tất cả giá trị vào một danh sách — hữu ích cho kiểm thử, first() trả về phần tử đầu tiên và hủy luồng, single() mong đợi chính xác một phần tử. fold(initial) tích lũy các giá trị qua một hàm được truyền vào. Tất cả các toán tử đầu cuối là hàm suspend và phải được gọi bên trong một coroutine hoặc một hàm suspend khác.

Ví dụ mã Flow

Ví dụ đầu tiên — Flow cơ bản tạo số với biến đổi qua toán tử map:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "Số: $it" }
        .collect { value ->
            println(value)
        }
}

Ví dụ thứ hai — biến đổi luồng với lọc và xử lý lỗi qua catch:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

Ví dụ thứ ba — sử dụng StateFlow trong ViewModel cho UI phản ứng trong Jetpack Compose:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow và SharedFlow

StateFlow là một Flow hot với một giá trị hiện tại duy nhất. Nó lưu trữ giá trị cuối cùng và chuyển ngay cho người đăng ký mới. StateFlow là một container trạng thái có thể quan sát, hỗ trợ so sánh equals — nếu giá trị mới trùng với giá trị hiện tại, không có phát nào xảy ra. Jetpack Compose sử dụng StateFlow qua collectAsState().

SharedFlow là một Flow hot linh hoạt hơn không có giá trị ban đầu bắt buộc. SharedFlow được cấu hình qua replay (số lượng giá trị cho người đăng ký mới), extraBufferCapacity (bộ đệm ngoài replay), và onBufferOverflow (chiến lược khi tràn). SharedFlow lý tưởng cho các sự kiện một lần: điều hướng, Snackbar, phân tích.

Flow trong kiến trúc Android được Google khuyến nghị làm nguồn dữ liệu chính (Lớp: Repository → UseCase → ViewModel). LiveData kém linh hoạt hơn Flow: Flow hỗ trợ coroutine, toán tử, backpressure và hoạt động bên ngoài lớp UI. Di chuyển từ LiveData sang Flow là thực tiễn tiêu chuẩn trong các dự án Android hiện đại.

Khi sử dụng Flow trong ViewModel, điều quan trọng là chọn đúng loại. StateFlow lý tưởng cho trạng thái UI cần tồn tại sau khi xoay màn hình. SharedFlow phù hợp cho các sự kiện nơi việc xử lý lại không được chấp nhận — ví dụ: điều hướng. Flow với collect() trong lifecycleScope cho kiểm soát tối đa về ngữ cảnh thực thi nhưng yêu cầu hủy thủ công khi rời khỏi màn hình.

Kiểm thử Flow được thực hiện qua kotlinx-coroutines-test. Thư viện cung cấp TestDispatcher — thời gian ảo cho phép tăng tốc độ trễ (delay) và kiểm soát thứ tự thực thi coroutine. TestScope.runTest { } tạo môi trường cách ly để kiểm thử Flow. Toán tử toList() thường được sử dụng trong kiểm thử để thu thập tất cả giá trị flow với thời gian chờ, nhằm xác minh luồng đã phát ra chuỗi dữ liệu chính xác.

Flow tích hợp tốt với Room (thư viện Android cho DB): các phương thức DAO có thể trả về Flow<List<Entity>>. Room tự động phát ra giá trị mới khi có bất kỳ thay đổi bảng nào — UI cập nhật mà không cần kích hoạt thủ công. Điều này được triển khai qua InvalidationTracker, bên trong sử dụng Flow với callbackFlow. Cách tiếp cận này loại bỏ nhu cầu về LiveData và làm cho lớp dữ liệu hoàn toàn hướng coroutine. Jetpack Compose qua collectAsState() đăng ký StateFlow và chỉ vẽ lại những thành phần có dữ liệu thay đổi — điều này mang lại hiệu suất không thể đạt được với kiến trúc hướng LiveData. DataStore (thay thế SharedPreferences) cũng trả về Flow<Preferences>, cung cấp đọc phản ứng cài đặt ứng dụng mà không cần kích hoạt cập nhật thủ công.

Flow hỗ trợ giao tiếp liên tiến trình qua kotlinx-coroutines-core trên JVM mà không cần thư viện bổ sung. Ví dụ, trong ứng dụng máy chủ trên Ktor, Flow có thể đại diện cho luồng tin nhắn WebSocket đến. Mỗi tin nhắn được phát vào luồng, trải qua lọc và tổng hợp qua các toán tử, và kết quả được gửi đến máy khách. Cách tiếp cận này thay thế các thư viện phản ứng như Reactor hoặc RxJava trong các dự án Kotlin.

Tương thích của Flow với mã RxJava hiện có được cung cấp bởi mô-đun kotlinx-coroutines-rx3. Hàm mở rộng Flow.asObservable() chuyển đổi Flow thành Observable của RxJava 3. Chuyển đổi ngược — CompletableSource.asFlow(), Observable.asFlow(). Điều này đơn giản hóa việc di chuyển từ RxJava sang coroutine: có thể viết lại dự án theo từng giai đoạn, để lại một số lớp trên RxJava. Khi chuyển đổi, cần xem xét sự khác biệt về ngữ nghĩa cold/hot: Observable có thể là cold hoặc hot, Flow luôn cold đối với Flow thông thường và hot đối với SharedFlow.

Xử lý lỗi và kiểm thử Flow

Xử lý lỗi trong Flow có một đặc điểm: nếu ngoại lệ xảy ra bên trong trình xây dựng flow trước toán tử đầu cuối, nó được truyền đến catch. Nếu ngoại lệ xảy ra trong một toán tử sau trình xây dựng, catch sau toán tử đó bắt nó. retryWhen cho phép thử lại đăng ký với điều kiện: thử lại khi lỗi mạng tối đa 3 lần, nhưng không thử lại khi CancellationException. Flow loại bỏ lỗi phụ thuộc trạng thái vì nó không lưu trữ trạng thái — điều này đơn giản hóa việc gỡ lỗi so với Observable, nơi Subject lưu trữ trạng thái nội bộ.

Kiểm thử Flow với kotlinx-coroutines-test sử dụng TestDispatcher để mô phỏng độ trễ. Turbine là thư viện cộng đồng phổ biến để kiểm thử Flow: test { } khởi chạy Flow, awaitItem() chờ giá trị tiếp theo, awaitComplete() chờ hoàn thành. Turbine thêm thời gian chờ mặc định, ngăn kiểm thử bị treo. Để kiểm thử StateFlow, sử dụng .testIn(scope) với xác minh giá trị theo thứ tự thời gian.