0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@HanaleiMoon

【Android】LiveData・RxJavaからFlowへの置き換えガイド

0
Posted at

はじめに

既存のAndroidアプリでは、ViewModelからViewへの状態通知に LiveData、非同期処理やデータ取得に RxJava を使っている、という構成をまだまだ見かけます。一方で、新規開発や Jetpack Compose を採用するプロジェクトでは、Kotlin Coroutines と Flow が標準的な選択肢になっています。

「動いているものをなぜわざわざ書き換えるのか」という疑問は当然です。置き換えにはコストがかかるため、メリットを理解しないまま着手すると、ただの書き換え作業で終わってしまいます。

本記事では、

  • なぜ LiveData / RxJava から Flow に置き換える価値があるのか
  • 置き換えると具体的に何が嬉しいのか
  • 実際のコードをどう置き換えるのか(Before / After)
  • 段階的に移行する方法と、ハマりやすい落とし穴

を、ユーザー検索画面という具体例を使って整理します。

結論を先に

LiveData RxJava Flow
位置づけ Androidのライフサイクル対応オブザーバブル リアクティブプログラミングのライブラリ Kotlin標準のコルーチンベースのストリーム
ライフサイクル対応 ◎ 自動 ✕ 自前で dispose 管理 ○ repeatOnLifecycle / collectAsStateWithLifecycle で対応
演算子の豊富さ △ map / switchMap 程度 ◎ 非常に豊富 ◎ 豊富(debounce, flatMapLatest, combine など)
非同期処理の書き方 別の仕組みが必要 subscribeOn / observeOn suspend 関数 + withContext で同期的に書ける
キャンセル 別の仕組みが必要 Disposable を管理 viewModelScope 等のスコープで自動キャンセル
Android依存 あり なし なし(Kotlin Multiplatformでも利用可)
状態の保持 LiveData 自体が保持 BehaviorSubject 等が必要 StateFlow が保持

一言でまとめると、

Flowは「LiveDataの手軽さ」と「RxJavaの表現力」を、Kotlin標準の仕組みだけで両立できる

というのが置き換える最大の理由です。

なぜ置き換える必要があるのか

LiveDataが抱える課題

LiveData はシンプルで使いやすい反面、アプリが成長するにつれて次のような限界が見えてきます。

  • 演算子が少ない: Transformations.map / switchMap と MediatorLiveData くらいしかなく、デバウンスや複数ストリームの結合といった処理は自前で書く必要がある
  • データ層まで使えない: LiveDataは androidx.lifecycle に依存するため、Android依存を避けたいRepositoryやUseCase層で使いにくい(Clean Architectureのドメイン層にAndroid依存を持ち込みたくない)
  • スレッドの制約: setValue はメインスレッド専用で、バックグラウンドからは postValue を使う必要がある。postValue は連続で呼ぶと途中の値が落ちることもある
  • 一度きりのイベントが扱いにくい: Snackbar表示や画面遷移のような「一度だけ消費したい通知」を、LiveDataで素直に書けない(そのため SingleLiveEvent のような独自クラスが生まれた)
  • Kotlin Multiplatformで使えない: Androidに閉じた仕組みのため、共通コードに持ち込めない

RxJavaが抱える課題

RxJava は非常に強力ですが、Kotlin Coroutines が普及した現在では、次のような点が負担になりがちです。

  • 学習コストが高い: Single / Maybe / Completable / Observable / Flowable の使い分けや、subscribeOn と observeOn の違い、Hot/Coldの理解など、覚える概念が多い
  • 購読の管理が必要: CompositeDisposable に追加して onCleared() で clear() する、という作法を毎回守らないとメモリリークの原因になる
  • エラーでストリームが終了する: onError が流れるとそのストリーム全体が終了するため、継続させたい場合は各所で onErrorReturn などを書く必要がある
  • スタックトレースが読みにくい: 演算子のチェーンの中でエラーが起きると、原因箇所の特定が難しい
  • ライブラリのサポートが Coroutines 中心に: Room、Retrofit、DataStore、Paging、WorkManager などの Jetpack ライブラリは、suspend 関数や Flow を第一級でサポートしており、新機能も Coroutines 向けが先行する

Flowに置き換えるメリット

1. 非同期処理を同期的なコードのように書ける

RxJavaでは subscribeOn / observeOn でスレッドを切り替えますが、Coroutines では suspend 関数を普通に呼ぶだけです。

