Android repository-pattern：RxJava 应选 Flowable 还是 Single？

2026-11-04技术老齐

既然已经确定数据库是 SSOT，建议采用接近方案 1 的设计：用 Flowable 持续观察列表，用 Completable 表达刷新、加载下一页和修改收藏，需要一次性结果时使用 Single。

Repository 可以同时提供这些类型，无须统一。具体选哪一种，取决于调用方需要持续接收数据变化，还是只需要一次操作的结果。

Flowable、Single 和 Completable 分别表达什么

类型	语义	适用场景
`Flowable<T>`	可以发出多个值，支持背压	持续观察新闻流、搜索结果、收藏列表
`Single<T>`	成功时返回一个值，或者失败	一次查询、一页网络响应、数据库快照
`Completable`	只报告完成或失败	刷新缓存、保存收藏、不需要返回值的写操作

这些定义来自 RxJava 的类型契约：Flowable、Single、Completable。

SSOT 和返回类型决定的是两件事：

SSOT 决定数据以哪里为准。
RxJava 返回类型决定调用方怎样接收数据。

Single<List<Article>> 可以从数据库读取，因此使用 Single 并不违反 SSOT。它只提供一次快照，当前订阅无法继续收到后续变化。

返回类型写成 Flowable，也不会自动获得数据库监听能力。例如，single.toFlowable() 仍然只转发那一次结果。要持续接收更新，底层需要接入可观察查询或其他变更通知机制。

推荐的接口：观察数据与更新数据分开

可以把接口中的 load 改成 observe，表明这个方法会持续提供数据，订阅期间可能收到多次更新。

import io.reactivex.Completable
import io.reactivex.Flowable

data class SearchKey(
    val query: String,
    val sourceIds: Set<String> = emptySet()
)

interface NewsFeedRepository {
    fun observeFeed(): Flowable<List<Article>>
    fun refreshFeed(): Completable
    fun loadMore(): Completable
}

interface SearchArticleRepository {
    fun observeResults(key: SearchKey): Flowable<List<Article>>
    fun refreshResults(key: SearchKey): Completable
    fun loadMore(key: SearchKey): Completable
}

interface BookmarkRepository {
    fun observeBookmarks(): Flowable<List<Article>>
    fun setBookmarked(id: String, bookmarked: Boolean): Completable
}

SearchKey 是应用内部定义的查询条件，并非 News API 的接口声明。实际项目可以加入排序、语言等影响结果的条件，同时约定空 sourceIds 的含义。

各方法的职责如下：

observe...() 从数据库提供当前列表，并持续观察变化。
refresh...() 请求远端数据，将结果写入数据库。
loadMore() 获取下一页，将文章、列表关系和分页信息写入数据库。
setBookmarked() 更新收藏状态，相关查询随后提供新的结果。

View 发出刷新操作
    → ViewModel 调用 refreshFeed()
    → Repository 请求 API
    → Repository 将结果写入数据库

数据库的可观察查询
    → Repository.observeFeed()
    → ViewModel 更新 LiveData
    → View 更新列表

Android 的数据层指南也区分一次性操作与持续观察，并要求 Repository 对外提供的数据来自其定义的 SSOT。这种接口设计将同样的职责划分用于 RxJava；官方并未要求 Repository 必须返回 Flowable。

新闻流：让数据库负责列表合并

方案 2 需要 ViewModel 判断何时清空列表、何时追加列表，还要处理去重、排序，以及刷新与翻页并发等问题。

既然数据库已经是 SSOT，可以由 Repository 在写入时处理这些规则。ViewModel 每次收到数据库提供的完整列表快照，直接替换当前展示的数据。

刷新逻辑可以这样写：

// 示意代码：remote、local 是应用自定义的数据源。
// fetchFirstPage() 返回 Single<FeedPage>。
// replaceFeedInTransaction() 是同步执行的数据库事务封装。

fun refreshFeed(): Completable =
    remote.fetchFirstPage()
        .flatMapCompletable { page ->
            Completable.fromAction {
                local.replaceFeedInTransaction(page)
            }
        }
        .subscribeOn(Schedulers.io())

这个方法应在数据库事务成功提交后才报告完成。即使网络请求成功，只要数据库写入失败，刷新就仍然失败。

刷新时不必删除所有文章。通常可以分别维护文章内容与新闻流成员关系：刷新时替换当前新闻流的关联和顺序，文章内容按标识更新。这样能保留仍被收藏或其他查询引用的文章。

Completable 完成只表示约定的写入已经完成，不保证 UI 已经收到并渲染对应的数据库结果。数据库查询和主线程分发可能稍后才执行。

搜索：按结果是否需要持续同步来选

搜索会发起 HTTP 请求，但这不足以决定 Repository 必须返回 Single。网络请求和搜索结果列表可以有不同的生命周期。

一次性搜索适合 Single

如果搜索只返回一次结果，不保存结果，也不要求随后同步数据库变化，可以使用这样的接口：

interface SearchService {
    fun searchOnce(key: SearchKey): Single<List<Article>>
}

