Kotlin Flow : flux de données asynchrone

1. Tổng quan

Une fonction suspend ne retourne qu’une seule valeur. Quand vous devez recevoir plusieurs valeurs au fil du temps, par exemple de nouvelles commandes, l’état du paiement ou le prix des actions, vous avez besoin d’un flux de données. Kotlin Flow fait exactement cela.

Cet article explique ce qu’est Flow et comment transformer les données avec des opérateurs. Ensuite vous apprendrez à commuter entre threads avec flowOn, à gérer les erreurs, à utiliser StateFlow et SharedFlow, et à écrire des tests pour Flow.

2. Cài đặt

Flow se trouve dans la bibliothèque kotlinx-coroutines. Cet article suppose que votre projet Maven a déjà configuré le plugin kotlin-maven-plugin. Vous n’avez qu’à ajouter les dépendances suivantes :

<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>

Si vous utilisez Spring Boot 3.x, la version de coroutines est déjà gérée via BOM, donc vous pouvez omettre la balise <version>.

3. Flow đầu tiên

3.1. Tạo et collecter un Flow

Le moyen le plus simple de créer un Flow est d’utiliser le builder flow { } et d’appeler emit() pour chaque valeur :

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("Commencer à émettre des commandes")
    for (id in 1..3) {
        delay(100)                      // simuler l'attente de données du système
        emit(Order(id, id * 50_000L))
    }
}

fun main() = runBlocking {
    val orders = orderStream()
    println("Flow créé")
    orders.collect { println("Reçu: $it") }
}

Explication :

• orderStream() n’est pas une fonction suspend. Elle ne retourne qu’une description du flux de données.
• À l’intérieur de flow { } vous pouvez appeler des fonctions suspend comme delay().
• collect est un opérateur terminal. Ce n’est que quand vous l’appelez que le code dans flow { } s’exécute.

Résultat :

Flow créé
Commencer à émettre des commandes
Reçu: Order(id=1, amount=50000)
Reçu: Order(id=2, amount=100000)
Reçu: Order(id=3, amount=150000)

La ligne « Flow créé » apparaît avant « Commencer à émettre des commandes ». Cela montre que créer un Flow n’active rien du tout.

3.2. Flow est « cold »

Un Flow créé avec flow { } est un cold stream. Chaque fois que vous appelez collect, tout le bloc de code est réexécuté du début :

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

Commencer à émettre des commandes
Commencer à émettre des commandes

Remarque : Si le bloc flow { } appelle une API ou écrit dans la base de données, chaque collect relancera l’appel depuis le début. C’est l’origine de nombreux bugs « pourquoi l’API est appelée deux fois ». Si vous voulez partager une source entre plusieurs collecteurs, consultez la section 6.

4. Opérateurs de transformation

4.1. Intermediate operator

Flow a un ensemble d’opérateurs familiers comme les collections : map, filter, take, onEach… Ces opérateurs sont des intermédiaires. Ils retournent un nouveau Flow et sont aussi lazy que le Flow original.

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

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

Commencer à émettre des commandes
Commande #2: 100000 VND
Commande #3: 150000 VND

Chaque élément passe par toute la chaîne filter → map → collect, puis l’élément suivant est émis. Flow s’exécute par défaut de manière séquentielle dans la même coroutine.

Contrairement à map des collections, la lambda dans map de Flow est suspend. Vous pouvez donc appeler une fonction asynchrone directement dedans :

suspend fun fetchCustomerName(orderId: Int): String {
    delay(50)
    return "Client $orderId"
}

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

4.2. Autres opérateurs terminaux

En plus de collect, vous pouvez utiliser d’autres opérateurs terminaux pour obtenir le résultat directement :

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("Nombre de commandes: ${all.size}, première grande commande: #${firstBig.id}, revenu: $revenue")
}

...
Nombre de commandes: 3, première grande commande: #2, revenu: 300000

first { } annule le Flow dès qu’il trouve un élément correspondant. Donc la commande #3 n’est jamais émise durant ce collect.

5. Commuter entre threads avec flowOn

Par défaut, le code dans flow { } s’exécute sur le contexte de la coroutine qui appelle collect. Pour déplacer la partie d’émission de données sur un autre thread, par exemple pour lire un fichier ou une base de données, vous utilisez flowOn :

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

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("Lecture des données sur ${Thread.currentThread().name}")
    listOf("An", "Binh").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

fun main() = runBlocking {
    loadCustomers().collect {
        println("Traitement de ${it.name} sur ${Thread.currentThread().name}")
    }
}

Lecture des données sur DefaultDispatcher-worker-1
Traitement de An sur main
Traitement de Binh sur main

flowOn n’affecte que la partie upstream, c’est-à-dire les opérateurs qui le précèdent. La partie collect s’exécute toujours sur le contexte de l’appelant. Ce principe s’appelle context preservation.

Remarque : N’utilisez pas withContext pour envelopper emit() à l’intérieur de flow { }. Le code ci-dessous lèvera une IllegalStateException: Flow invariant is violated :

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

Utilisez toujours flowOn. Si vous avez vraiment besoin d’émettre depuis plusieurs coroutines différentes, utilisez channelFlow { }.

6. Gestion des erreurs

6.1. catch et onCompletion

Les exceptions dans Flow se propagent vers collect comme dans le code normal. L’opérateur catch vous permet d’intercepter l’erreur et d’émettre une valeur à la place :

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

fun paymentStatus(): Flow<String> = flow {
    emit("PENDING")
    throw IllegalStateException("Passerelle de paiement timeout")
}

