从RxJava转Kotlin Flow：是否有类似RxJava GroupBy的操作符？

嘿，好问题！从 RxJava 迁移到 Kotlin Flow 时，groupBy 确实是很多人会关心的操作符——Flow 里同样有同名的 groupBy 操作符，功能和 RxJava 的版本对齐，只是在流的冷热特性上有一点差异，我来给你讲明白怎么用：

基本用法：和 RxJava 几乎一致 #

Flow 的 groupBy 会把上游流中的元素按照你指定的键进行分组，返回一个 Flow<GroupedFlow<K, T>>，其中每个 GroupedFlow 包含分组的键（key）和对应分组内的元素流。

举个实际的代码例子：

// 定义一个数据类用于演示
data class User(val id: Int, val name: String, val department: String)

// 模拟一个用户数据流
val usersFlow = flowOf(
    User(1, "Alice", "Engineering"),
    User(2, "Bob", "Marketing"),
    User(3, "Charlie", "Engineering"),
    User(4, "Diana", "Marketing")
)

// 使用 groupBy 按部门分组
usersFlow.groupBy { it.department }
    .collect { groupedFlow ->
        println("\n=== 处理部门：${groupedFlow.key} ===")
        // 收集当前分组内的所有用户
        groupedFlow.collect { user ->
            println("用户：${user.name}")
        }
    }

运行这段代码会输出：

=== 处理部门：Engineering ===
用户：Alice
用户：Charlie

=== 处理部门：Marketing ===
用户：Bob
用户：Diana

和 RxJava GroupBy 的核心差异 #

这里要注意一个关键区别：

• RxJava 的 GroupedObservable 是热流，一旦分组创建，即使没有订阅，元素也会被缓存或丢弃（取决于配置）。
• Flow 的 GroupedFlow 是冷流——只有当你订阅（collect）某个 GroupedFlow 时，它才会从上游拉取对应分组的元素。如果多次订阅同一个 GroupedFlow，上游会重新发射对应元素。

如果需要模拟 RxJava 热流的行为，可以给每个分组流加上 shareIn 操作符，把它变成热流：

usersFlow.groupBy { it.department }
    .map { groupedFlow ->
        // 把分组流转为热流，共享给多个订阅者
        groupedFlow.shareIn(
            scope = CoroutineScope(Dispatchers.Default),
            started = SharingStarted.WhileSubscribed()
        ) to groupedFlow.key
    }

进阶：分组后做聚合操作 #

和 RxJava 一样，你可以对每个分组的流做进一步的操作，比如统计数量、求和、过滤等。用 flatMapMerge 或 flatMapConcat 可以很方便地处理分组后的流：

比如统计每个部门的用户数量：

usersFlow.groupBy { it.department }
    .flatMapMerge { groupedFlow ->
        groupedFlow.count()
            .map { count -> "部门 ${groupedFlow.key} 共有 $count 个用户" }
    }
    .collect { println(it) }

输出结果：

部门 Engineering 共有 2 个用户
部门 Marketing 共有 2 个用户

总的来说，Flow 的 groupBy 完全可以替代 RxJava 的同名操作符，只是需要注意它的冷流特性，根据场景调整即可。

内容的提问来源于stack exchange，提问作者Sundaravel