// RxJava
repository.searchUsers(query)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ users -> /* ... */ }, { error -> /* ... */ })

// Coroutines
viewModelScope.launch {
    try {
        val users = repository.searchUsers(query)
        // ...
    } catch (e: Exception) {
        // ...
    }
}

2. キャンセルが構造化されている

viewModelScope の中で起動した処理は、ViewModelが破棄されると自動でキャンセルされます。CompositeDisposable を管理する必要がなくなり、購読の解除し忘れによるリークを構造的に防げます。

3. LiveData相当の機能が StateFlow で手に入る

StateFlow は「常に最新の値を1つ保持する」ストリームで、LiveDataと同じ用途で使えます。さらに初期値が必須なので、null チェックから解放されます。

4. RxJava並みの演算子が使える

debounce、distinctUntilChanged、flatMapLatest(RxJavaの switchMap 相当)、combine、zip など、検索画面や複数データの結合といった典型的な処理を簡潔に書けます。

5. Android依存がなくなる

Flowは Kotlin標準の仕組みなので、Repository / UseCase層でも Android に依存せずに使えます。Kotlin Multiplatform への展開もしやすくなります。

6. Jetpack Composeと相性がよい

collectAsStateWithLifecycle() で Flow をそのまま Compose の State に変換できます。LiveDataでも observeAsState() で扱えますが、StateFlow の方が Compose の「状態から UI を宣言する」思想と自然に噛み合います。

置き換えなくてもよいケース

メリットばかりを並べましたが、すべてを無理に置き換える必要はありません。

  • 安定稼働していて、今後ほとんど触らない画面・機能は、そのままでも問題ない
  • Javaで書かれたコードが多い場合、Coroutines/Flow は Kotlin 前提なので移行の恩恵が小さい
  • RxJavaの高度な演算子や Backpressure の制御に強く依存している処理は、無理に移さず RxJava のまま残す選択肢もある

現実的には、新規画面・改修頻度の高い画面から順に Flow に寄せていくのがおすすめです。後述するように、RxJavaと Flow は相互変換できるため、段階的な移行が可能です。

置き換え前の状態: RxJava + LiveData のユーザー検索画面

題材として、「検索キーワードを入力するとユーザー一覧を取得して表示する」画面を使います。入力中の無駄なAPI呼び出しを避けるため、デバウンスと「最新の入力だけ結果を採用する」処理を入れています。

Repository(RxJava)

interface UserRepository {
    fun searchUsers(query: String): Single<List<User>>
}

class UserRepositoryImpl(private val api: UserApi) : UserRepository {
    override fun searchUsers(query: String): Single<List<User>> =
        api.searchUsers(query) // Retrofit + RxJava3CallAdapter で Single を返す
}

ViewModel(RxJava + LiveData)

class UserSearchViewModel(
    private val repository: UserRepository
) : ViewModel() {

    private val disposables = CompositeDisposable()
    private val querySubject = PublishSubject.create<String>()

    private val _users = MutableLiveData<List<User>>(emptyList())
    val users: LiveData<List<User>> = _users

    private val _isLoading = MutableLiveData(false)
    val isLoading: LiveData<Boolean> = _isLoading

    // 一度きりのイベント(Snackbar表示用)
    private val _errorEvent = SingleLiveEvent<String>()
    val errorEvent: LiveData<String> = _errorEvent

    init {
        disposables.add(
            querySubject
                .debounce(300, TimeUnit.MILLISECONDS)
                .distinctUntilChanged()
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext { _isLoading.value = true }
                .switchMapSingle { query ->
                    repository.searchUsers(query)
                        .subscribeOn(Schedulers.io())
                }
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                    { users ->
                        _isLoading.value = false
                        _users.value = users
                    },
                    { error ->
                        // ここに来るとストリームが終了し、以降の検索が反応しなくなる
                        _isLoading.value = false
                        _errorEvent.value = "検索に失敗しました"
                    }
                )
        )
    }

    fun onQueryChange(query: String) {
        querySubject.onNext(query)
    }

    override fun onCleared() {
        disposables.clear()
    }
}

Fragment(LiveDataをobserve)

viewModel.users.observe(viewLifecycleOwner) { users ->
    adapter.submitList(users)
}
viewModel.isLoading.observe(viewLifecycleOwner) { isLoading ->
    binding.progressBar.isVisible = isLoading
}
viewModel.errorEvent.observe(viewLifecycleOwner) { message ->
    Snackbar.make(binding.root, message, Snackbar.LENGTH_SHORT).show()
}

