2023/8/18

【Kotlin/Android Studio】Flowの使い方！collectやemit関数とは？

この記事からわかること

Android Studio/KotlinのFlowの使い方
非同期処理を実装する方法
コルーチンやストリームとは？
Kotlin Coroutinesとの違い
Flow buildersの種類と作成方法
flowOfやflow { ... }関数の使い方
collectやemit関数の使い方
StateFlowの使い方

index

[open]

＼アプリをリリースしました／

友達や家族の誕生日をメモ！通知も届く-みんなの誕生日-

posted withアプリーチ

参考文献：公式リファレンス：Flow

環境

Android Studio：Koala
Kotlin：1.9.0

Flowとは?

interface Flow<out T>

AndroidのFlowとはコルーチン(非同期処理)の一種でコールドフロー特性を持ったデータストリームオブジェクトです。そもそもストリームとは日本語で「流れ」を意味する英単語であり、プログラミングの分野ではデータの入出力の流れを保持する抽象的なオブジェクトのことを指します。ストリームの中では変化するデータと「完了」または「例外」が発行されます。

データストリームなので複数の値を順番(シーケンシャル)に出力することができ、更新時にUI変更が起きやすいDB操作などに利用されることが多い印象です。また流れてくるデータに対してフィルタリングをかけたり、データを加工したりといったことも可能なため、リアルタイムでのデータの変化に対して柔軟な処理を実装することが可能となっています。

コールドストリームとホットストリームの違い

ストリームには「明示的に購読されている時のみ値が流れるCold」と「値が常に流れるHot」があります。Flowは前者のコールドストリームであり、後述するStateFlowがホットストリームになります。

コールドの特徴

購読されるまでアクティブにならない
複数回購読すると、そのたびに新しいストリームが生成
データの保持や状態管理は行わない

ホットの特徴

常にアクティブで、購読者がいなくても最新の状態を保持
新しい購読者にはその最新値が即座に渡される
初期値が必要

Kotlin Coroutinesとの使い分け

Androidでは他にも非同期処理を実装する方法としてKotlin Coroutinesなどがあります。Kotlin Coroutinesは非同期的な処理を同期的なコードのように記述できる特徴を持っています。

Flowはデータのストリーミング(データの連続生成)性を持った非同期処理を得意とするのに対し、Kotlin Coroutinesは単発的な非同期処理を得意とします。

Kotlin Coroutines

CRUD処理
スレッド管理
単発の非同期タスクの並行実行

Flow

リアルタイムなデータの変化
データのフィルタリングや加工

Kotlin Coroutinesの使用例
fun main() {
    runBlocking {
        // Deferredオブジェクトを取得
        val result1 = async { fetchDataFromRemote() }
        // 中身を取得
        val remoteData = result1.await()

        print("$remoteData")
    }
}

suspend fun fetchDataFromRemote(): String {
    delay(2000)
    return "リモートサーバーからフェッチ"
}

Flowオブジェクトの作成方法

Flowオブジェクトを作成するにはFlow builders(フロービルダー)を使用します。Flow buildersは以下のように複数用意されています。

flowOf関数：単純な値のセットからFlowを作成
asFlow関数：既存のコレクションやシーケンスなどをFlowに変換
flow { ... }関数：ブロック内でemitされたデータのFlowを作成
channelFlow { ... }関数：send関数への同時呼び出しの可能性から任意のフローを構築
MutableStateFlow/MutableSharedFlow：対応するコンストラクター関数を定義して、直接更新できるホットフローを作成

flowOf関数

公式リファレンス：flowOf関数

fun <T> flowOf(vararg elements: T): Flow<T>

flowOf関数を使用してFlowオブジェクトを生成し、データの変換の流れを取得してみたいと思います。Web実行環境で動作確認できるようにimport文も記載しています。

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val numFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)

    numFlow.collect { value ->
        print(value)
    }
}

flowOf関数は引数にデータとして流したい値を羅列して渡すだけでそのデータを流すFlowオブジェクトを作成する関数です。collect関数については後述しますが、これでデータを受け取ってる感じです。またrunBlockingに関してはKotlin Coroutinesのコードになります。

asFlow関数