需要分页时，应显式传递查询条件和分页参数，并返回包含分页信息的结果类型。无须为了统一接口，把这种一次性操作包装成长时间存在的 Flowable。

缓存并持续展示的搜索适合 Flowable

如果搜索结果需要缓存、分页，或者用户在详情页修改收藏后，搜索页也要更新收藏标记，可以按以下流程处理：

通过一次性网络请求获取某个查询的一页结果。
将结果及其顺序写入数据库。
通过 observeResults(key) 持续观察数据库中的查询结果。

搜索条件必须参与缓存和分页状态的标识。同一篇文章可以出现在多个搜索中，排名也可能不同，因此只保存文章内容还不够。可以分别维护文章表、查询结果关联和分页状态。

原接口中的无参 moreArticles() 隐含了“当前搜索”的概念。两个页面或两次搜索共用 Repository 时，可能因此加载错误查询的下一页。显式传入 SearchKey 可以避免这个歧义。如果新闻流也支持多个筛选条件，同样应引入对应的 key。

输入条件变化时，可以用 debounce、distinctUntilChanged 和 switchMap 切换结果订阅。不过，switchMap 不能保证撤销所有已经启动的写入。Repository 仍须按查询条件隔离数据，必要时检查请求代次，防止旧响应覆盖同一查询的新结果。

收藏：写操作通常不需要返回 Article

收藏列表适合持续观察：

fun observeBookmarks(): Flowable<List<Article>>

修改收藏通常只需要知道成功或失败：

fun setBookmarked(id: String, bookmarked: Boolean): Completable

文章对象不可变，也不要求收藏操作必须返回新的 Article。数据库更新后，可观察查询会读取新数据，再构造新的不可变对象。

如果新闻流和搜索结果也展示收藏标记，它们的数据库查询或数据组合必须依赖收藏数据。只查询文章表，再附加一个不会更新的收藏字段，无法实现跨页面同步。

刷新远端文章时，还应保留本地收藏状态。可以单独存储收藏关系，或明确限制更新字段，避免远端数据覆盖本地操作。

ViewModel 管理 UI 状态，列表由数据库更新

ViewModel 仍需管理刷新中、加载更多失败、当前搜索条件等 UI 状态。

可以把列表快照和 UI 状态放进一个不可变对象。以下示例使用 RxJava 2，省略了依赖注入和 import：

data class FeedUiState(
    val articles: List<Article> = emptyList(),
    val refreshing: Boolean = false,
    val refreshError: Throwable? = null,
    val readError: Throwable? = null
)

class FeedViewModel(
    private val repository: NewsFeedRepository
) : ViewModel() {

    private val disposables = CompositeDisposable()
    private val mutableState = MutableLiveData(FeedUiState())

    val state: LiveData<FeedUiState> = mutableState

    init {
        disposables.add(
            repository.observeFeed()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                    { articles ->
                        mutableState.value = mutableState.value!!.copy(
                            articles = articles,
                            readError = null
                        )
                    },
                    { error ->
                        mutableState.value = mutableState.value!!.copy(
                            readError = error
                        )
                    }
                )
        )
    }

    // 约定从主线程调用；Repository 自行安排网络和数据库线程。
    fun refresh() {
        val current = mutableState.value!!
        if (current.refreshing) return

        mutableState.value = current.copy(
            refreshing = true,
            refreshError = null
        )

        disposables.add(
            repository.refreshFeed()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                    {
                        mutableState.value = mutableState.value!!.copy(
                            refreshing = false
                        )
                    },
                    { error ->
                        mutableState.value = mutableState.value!!.copy(
                            refreshing = false,
                            refreshError = error
                        )
                    }
                )
        )
    }

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

这里的 articles 是数据库结果的展示快照。刷新回调只更新操作状态，列表内容始终由 observeFeed() 更新，ViewModel 无须再通过 articles.clear() 或 articles.addAll() 维护另一份文章集合。

加载更多时，可以用相同方式管理独立的 loadingMore 和错误状态。刷新失败后保留已有列表，通常也比清空列表更符合使用预期。

实现时需要明确的边界

分页并发由数据层协调。ViewModel 防止按钮重复触发，只能约束当前页面。Repository 还要处理重复翻页，以及刷新期间旧分页请求返回的问题，保持文章合并、顺序和分页状态一致。
网络错误与观察错误分开处理。刷新失败应通过刷新命令报告，并保留原有数据库订阅。观察流一旦收到 onError 就已终止，需要明确重新订阅或恢复策略。
数据库通知不对应每一次业务事件。Room 的可观察查询可能因所查询表中的其他行变化而重新执行。结果相等时可考虑使用 distinctUntilChanged()，前提是模型正确实现相等比较。参见 Room 异步查询说明。
Flowable 支持背压，但背压不会自动防止重复网络请求，也不负责分页、并发控制或数据合并规则。只需要普通的多次通知时，Observable 也可以表达持续观察。
建议让 observe...() 只观察数据，刷新单独触发。如果订阅时自动刷新，应规定缓存有效期和请求去重规则，避免多个订阅重复请求。
RxJava 2 与 RxJava 3 的包名及 Room 适配依赖不同。接口设计原则相同，具体配置应与项目版本匹配。