このコードには、これまで挙げた課題が凝縮されています。

  • CompositeDisposable の管理と onCleared() での clear() が必要
  • observeOn / subscribeOn の位置を間違えると、メインスレッド以外から setValue を呼んでクラッシュする
  • onError に到達すると、ストリーム全体が終了して以降の検索が一切反応しなくなる(一度でもAPIエラーが起きるとその画面の検索が壊れる)
  • 一度きりのイベントのために SingleLiveEvent という独自クラスが必要

置き換え後: Flow + StateFlow

Repository(suspend関数)

interface UserRepository {
    suspend fun searchUsers(query: String): List<User>
}

class UserRepositoryImpl(private val api: UserApi) : UserRepository {
    // Retrofit の suspend 関数はメインスレッドから呼んでも安全(main-safe)
    override suspend fun searchUsers(query: String): List<User> =
        api.searchUsers(query)
}

Single<List<User>> が suspend fun ...: List<User> というシンプルな形になりました。1回きりの結果を返すだけの処理は、Flow ではなく suspend 関数で表現するのが基本です。

ViewModel(StateFlow + Channel)

sealed interface UserSearchEvent {
    data class ShowError(val message: String) : UserSearchEvent
}

@OptIn(FlowPreview::class, ExperimentalCoroutinesApi::class)
class UserSearchViewModel(
    private val repository: UserRepository
) : ViewModel() {

    private val _query = MutableStateFlow("")

    private val _isLoading = MutableStateFlow(false)
    val isLoading: StateFlow<Boolean> = _isLoading.asStateFlow()

    // 一度きりのイベントは Channel + receiveAsFlow で表現する
    private val _events = Channel<UserSearchEvent>(Channel.BUFFERED)
    val events: Flow<UserSearchEvent> = _events.receiveAsFlow()

    val users: StateFlow<List<User>> = _query
        .debounce(300)
        .distinctUntilChanged()
        .flatMapLatest { query ->
            flow { emit(repository.searchUsers(query)) }
                .onStart { _isLoading.value = true }
                .catch {
                    // 内側のFlowでエラーを処理するので、外側のストリームは生き続ける
                    _events.send(UserSearchEvent.ShowError("検索に失敗しました"))
                    emit(emptyList())
                }
                .onCompletion { _isLoading.value = false }
        }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5_000),
            initialValue = emptyList()
        )

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

RxJava版と見比べると、次の点が変わっています。

  • CompositeDisposable と onCleared() が不要になった(viewModelScope が自動でキャンセルしてくれる)
  • subscribeOn / observeOn が消えた(スレッドの切り替えを意識しなくてよい)
  • switchMapSingle が flatMapLatest に置き換わった(新しい入力が来ると前回の検索を自動キャンセルする点は同じ)
  • エラー処理を内側の flow { } に閉じ込めたことで、APIエラーが起きても外側のストリームは終了せず、次の検索がちゃんと動く
  • SingleLiveEvent が Channel + receiveAsFlow() に置き換わった(独自クラス不要)

SharingStarted.WhileSubscribed(5_000) は、「購読者がいなくなってから5秒間は上流を維持し、それ以降は停止する」という指定です。画面回転のような一瞬の購読解除では再取得せず、本当にバックグラウンドに回ったときだけ上流を止められます。

Fragment(repeatOnLifecycleでcollect)

viewLifecycleOwner.lifecycleScope.launch {
    viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
        launch {
            viewModel.users.collect { users -> adapter.submitList(users) }
        }
        launch {
            viewModel.isLoading.collect { isLoading ->
                binding.progressBar.isVisible = isLoading
            }
        }
        launch {
            viewModel.events.collect { event ->
                when (event) {
                    is UserSearchEvent.ShowError ->
                        Snackbar.make(binding.root, event.message, Snackbar.LENGTH_SHORT).show()
                }
            }
        }
    }
}

repeatOnLifecycle(STARTED) を使うことで、LiveDataの observe と同じく「画面が見えている間だけ購読する」挙動になります。

Composeの場合

Jetpack Compose を使っている場合は、さらにシンプルになります。

@Composable
fun UserSearchScreen(viewModel: UserSearchViewModel) {
    val users by viewModel.users.collectAsStateWithLifecycle()
    val isLoading by viewModel.isLoading.collectAsStateWithLifecycle()

    // ...
}

