Kotlin Flow: 非同期データストリーム

1. 概要

サスペンド関数は1つの値のみを返します。新しい注文、支払いステータス、株価など、時間とともに複数の値を受け取る必要がある場合は、データストリームが必要です。Kotlin Flowはまさくこれを解決します。

この記事では、Flowが何であるか、および演算子を使ってデータを変換する方法について説明します。その後、flowOnでスレッドを切り替える方法、エラーを処理する方法、StateFlowとSharedFlowを使う方法、およびFlowのテストを書く方法を学びます。

2. セットアップ

Flowはkotlinx-coroutinesライブラリに含まれています。この記事では、MavenプロジェクトにはすでにKotlin Maven Pluginが設定されていることを前提としています。以下の依存関係を追加するだけです:

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

Spring Boot 3.xを使用している場合、coroutinesのバージョンはBOMで管理されているため、<version>タグを省略できます。

3. 最初のFlow

3.1. Flowの作成と収集

Flowを作成する最も簡単な方法は、flow { }ビルダーを使用して、各値に対してemit()を呼び出すことです:

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("注文の発行を開始")
    for (id in 1..3) {
        delay(100)                      // システムからのデータ待機をシミュレート
        emit(Order(id, id * 50_000L))
    }
}

fun main() = runBlocking {
    val orders = orderStream()
    println("Flowを作成しました")
    orders.collect { println("受け取り: $it") }
}

説明:

• orderStream()はsuspend関数ではありません。データストリームの説明を返すだけです。
• flow { }の内部では、delay()のようなsuspend関数を呼び出すことができます。
• collectは終端演算子です。それを呼び出すときのみ、flow { }内のコードが実行されます。

結果:

Flowを作成しました
注文の発行を開始
受け取り: Order(id=1, amount=50000)
受け取り: Order(id=2, amount=100000)
受け取り: Order(id=3, amount=150000)

「Flowを作成しました」の行が「注文の発行を開始」の前に表示されます。これは、Flowの作成が何もトリガーしないことを証明しています。

3.2. Flowは「cold」です

flow { }で作成されたFlowはコールドストリームです。collectが呼ばれるたびに、コード全体が最初から実行されます:

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

注文の発行を開始
注文の発行を開始

注意: flow { }ブロック内でAPIを呼び出したりデータベースに書き込んだりする場合、collectが呼ばれるたびに最初から実行されます。これは「なぜAPIが2回呼ばれるのか」というバグの原因の1つです。複数のcollectorで1つのソースを共有したい場合は、セクション7を参照してください。

4. 変換演算子

4.1. 中間演算子

Flowには、map、filter、take、onEachなど、コレクションでよく見られる演算子があります。これらの演算子は中間です。新しいFlowを返し、元のFlowと同様にlazyです。

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

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

注文の発行を開始
注文 #2: 100000 VND
注文 #3: 150000 VND

各要素はfilter → map → collectの全チェーンを通過してから、次の要素が発行されます。Flowはデフォルトで同じcoroutine内で順序付きで実行されます。

コレクションのmapとは異なり、Flowのmapのラムダはsuspendです。したがって、その内部で非同期関数を直接呼び出すことができます:

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

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

4.2. その他の終端演算子

collectの他に、結果を直接取得するために他の終端演算子を使用できます:

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("注文数: ${all.size}, 初めての大型注文: #${firstBig.id}, 売上: $revenue")
}

...
注文数: 3, 初めての大型注文: #2, 売上: 300000

first { }は、適切な要素が見つかったらFlowをすぐにキャンセルします。したがって、その収集ではオーダー#3は決して発行されません。

5. flowOnでスレッドを切り替える

デフォルトでは、flow { }内のコードはcollectを呼び出すcoroutineのcontextで実行されます。ファイルやデータベースの読み取りなど、データ発行部分を別のスレッドにプッシュするには、flowOnを使用します:

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

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("${Thread.currentThread().name}でデータを読み込み")
    listOf("太郎", "花子").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

fun main() = runBlocking {
    loadCustomers().collect {
        println("${Thread.currentThread().name}で${it.name}を処理")
    }
}

DefaultDispatcher-worker-1でデータを読み込み
mainで太郎を処理
mainで花子を処理

flowOnはupstream部分、つまり前に来る演算子にのみ影響します。collect部分は呼び出し元のcontextで実行されたままです。この原則はcontext preservationと呼ばれます。

注意: flow { }内でemit()をラップするためにwithContextを使用しないでください。次のコードはIllegalStateException: Flow invariant is violatedをスローします:

flow {
    withContext(Dispatchers.IO) { emit(1) } // 間違い
}

常にflowOnを使用してください。複数の異なるcoroutineから発行する必要がある場合は、channelFlow { }を使用してください。

6. エラーハンドリング

6.1. catchとonCompletion

Flow内の例外は、通常のコードと同じようにcollectに伝播します。catch演算子を使用すると、エラーをキャッチして代わりの値を発行できます:

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

fun paymentStatus(): Flow<String> = flow {
    emit("保留中")
    throw IllegalStateException("決済ゲートウェイのタイムアウト")
}

