Kotlin Flow: luồng dữ liệu bất đồng bộ

1. Tổng quan

Một hàm suspend chỉ trả về một giá trị. Khi bạn cần nhận nhiều giá trị theo thời gian, ví dụ đơn hàng mới, trạng thái thanh toán hay giá cổ phiếu, bạn cần một luồng dữ liệu. Kotlin Flow giải quyết đúng việc đó.

Bài này giải thích Flow là gì và cách biến đổi dữ liệu bằng toán tử. Sau đó bạn sẽ học cách chuyển thread với flowOn, xử lý lỗi, dùng StateFlow và SharedFlow, và viết test cho Flow.

2. Cài đặt

Flow nằm trong thư viện kotlinx-coroutines. Bài viết giả định project Maven của bạn đã cấu hình sẵn kotlin-maven-plugin. Bạn chỉ cần thêm các dependency sau:

<dependencies>
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-core</artifactId>
        <version>1.10.2</version>
    </dependency>

    <!-- Test -->
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-test</artifactId>
        <version>1.10.2</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.junit.jupiter</groupId>
        <artifactId>junit-jupiter</artifactId>
        <version>5.11.4</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.assertj</groupId>
        <artifactId>assertj-core</artifactId>
        <version>3.27.3</version>
        <scope>test</scope>
    </dependency>
</dependencies>

Nếu bạn dùng Spring Boot 3.x, phiên bản coroutines đã được quản lý qua BOM, nên bạn có thể bỏ thẻ <version>.

3. Flow đầu tiên

3.1. Tạo và thu thập Flow

Cách đơn giản nhất để tạo Flow là dùng builder flow { } và gọi emit() cho từng giá trị:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

data class Order(val id: Int, val amount: Long)

fun orderStream(): Flow<Order> = flow {
    println("Bắt đầu phát đơn hàng")
    for (id in 1..3) {
        delay(100)                      // giả lập chờ dữ liệu từ hệ thống
        emit(Order(id, id * 50_000L))
    }
}

fun main() = runBlocking {
    val orders = orderStream()
    println("Đã tạo flow")
    orders.collect { println("Nhận: $it") }
}

Giải thích:

• orderStream() không phải hàm suspend. Nó chỉ trả về một mô tả về luồng dữ liệu.
• Bên trong flow { } bạn được gọi hàm suspend như delay().
• collect là terminal operator. Chỉ khi gọi nó, code trong flow { } mới chạy.

Kết quả:

Đã tạo flow
Bắt đầu phát đơn hàng
Nhận: Order(id=1, amount=50000)
Nhận: Order(id=2, amount=100000)
Nhận: Order(id=3, amount=150000)

Dòng “Đã tạo flow” xuất hiện trước “Bắt đầu phát đơn hàng”. Điều này chứng tỏ việc tạo Flow không kích hoạt gì cả.

3.2. Flow là “cold”

Flow tạo bằng flow { } là cold stream. Mỗi lần collect sẽ chạy lại toàn bộ khối code từ đầu:

fun main() = runBlocking {
    val orders = orderStream()
    orders.collect { }
    orders.collect { }
}

Bắt đầu phát đơn hàng
Bắt đầu phát đơn hàng

Lưu ý: Nếu khối flow { } gọi API hoặc ghi database, mỗi lần collect sẽ gọi lại từ đầu. Đây là nguồn gốc của nhiều bug “sao API bị gọi hai lần”. Nếu muốn chia sẻ một nguồn cho nhiều collector, hãy xem mục 6.

4. Toán tử biến đổi

4.1. Intermediate operator

Flow có bộ toán tử quen thuộc như collection: map, filter, take, onEach… Các toán tử này là intermediate. Chúng trả về Flow mới và cũng lazy giống Flow gốc.

import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map

fun main() = runBlocking {
    orderStream()
        .filter { it.amount >= 100_000 }
        .map { "Đơn #${it.id}: ${it.amount} VND" }
        .collect { println(it) }
}

Bắt đầu phát đơn hàng
Đơn #2: 100000 VND
Đơn #3: 150000 VND

Từng phần tử đi qua toàn bộ chuỗi filter → map → collect rồi phần tử kế tiếp mới được phát. Flow mặc định chạy tuần tự trong cùng một coroutine.

Khác với map của collection, lambda trong map của Flow là suspend. Vì vậy bạn có thể gọi hàm bất đồng bộ ngay bên trong:

suspend fun fetchCustomerName(orderId: Int): String {
    delay(50)
    return "Khách hàng $orderId"
}

orderStream()
    .map { fetchCustomerName(it.id) }
    .collect { println(it) }

4.2. Terminal operator khác