collectAsStateWithLifecycle() がライフサイクルを考慮した購読を行うため、repeatOnLifecycle を自分で書く必要もありません。

LiveData / RxJava → Flow の置き換え対応表

置き換え前 置き換え後
MutableLiveData<T>(初期値) MutableStateFlow<T>(初期値)
LiveData<T> の公開 StateFlow<T>(asStateFlow())
liveData.value = x stateFlow.value = x
liveData.postValue(x) stateFlow.value = x(スレッドを問わず設定できる)
Transformations.map Flow.map
Transformations.switchMap Flow.flatMapLatest
MediatorLiveData combine / merge
SingleLiveEvent Channel + receiveAsFlow()
liveData.observe(owner) { } repeatOnLifecycle + collect / collectAsStateWithLifecycle()
Single<T> / Maybe<T> suspend fun ...: T(T?)
Completable suspend fun ...: Unit
Observable<T> / Flowable<T> Flow<T>
BehaviorSubject<T> MutableStateFlow<T>
PublishSubject<T> MutableSharedFlow<T>
subscribeOn(Schedulers.io()) flowOn(Dispatchers.IO) / withContext(Dispatchers.IO)
observeOn(AndroidSchedulers.mainThread()) 不要(viewModelScope はMainで動く)
debounce / distinctUntilChanged 同名の演算子がそのまま存在
switchMap flatMapLatest
CompositeDisposable 不要(viewModelScope が管理)
onErrorReturn / onErrorResumeNext catch { emit(...) } / try-catch

段階的に移行する方法

既存のプロジェクトで全部を一気に置き換えるのは現実的ではありません。RxJava / LiveData と Flow は相互に変換できるので、データ層から内側→外側へ、あるいは画面単位で少しずつ進められます。

RxJava ↔ Flow(kotlinx-coroutines-rx3)

// build.gradle.kts
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-rx3:<coroutinesのバージョン>")

// RxJava → Coroutines
val users: List<User> = rxRepository.searchUsers(query).await()     // Single -> suspend
val usersFlow: Flow<List<User>> = rxRepository.observeUsers().asFlow() // Observable -> Flow

// Coroutines → RxJava
val observable: Observable<List<User>> = usersFlow.asObservable()

これを使うと、まず ViewModelだけを先に Flow 化し、Repository は RxJava のまま await() / asFlow() で橋渡しする、あるいはその逆といった中間状態を作れます。

LiveData ↔ Flow(lifecycle-livedata)

// LiveData → Flow
val flow: Flow<List<User>> = liveData.asFlow()

// Flow → LiveData
val liveData: LiveData<List<User>> = flow.asLiveData()

なお、asFlow() / asLiveData() は androidx.lifecycle:lifecycle-livedata-ktx(Lifecycle 2.6以降は lifecycle-livedata に統合)で提供されています。ViewModelはFlow化したいが、Fragment側の observe はまだ書き換えたくない、という場合に asLiveData() で橋渡しできます。

おすすめの移行ステップ

  1. Repositoryの Single を suspend 関数に置き換える(まずは await() で橋渡し)
  2. ViewModelの MutableLiveData を MutableStateFlow に置き換える(Viewとの境界は asLiveData() で一時的に橋渡し)
  3. Fragment / Compose側を collect や collectAsStateWithLifecycle() に置き換える
  4. 橋渡しのコードと不要になった依存(RxJava、SingleLiveEvent など)を削除する

DataBindingを使っているプロジェクトでも、AGP 7.0以降であれば StateFlow を直接バインドできるため、lifecycleOwner を設定していれば、XML側の LiveData を StateFlow に差し替えるだけで済むケースもあります。

テストのしやすさも変わる

LiveDataをテストする際は InstantTaskExecutorRule が、RxJavaをテストする際は RxJavaPlugins でSchedulerを差し替える設定が必要でした。Flow(Coroutines)では、kotlinx-coroutines-test の仮想時間を使うことで、デバウンスの待ち時間もテスト内で一瞬で進められます。

@Test
fun `検索クエリ入力後にユーザー一覧が更新される`() = runTest {
    // MainDispatcherRule で Dispatchers.Main をテスト用に差し替え済みとする
    val viewModel = UserSearchViewModel(
        FakeUserRepository(users = listOf(User(id = "1", name = "Alice")))
    )
    // WhileSubscribed の場合、購読者がいないと上流が動かないため collect を開始しておく
    backgroundScope.launch(UnconfinedTestDispatcher(testScheduler)) {
        viewModel.users.collect()
    }

    viewModel.onQueryChange("Ali")
    advanceTimeBy(301) // debounce(300) の待ち時間を仮想時間で進める
    runCurrent()

    assertEquals(listOf(User(id = "1", name = "Alice")), viewModel.users.value)
}

