我正在使用Kotlin和Webflux构建一个简单的Spring服务。
我有一个返回Flow的端点。该Flow包含需要长时间计算的元素,这是通过
它的构造方式如下:
当运行时,结果与预期相符。
我有一个返回Flow的端点。该Flow包含需要长时间计算的元素,这是通过
delay
模拟的。它的构造方式如下:
suspend fun latest(): Flow<Message> {
println("generating messages")
return flow {
for (i in 0..20) {
println("generating $i")
if (i % 2 == 0) delay(1000)
else delay(200)
println("generated messsage $i")
emit(generateMessage(i))
}
println("messages generated")
}
}
我原本期望的是它会按照Message1,然后是Message3、Message5…最后是Message0的顺序返回,因为每个生成器需要不同的延迟时间。
但实际上,流程以顺序包含了这些元素。
我想我在协程和流方面错过了一些重要的东西,我尝试使用不同的协程来实现我的想法,但我无法理解如何做到。
解决方案
正如Marko Topolnik和William Reed所指出的,使用channelFlow
可以起到作用。
fun latest(): Flow<Message> {
println("generating numbers")
return channelFlow {
for (i in 0..20) {
launch {
send(generateMessage(i))
}
}
}
}
suspend fun generateMessage(i: Int): Message {
println("generating $i")
val time = measureTimeMillis {
if (i % 2 == 0) delay(1000)
else delay(500)
}
println("generated messsage $i in ${time}ms")
return Message(UUID.randomUUID(), "This is Message $i")
}
当运行时,结果与预期相符。
generating numbers
generating 2
generating 0
generating 1
generating 6
...
generated messsage 5 in 501ms
generated messsage 9 in 501ms
generated messsage 13 in 501ms
generated messsage 15 in 505ms
generated messsage 4 in 1004ms
...
Dispatchers.IO
,只需编写launch { ... }
。 - Marko Topolniklatest()
不需要成为一个挂起函数。 - Tenfour04