Android中如何按数量或时间条件从Kotlin Flow收集数据后调用API？

核心需求 #

• 从Int类型Flow收集数据，满足以下任一条件时触发API调用：
  • 已收集满5条数据
  • 距离上次API调用已过去3秒（即使数据不满5条）
• 给定Flow发射序列需得到预期的三次API调用结果
• 收集过程随Activity销毁自动终止

实现方案 #

使用lifecycleScope管理收集协程（自动绑定Activity生命周期），结合本地缓存列表+计时器的方式实现批量触发逻辑，同时满足数量阈值与时间阈值的要求。

完整代码示例 #

import androidx.appcompat.app.AppCompatActivity
import android.os.Bundle
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch

class MainActivity : AppCompatActivity() {
    // 定义题目中的原始Flow
    private val intFlow: Flow<Int> = flow {
        emit(1)
        emit(2)
        emit(3)
        emit(4)
        emit(5)
        emit(6)
        emit(7)
        emit(8)
        delay(4000)
        emit(9)
        emit(10)
        emit(11)
        emit(12)
        emit(13)
    }

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        // 启动收集逻辑，绑定Activity生命周期
        lifecycleScope.launch {
            val batch = mutableListOf<Int>()
            var timerJob: Job? = null

            // 封装批次发送与计时器重置逻辑
            fun flushBatch() {
                if (batch.isNotEmpty()) {
                    // 此处替换为实际API调用逻辑
                    println("call an api with value [${batch.joinToString(",")}]....")
                    batch.clear()
                }
                // 取消旧计时器，重新启动新的3秒倒计时
                timerJob?.cancel()
                timerJob = launch {
                    delay(3000)
                    flushBatch()
                }
            }

            // 收集Flow数据
            intFlow.collect { value ->
                batch.add(value)
                when {
                    batch.size == 5 -> flushBatch() // 满5条立即发送
                    timerJob?.isActive != true -> { // 未启动计时器则启动
                        timerJob = launch {
                            delay(3000)
                            flushBatch()
                        }
                    }
                }
            }

            // Flow发射完成后，发送剩余未处理的批次
            flushBatch()
            timerJob?.cancel()
        }
    }
}

关键逻辑说明 #

1. 生命周期绑定：使用lifecycleScope启动协程，Activity销毁时协程自动取消，无需手动管理收集终止逻辑。
2. 批次缓存：用mutableList<Int>缓存待发送的数据，避免重复创建集合。
3. 双阈值触发：
   • 数量阈值：每添加一条数据就检查是否达到5条，满额则立即发送批次并重置计时器。
   • 时间阈值：每次添加数据或发送批次后，启动/重置3秒倒计时，超时则发送当前缓存的所有数据。
4. Flow收尾处理：Flow发射完成后主动调用flushBatch()，确保剩余未发送的数据被处理。

结果验证 #

运行代码后会输出预期结果：

call an api with value [1,2,3,4,5]....
call an api with value [6,7,8]....
call an api with value [9,10,11,12,13]....

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