ありがちなアンチパターン

アンチパターン1: ライフサイクルを考慮せずにcollectする

// Fragment
override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
    viewLifecycleOwner.lifecycleScope.launch {
        // バックグラウンドに回っても collect が止まらない
        viewModel.users.collect { users -> adapter.submitList(users) }
    }
}

LiveDataの observe は自動でライフサイクルを考慮してくれましたが、単に lifecycleScope.launch { flow.collect { } } と書くと、画面が見えていない間も購読が続いてしまいます。無駄な処理やリソース消費の原因になります。

解決策: repeatOnLifecycle で囲む(Composeなら collectAsStateWithLifecycle())

override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
    viewLifecycleOwner.lifecycleScope.launch {
        viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
            viewModel.users.collect { users -> adapter.submitList(users) }
        }
    }
}

アンチパターン2: StateFlowの中身を直接書き換える

data class SearchState(val users: MutableList<User> = mutableListOf())

private val _state = MutableStateFlow(SearchState())

fun onUserLoaded(user: User) {
    // 同じインスタンスのままリストを変更しているので、StateFlowは変化を検知しない
    _state.value.users.add(user)
    _state.value = _state.value
}

StateFlow は 新しい値が直前の値と equals で等しい場合は通知しません。LiveDataは同じ値を setValue しても通知されましたが、StateFlowは通知されない点が大きな違いです。可変オブジェクトを書き換えても、同一インスタンスなので「変化なし」と判断されてしまいます。

解決策: Immutableなdata classを使い、copy で新しい値を作る

data class SearchState(val users: List<User> = emptyList())

private val _state = MutableStateFlow(SearchState())

fun onUserLoaded(user: User) {
    _state.update { it.copy(users = it.users + user) }
}

アンチパターン3: 一度きりのイベントをStateFlowで流す

private val _errorMessage = MutableStateFlow<String?>(null)
val errorMessage: StateFlow<String?> = _errorMessage.asStateFlow()

// Fragment
viewLifecycleOwner.lifecycleScope.launch {
    viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.errorMessage.collect { message ->
            // 画面回転のたびに、同じSnackbarが再表示されてしまう
            message?.let { showSnackbar(it) }
        }
    }
}

StateFlow は常に最新の値を保持し、新しい購読者にもその値を即座に流します。そのため、Snackbarや画面遷移のような「一度だけ消費したいイベント」をStateFlowで表現すると、画面回転や再購読のたびに同じイベントが再実行されます(LiveDataで SingleLiveEvent が必要だった理由と同じです)。

解決策: Channel + receiveAsFlow() でイベントを流す

private val _events = Channel<UserSearchEvent>(Channel.BUFFERED)
val events: Flow<UserSearchEvent> = _events.receiveAsFlow()

fun onError() {
    viewModelScope.launch {
        _events.send(UserSearchEvent.ShowError("検索に失敗しました"))
    }
}

Channelは1つの値が1回だけ消費されるため、再購読しても同じイベントが再送されません。

まとめ

  • LiveDataは演算子の少なさ・Android依存・イベントの扱いにくさ、RxJavaは学習コスト・購読管理・エラーによるストリーム終了といった課題を抱えており、Flowはこれらを Kotlin 標準の仕組みだけで解決できる
  • 1回きりの非同期処理は suspend 関数、継続的な値の流れは Flow、画面の状態は StateFlow、一度きりのイベントは Channel + receiveAsFlow() と使い分ける
  • debounce / flatMapLatest / combine など RxJava に近い演算子が揃っており、viewModelScope によって Disposable の管理が不要になる
  • await() / asFlow() / asLiveData() で相互変換できるため、Repository → ViewModel → View の順に段階的に移行できる
  • 置き換え後は repeatOnLifecycle / collectAsStateWithLifecycle() でのライフサイクル対応、StateFlow の重複排除(equals 比較)といった LiveData との挙動の違いに注意する

すべてを一度に書き換える必要はありません。新規画面や改修頻度の高い画面から、少しずつ Flow に寄せていくのが現実的な進め方です。

0
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?