Ngoài collect, bạn có thể dùng các terminal operator khác để lấy kết quả trực tiếp:

import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.fold
import kotlinx.coroutines.flow.toList

fun main() = runBlocking {
    val all = orderStream().toList()
    val firstBig = orderStream().first { it.amount > 60_000 }
    val revenue = orderStream().fold(0L) { acc, order -> acc + order.amount }

    println("Số đơn: ${all.size}, đơn lớn đầu tiên: #${firstBig.id}, doanh thu: $revenue")
}

...
Số đơn: 3, đơn lớn đầu tiên: #2, doanh thu: 300000

first { } hủy Flow ngay khi tìm được phần tử phù hợp. Vì vậy đơn #3 không bao giờ được phát ra trong lần collect đó.

5. Chuyển thread với flowOn

Mặc định, code trong flow { } chạy trên context của coroutine gọi collect. Để đẩy phần phát dữ liệu sang thread khác, ví dụ đọc file hay database, bạn dùng flowOn:

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.flowOn

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("Đọc dữ liệu trên ${Thread.currentThread().name}")
    listOf("An", "Bình").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

fun main() = runBlocking {
    loadCustomers().collect {
        println("Xử lý ${it.name} trên ${Thread.currentThread().name}")
    }
}

Đọc dữ liệu trên DefaultDispatcher-worker-1
Xử lý An trên main
Xử lý Bình trên main

flowOn chỉ tác động lên phần upstream, tức các toán tử đứng trước nó. Phần collect vẫn chạy trên context của người gọi. Nguyên tắc này gọi là context preservation.

Lưu ý: Đừng dùng withContext để bọc emit() bên trong flow { }. Đoạn code dưới đây sẽ ném IllegalStateException: Flow invariant is violated:

flow {
    withContext(Dispatchers.IO) { emit(1) } // SAI
}

Hãy luôn dùng flowOn. Nếu bạn thật sự cần phát từ nhiều coroutine khác nhau, hãy dùng channelFlow { }.

6. Xử lý lỗi

6.1. catch và onCompletion

Exception trong Flow lan truyền xuống collect như code thông thường. Toán tử catch cho phép bạn chặn lỗi và phát một giá trị thay thế:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.onCompletion

fun paymentStatus(): Flow<String> = flow {
    emit("PENDING")
    throw IllegalStateException("Cổng thanh toán timeout")
}

fun main() = runBlocking {
    paymentStatus()
        .catch { e -> emit("FAILED: ${e.message}") }
        .onCompletion { cause -> println("Hoàn tất, cause=$cause") }
        .collect { println(it) }
}

PENDING
FAILED: Cổng thanh toán timeout
Hoàn tất, cause=null

Giải thích:

• catch chỉ bắt lỗi từ upstream. Exception ném ra trong lambda của collect sẽ không bị nó bắt.
• onCompletion đứng sau catch. Lỗi đã được xử lý ở bước trước, nên cause là null. Nếu đổi chỗ hai toán tử, bạn sẽ thấy cause chứa exception.

6.2. retry

Với lỗi tạm thời như mất mạng, bạn có thể dùng retry. Toán tử này collect lại Flow upstream từ đầu:

import java.io.IOException
import kotlinx.coroutines.flow.retry

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("Mất kết nối")
        emit("OK sau $attempts lần")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}

OK sau 3 lần

Predicate { e is IOException } giúp bạn chỉ thử lại với lỗi mạng. Lỗi nghiệp vụ như IllegalArgumentException sẽ không được thử lại. Trong predicate, bạn có thể gọi delay() để tạo backoff.

7. StateFlow và SharedFlow: hot stream

flow { } là cold. Nhưng nhiều khi bạn cần một nguồn dữ liệu tồn tại độc lập với collector, ví dụ trạng thái giỏ hàng hay sự kiện hệ thống. Khi đó bạn cần hot stream.

7.1. StateFlow

StateFlow luôn giữ một giá trị hiện tại. Collector mới sẽ nhận ngay giá trị đó:

import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch

class CartService {
    private val _total = MutableStateFlow(0L)
    val total: StateFlow<Long> = _total.asStateFlow()

    fun add(price: Long) = _total.update { it + price }
}