asFlow関数は既存のコレクションやシーケンスなどをFlowに変換する関数です。ListやArray、Set、mapなどが変換可能です。

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
    val numList = listOf(1, 2, 3, 4, 5)
    val numFlow: <Int> = numbersList.asFlow()

    numFlow.collect { value ->
        print(value)
    }
}

flow { ... }関数

fun <T> flow(block: suspend FlowCollector<T>.() -> Unit): Flow<T>

flow { ... }関数はブロック内でemitされたデータを出力するFlowオブジェクトを作成する関数です。

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

fun generateNums(): Flow<Int> = flow {
    for (i in 1..5) {
        emit(i) 
    }
}

fun main() = runBlocking {
    val numbersFlow: Flow<Int> = generateNums()

    numbersFlow.collect { value ->
        print(value)
    }
}

collect関数

abstract suspend fun collect(collector: FlowCollector<T>)

既に何度も登場しましたがcollect関数はFlowからデータを収集するための関数です。収集されたデータはブロック内で参照することができます。またこの関数のようなFlowの収集を開始するオペレーターが呼び出されることでFlowは動き出します。

またsuspend関数として定義されているため通常の関数内からは呼び出せずコルーチン内で実装する必要があるためrunBlockingなどを使用する必要があります。

emit関数

emit関数はFlow内からデータを発行(emit)するための関数です。emit関数を使用してデータをストリームに送信します。

map関数

inline fun <T, R> Flow<T>.map(crossinline transform: suspend (value: T) -> R): Flow<R>

map関数は流れてきた値に対して任意の処理を加えて流し返す関数です。定義を見ればわかるように再度Flowオブジェクトを返します。

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.map

fun main() = runBlocking {
    val numFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)

    numFlow
      .map { value ->
              value * 2 
      }
      .collect { value ->
          print(value) // 2 4 6 8 10
      }
}

filter関数

inline fun <T> Flow<T>.filter(crossinline predicate: suspend (T) -> Boolean): Flow<T>

filter関数は流れてきた値に任意の条件でフィルタリングをかけて流し返す関数です。こちらも再度Flowオブジェクトを返します。

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.filter

fun main() = runBlocking {
    val numFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)

    numFlow
      .filter { value ->
            value % 2 == 0 
      }
      .collect { value ->
          print(value) // 2 4 
      }
}

flowOn関数

fun <T> Flow<T>.flowOn(context: CoroutineContext): Flow<T>

filter関数はFlow内でのコルーチンの実行スレッド(CoroutineContext)を変更するための関数です。こちらも再度Flowオブジェクトを返します。

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.Dispatchers


fun main() = runBlocking {
    val numFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)

    numFlow
      .filter { value ->
            value % 2 == 0 
      }
      .map { value ->
              value * 2 
      }
      .flowOn(Dispatchers.IO)
      .collect { value ->
          print(value)
      }
}

例えば上記のような箇所にflowOnを入れ込んだ場合、flowOf〜filterまでは通常のデフォルトスレッド(メイン)で実行され、map〜collectに関してはDispatchers.IOというIOスレッド上で実行されます。

StateFlow

公式リファレンス：StateFlow

interface StateFlow<out T> : SharedFlow<T>

StateFlowはCold特性を持つFlowと少し異なりHot特性を持つデータストリームオブジェクトです。要は常に最新の値を保持しており、購読されたタイミングで保持している値を流し、その後は更新された値を流すような特徴を持っています。そのためViewModelなど最新の状態を保持したい場合などに活用することが可能です。

StateFlowのvalueプロパティを変更することで保持する値が変更され外部へ変更を流すことが可能になります。

import androidx.lifecycle.ViewModel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow

class CounterViewModel : ViewModel() {
    // MutableStateFlow: カウント値を保持する
    private val _count = MutableStateFlow(0)
    // 外部には StateFlow として公開
    val count: StateFlow<Int> = _count.asStateFlow() 

    // カウントを増加させる関数
    fun increment() {
        _count.value += 1
    }

    // カウントをリセットする関数
    fun reset() {
        _count.value = 0
    }
}

まだまだ勉強中ですので間違っている点や至らぬ点がありましたら教えていただけると助かります。

ご覧いただきありがとうございました。

Android