fun main() = runBlocking {
    paymentStatus()
        .catch { e -> emit("FAILED: ${e.message}") }
        .onCompletion { cause -> println("Terminé, cause=$cause") }
        .collect { println(it) }
}

PENDING
FAILED: Passerelle de paiement timeout
Terminé, cause=null

Explication :

• catch ne capture que les erreurs de upstream. Les exceptions levées dans la lambda de collect ne seront pas capturées.
• onCompletion vient après catch. L’erreur a déjà été traitée à l’étape précédente, donc cause est null. Si vous inversez les deux opérateurs, vous verrez cause contenant l’exception.

6.2. retry

Pour les erreurs temporaires comme la perte de connexion réseau, vous pouvez utiliser retry. Cet opérateur collecte à nouveau le Flow upstream depuis le début :

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

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("Perte de connexion")
        emit("OK après $attempts tentatives")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}

OK après 3 tentatives

Le prédicat { e is IOException } vous aide à réessayer uniquement avec les erreurs réseau. Les erreurs métier comme IllegalArgumentException ne seront pas réessayées. Dans le prédicat, vous pouvez appeler delay() pour créer un backoff.

7. StateFlow et SharedFlow : hot stream

flow { } est cold. Mais parfois vous avez besoin d’une source de données qui existe indépendamment du collecteur, par exemple l’état du panier ou les événements système. Vous avez alors besoin d’un hot stream.

7.1. StateFlow

StateFlow garde toujours une valeur courante. Un nouveau collecteur recevra immédiatement cette valeur :

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("Total du panier: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}

Total du panier: 0
Total du panier: 100000
Total du panier: 350000

Points à noter :

• L’attribut _total est private et mutable. À l’extérieur, on ne voit que le StateFlow en lecture seule. C’est le motif standard pour encapsuler l’état.
• update { } effectue une mise à jour atomique, sûre quand plusieurs threads appellent add.
• StateFlow ne se termine jamais. Vous devez donc annuler le job, sinon runBlocking attendra indéfiniment.
• StateFlow ignore les valeurs identiques à la valeur courante. Il conflate aussi : un collecteur lent ne voit que la dernière valeur, pas les étapes intermédiaires.

7.2. SharedFlow

SharedFlow convient aux événements, par exemple « la commande a été payée ». Ici, chaque émission a du sens et il n’y a pas de concept de « valeur courante » :

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("Commande #$orderId payée")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // pas encore d'écouteur -> perdu

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

E-mail: Commande #1 payée
E-mail: Commande #2 payée

Avec la configuration par défaut (replay = 0), les événements émis avant qu’il y ait un abonné sont perdus. Si un collecteur arrive tard et doit recevoir quelques événements récents, utilisez MutableSharedFlow<String>(replay = 1).

Vous pouvez aussi transformer un cold Flow en hot Flow avec stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue) ou shareIn(...). Plusieurs collecteurs partagent alors une seule exécution upstream.

8. Test de Flow

La bibliothèque kotlinx-coroutines-test fournit runTest. Cette fonction ignore les délais avec du temps virtuel, donc les tests s’exécutent presque instantanément :

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 `ne garder que les commandes à partir de 100 mille`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

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

    @Test
    fun `l'erreur de paiement se transforme en état FAILED`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

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

    @Test
    fun `le panier accumule les valeurs`() = runTest {
        val cart = CartService()
        cart.add(100_000)
        cart.add(250_000)

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

Avec un Flow fini, toList() est le moyen le plus simple de vérifier toutes les valeurs émises. Avec StateFlow, vous lisez .value directement. N’appelez pas toList() sur StateFlow, car ce Flow ne se termine jamais et le test se bloquera.

9. Erreurs courantes

• Le code après collect ne s’exécute pas. collect est une fonction suspend et ne revient que quand le Flow se termine. Avec StateFlow/SharedFlow, cela ne se produit jamais. Placez chaque collect infini dans un launch séparé.
• Appeler une API plusieurs fois par accident. Chaque collecteur d’un cold Flow réexécute upstream. Utilisez shareIn/stateIn quand vous avez besoin de partager.
• Utiliser withContext autour de emit. Remplacez-le par flowOn ou channelFlow.
• Penser que catch capture toutes les erreurs. Cet opérateur ne capture que les erreurs upstream. Si vous voulez capturer les erreurs dans la partie traitement, déplacez la logique dans onEach { } placé avant catch, puis appelez collect() sans paramètres.
• Utiliser StateFlow pour les événements. Deux événements identiques consécutifs seront fusionnés en un. Utilisez SharedFlow pour les événements.

Remarque : Vous ne devriez pas utiliser Flow quand vous n’avez qu’un seul résultat. Une simple fonction suspend fun findOrder(id: Int): Order est beaucoup plus simple et lisible qu’un Flow<Order> qui n’émet qu’un seul élément.

10. Kết luận

Dans cet article, vous avez vu que Flow est un flux de données asynchrone, cold et lazy. Vous avez appris comment transformer les données avec map/filter, commuter entre threads en toute sécurité avec flowOn et gérer les erreurs avec catch, retry, onCompletion. Pour le partage d’état et d’événements, StateFlow et SharedFlow sont les deux outils principaux. runTest vous aide à les tester rapidement et de manière stable.

À partir de là, vous pouvez essayer les opérateurs avancés comme combine, flatMapLatest, debounce ou buffer pour construire des pipelines de données plus complexes.