fun main() = runBlocking {
    val cart = CartService()
    val job = launch {
        cart.total.collect { println("Tổng giỏ hàng: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}

Tổng giỏ hàng: 0
Tổng giỏ hàng: 100000
Tổng giỏ hàng: 350000

Những điểm cần chú ý:

• Thuộc tính _total là private và có thể ghi. Bên ngoài chỉ thấy StateFlow chỉ đọc. Đây là pattern chuẩn để đóng gói trạng thái.
• update { } cập nhật nguyên tử (atomic), an toàn khi nhiều thread cùng gọi add.
• StateFlow không bao giờ kết thúc. Vì vậy bạn phải cancel job, nếu không runBlocking sẽ chờ mãi.
• StateFlow bỏ qua giá trị trùng với giá trị hiện tại. Nó cũng conflate: collector chậm chỉ thấy giá trị mới nhất, không thấy từng bước trung gian.

7.2. SharedFlow

SharedFlow phù hợp cho sự kiện, ví dụ “đơn hàng đã được thanh toán”. Ở đây mỗi lần phát đều có ý nghĩa và không có khái niệm “giá trị hiện tại”:

import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow

class PaymentService {
    private val _events = MutableSharedFlow<String>()
    val events: SharedFlow<String> = _events

    suspend fun pay(orderId: Int) {
        _events.emit("Đơn #$orderId đã thanh toán")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // chưa có ai lắng nghe -> mất

    val job = launch { payments.events.collect { println("Email: $it") } }
    delay(50)
    payments.pay(1)
    payments.pay(2)
    delay(50)
    job.cancel()
}

Email: Đơn #1 đã thanh toán
Email: Đơn #2 đã thanh toán

Với cấu hình mặc định (replay = 0), sự kiện phát ra khi chưa có subscriber sẽ bị bỏ. Nếu collector đến muộn cần nhận vài sự kiện gần nhất, hãy dùng MutableSharedFlow<String>(replay = 1).

Bạn cũng có thể biến một cold Flow thành hot Flow bằng stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue) hoặc shareIn(...). Khi đó nhiều collector dùng chung một lần chạy upstream.

8. Kiểm thử Flow

Thư viện kotlinx-coroutines-test cung cấp runTest. Hàm này bỏ qua delay bằng thời gian ảo, nên test chạy gần như tức thì:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test

class OrderFlowTest {

    @Test
    fun `chỉ giữ các đơn từ 100 nghìn trở lên`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

        assertThat(ids).containsExactly(2, 3)
    }

    @Test
    fun `lỗi thanh toán được chuyển thành trạng thái FAILED`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

        assertThat(states).containsExactly("PENDING", "FAILED")
    }

    @Test
    fun `giỏ hàng cộng dồn giá trị`() = runTest {
        val cart = CartService()
        cart.add(100_000)
        cart.add(250_000)

        assertThat(cart.total.value).isEqualTo(350_000L)
    }
}

Với Flow hữu hạn, toList() là cách đơn giản nhất để kiểm tra toàn bộ giá trị phát ra. Với StateFlow, bạn đọc .value trực tiếp. Đừng gọi toList() trên StateFlow, vì Flow này không bao giờ kết thúc nên test sẽ bị treo.

9. Lỗi hay gặp

• Code sau collect không chạy. collect là hàm suspend và chỉ trả về khi Flow kết thúc. Với StateFlow/SharedFlow, điều đó không bao giờ xảy ra. Hãy đặt mỗi collect vô hạn trong một launch riêng.
• Gọi API nhiều lần ngoài ý muốn. Mỗi collector của cold Flow chạy lại upstream. Hãy dùng shareIn/stateIn khi cần chia sẻ.
• Dùng withContext quanh emit. Hãy thay bằng flowOn hoặc channelFlow.
• Nghĩ rằng catch bắt mọi lỗi. Toán tử này chỉ bắt lỗi upstream. Nếu muốn bắt lỗi trong phần xử lý, hãy chuyển logic vào onEach { } đặt trước catch, rồi gọi collect() không tham số.
• Dùng StateFlow cho sự kiện. Hai sự kiện giống nhau liên tiếp sẽ bị gộp thành một. Hãy dùng SharedFlow cho sự kiện.

Lưu ý: Không nên dùng Flow khi bạn chỉ có một kết quả. Một hàm suspend fun findOrder(id: Int): Order đơn giản và dễ đọc hơn nhiều so với Flow<Order> chỉ phát một phần tử.

10. Kết luận

Trong bài này, bạn đã thấy Flow là luồng dữ liệu bất đồng bộ, cold và lazy. Bạn đã biết cách biến đổi dữ liệu bằng map/filter, chuyển thread an toàn với flowOn và xử lý lỗi bằng catch, retry, onCompletion. Với nhu cầu chia sẻ trạng thái và sự kiện, StateFlow và SharedFlow là hai công cụ chính. runTest giúp bạn kiểm thử chúng nhanh và ổn định.

Từ đây, bạn có thể thử các toán tử nâng cao như combine, flatMapLatest, debounce hay buffer để xây dựng những pipeline dữ liệu phức tạp hơn.