fun main() = runBlocking {
    paymentStatus()
        .catch { e -> emit("失敗: ${e.message}") }
        .onCompletion { cause -> println("完了、cause=$cause") }
        .collect { println(it) }
}

保留中
失敗: 決済ゲートウェイのタイムアウト
完了、cause=null

説明:

• catchはupstreamからのエラーのみをキャッチします。collectのラムダでスローされた例外はキャッチされません。
• onCompletionはcatchの後に来ます。エラーは前のステップで処理されているため、causeはnullです。2つの演算子を入れ替えると、causeに例外が含まれます。

6.2. retry

ネットワーク断などの一時的なエラーの場合、retryを使用できます。この演算子はupstream Flowを最初から再度収集します:

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

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("接続失敗")
        emit("$attempts回後にOK")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}

3回後にOK

述語{ e is IOException }は、ネットワークエラーのみで再試行するのに役立ちます。IllegalArgumentExceptionなどのビジネスエラーは再試行されません。述語内でdelay()を呼び出してbackoffを作成できます。

7. StateFlowとSharedFlow: ホットストリーム

flow { }はcoldです。しかし、ショッピングカートの状態やシステムイベントなど、collectorから独立して存在するデータソースが必要な場合があります。その場合はホットストリームが必要です。

7.1. StateFlow

StateFlowは常に現在の値を保持しています。新しいcollectorはその値をすぐに受け取ります:

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("カート合計: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}

カート合計: 0
カート合計: 100000
カート合計: 350000

注意すべき点:

• プロパティ_totalはprivateで書き込み可能です。外部には読み取り専用のStateFlowが見えるだけです。これはステート状態をカプセル化するための標準パターンです。
• update { }はアトミックに更新され、複数のスレッドが同時にaddを呼び出す場合でも安全です。
• StateFlowは終了しません。したがって、jobをcancelする必要があります。そうしないと、runBlockingは永遠に待つでしょう。
• StateFlowは現在の値と同じ値をスキップします。また、conflateします: 遅いcollectorは最新の値のみを見え、中間のステップは見えません。

7.2. SharedFlow

SharedFlowはイベントに適しており、例えば「注文が支払われた」などです。ここでは、発行されるたびに意味を持ち、「現在の値」という概念がありません:

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("注文 #$orderId が支払われました")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // まだリスナーがいない -> 失われる

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

メール: 注文 #1 が支払われました
メール: 注文 #2 が支払われました

デフォルト設定(replay = 0)では、subscriberがない間に発行されたイベントは破棄されます。遅れてやってきたcollectorが最後のいくつかのイベントを受け取る必要がある場合は、MutableSharedFlow<String>(replay = 1)を使用してください。

また、stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue)またはshareIn(...)を使用してcold FlowをhotFlowに変えることもできます。その場合、複数のcollectorが1つのupstream実行を共有します。

8. Flowのテスト

kotlinx-coroutines-testライブラリはrunTestを提供します。この関数はdelayを仮想時間でスキップするため、テストはほぼ即座に実行されます:

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 `100000以上の注文のみを保持する`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

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

    @Test
    fun `支払いエラーはFAILED状態に変換される`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

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

    @Test
    fun `カートは値を累積する`() = runTest {
        val cart = CartService()
        cart.add(100_000)
        cart.add(250_000)

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

有限のFlowの場合、toList()は発行されるすべての値をチェックする最も簡単な方法です。StateFlowの場合は、.valueを直接読み取ります。StateFlowでtoList()を呼び出さないでください。このFlowは終了しないため、テストがハングします。

9. よくあるエラー

• collectの後のコードが実行されない。 collectはsuspend関数であり、Flowが終了したときのみ返されます。StateFlow/SharedFlowの場合、これは起こりません。各無限collectを独立したlaunchに配置します。
• 予期しないAPIの複数回呼び出し。 cold Flowの各collectorはupstreamを再実行します。共有が必要な場合はshareIn/stateInを使用してください。
• emitをwithContextでラップする。 flowOnまたはchannelFlowに置き換えてください。
• catchがすべてのエラーをキャッチすると思う。 この演算子はupstreamエラーのみをキャッチします。処理部分でエラーをキャッチしたい場合は、ロジックをcatchの前に配置されたonEach { }に移し、引数なしでcollect()を呼び出します。
• イベントにStateFlowを使用する。 連続した同じイベントは1つに統合されます。イベントにはSharedFlowを使用してください。

注意: 結果が1つだけの場合、Flowを使用しないでください。シンプルなsuspend fun findOrder(id: Int): Orderは、1つの要素のみを発行するFlow<Order>よりも簡潔で読みやすいです。

10. まとめ

この記事では、Flowが非同期データストリームであり、coldでlazyであることを知りました。map/filterを使ってデータを変換する方法、flowOnで安全にスレッドを切り替える方法、catch、retry、onCompletionでエラーを処理する方法を学びました。状態とイベントを共有する必要がある場合、StateFlowとSharedFlowは2つの主要なツールです。runTestを使用すると、すばやく安定してテストできます。

ここから、combine、flatMapLatest、debounce、bufferなどの高度な演算子を試して、より複雑なデータパイプラインを構築できます。