Kotlin协程flow瞬时密集数据流去重debounce（1）-编程实验室

Kotlin协程flow瞬时密集数据流去重debounce（1）

这个功能很像Android里面利用Handler发送一些列delay的message，然后再handleMessage里面，根据收到的前后时延是否大于某个值，如果大于等于，则处理，否则丢弃。

import kotlinx.coroutines.async import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.debounce import kotlinx.coroutines.flow.flowOn import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.receiveAsFlow import kotlinx.coroutines.newFixedThreadPoolContext import kotlinx.coroutines.runBlocking const val mTimeOut = 300L val mChannel = Channel<Int>() val mThreadPool = newFixedThreadPoolContext(nThreads = 4, name = "my-thread") fun main() { val totalTaskSize = 20 runBlocking { //接收任务 async { mChannel.receiveAsFlow() .onEach { it -> //生产者 println("onEach $it ${Thread.currentThread().name}") }.flowOn(mThreadPool) .debounce(mTimeOut) .collect { it -> //消费者 recv(it) } } //源源不断的密集发送加载任务。 async(mThreadPool) { repeat(totalTaskSize) { it -> println("send $it ${Thread.currentThread().name}") mChannel.send(it) val t = (mTimeOut * Math.random()).toLong() + 100 println("send $it over, delay=$t ${Thread.currentThread().name}") delay(t) } } } } private fun recv(n: Int) { runBlocking { async(mThreadPool) { println("collect $n ${Thread.currentThread().name}") } } }

输出：

send 0 my-thread-1
onEach 0 my-thread-1
send 0 over, delay=276 my-thread-2
send 1 my-thread-3
send 1 over, delay=184 my-thread-3
onEach 1 my-thread-1
send 2 my-thread-4
send 2 over, delay=169 my-thread-4
onEach 2 my-thread-2
send 3 my-thread-3
send 3 over, delay=175 my-thread-3
onEach 3 my-thread-1
send 4 my-thread-4
send 4 over, delay=129 my-thread-4
onEach 4 my-thread-2
send 5 my-thread-2
send 5 over, delay=367 my-thread-2
onEach 5 my-thread-1
collect 5 my-thread-3
send 6 my-thread-2
send 6 over, delay=271 my-thread-2
onEach 6 my-thread-4
send 7 my-thread-1
send 7 over, delay=200 my-thread-1
onEach 7 my-thread-3
send 8 my-thread-2
send 8 over, delay=356 my-thread-2
onEach 8 my-thread-4
collect 8 my-thread-3
send 9 my-thread-2
send 9 over, delay=222 my-thread-2
onEach 9 my-thread-1
send 10 my-thread-4
send 10 over, delay=146 my-thread-4
onEach 10 my-thread-3
send 11 my-thread-2
send 11 over, delay=215 my-thread-2
onEach 11 my-thread-1
send 12 my-thread-2
send 12 over, delay=200 my-thread-2
onEach 12 my-thread-3
send 13 my-thread-1
send 13 over, delay=298 my-thread-1
onEach 13 my-thread-4
send 14 my-thread-2
send 14 over, delay=226 my-thread-2
onEach 14 my-thread-3
collect 13 my-thread-1
send 15 my-thread-2
send 15 over, delay=337 my-thread-2
onEach 15 my-thread-3
collect 15 my-thread-1
send 16 my-thread-2
send 16 over, delay=160 my-thread-2
onEach 16 my-thread-4
send 17 my-thread-2
send 17 over, delay=147 my-thread-2
onEach 17 my-thread-1
send 18 my-thread-3
send 18 over, delay=262 my-thread-3
onEach 18 my-thread-4
send 19 my-thread-2
send 19 over, delay=163 my-thread-2
onEach 19 my-thread-1
collect 19 my-thread-4