19. 处理最新值

当发射器和收集器都很慢的时候，合并是加快处理速度的一种方式。它通过删除发射值来实现。 另一种方式是取消缓慢的收集器，并在每次发射新值的时候重新启动它。有一组与 xxx 操作符执行相同基本逻辑的 xxxLatest 操作符，但是在新值产生的时候取消执行其块中的代码。让我们在先前的示例中尝试更换 [conflate]

val time = measureTimeMillis {
    simple()
        .collectLatest { value -> // 取消并重新发射最后一个值
            println("Collecting $value") 
            delay(300) // 假装我们花费 300 毫秒来处理它
            println("Done $value") 
        } 
}   
println("Collected in $time ms")

Collecting 1
Collecting 2
Collecting 3
Done 3
Collected in 680 ms

20. 组合多个流

Zip

就像 Kotlin 标准库中的 Sequence.zip 扩展函数一样， 流拥有一个 zip 操作符用于组合两个流中的相关值：

val nums = (1..3).asFlow() // 数字 1..3
val strs = flowOf("one", "two", "three") // 字符串
nums.zip(strs) { a, b -> "$a -> $b" } // 组合单个字符串
    .collect { println(it) } // 收集并打印
1 -> one
2 -> two
3 -> three

Combine

当流表示一个变量或操作的最新值时，可能需要执行计算，这依赖于相应流的最新值，并且每当上游流产生值的时候都需要重新计算。这种相应的操作符家族称为 [combine]

例如，先前示例中的数字如果每 300 毫秒更新一次，但字符串每 400 毫秒更新一次， 然后使用 操作符合并它们，但仍会产生相同的结果， 尽管每 400 毫秒打印一次结果：

val nums = (1..3).asFlow().onEach { delay(300) } // 发射数字 1..3，间隔 300 毫秒
val strs = flowOf("one", "two", "three").onEach { delay(400) } // 每 400 毫秒发射一次字符串
val startTime = System.currentTimeMillis() // 记录开始的时间
nums.zip(strs) { a, b -> "$a -> $b" } // 使用“zip”组合单个字符串
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    } 

1 -> one at 427 ms from start
2 -> two at 827 ms from start
3 -> three at 1229 ms from start

然而，当在这里使用 [combine]操作符来替换 [zip]

val nums = (1..3).asFlow().onEach { delay(300) } // 发射数字 1..3，间隔 300 毫秒
val strs = flowOf("one", "two", "three").onEach { delay(400) } // 每 400 毫秒发射一次字符串
val startTime = System.currentTimeMillis() // 记录开始的时间
nums.combine(strs) { a, b -> "$a -> $b" } // 使用“combine”组合单个字符串
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 

}
1 -> one at 452 ms from start
2 -> one at 651 ms from start
2 -> two at 854 ms from start
3 -> two at 952 ms from start
3 -> three at 1256 ms from start

21. 展平流

流表示异步接收的值序列，所以很容易遇到这样的情况： 每个值都会触发对另一个值序列的请求。比如说，我们可以拥有下面这样一个返回间隔 500 毫秒的两个字符串流的函数

fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First") 
    delay(500) // 等待 500 毫秒
    emit("$i: Second")    
}

现在，如果我们有一个包含三个整数的流，并为每个整数调用 requestFlow，如下所示：

(1..3).asFlow().map { requestFlow(it) }

然后我们得到了一个包含流的流（Flow<Flow<String>>），需要将其进行展平为单个流以进行下一步处理。集合与序列都拥有 [flatten]与 [flatMap] 操作符来做这件事。然而，由于流具有异步的性质，因此需要不同的展平模式， 为此，存在一系列的流展平操作符。
flatMapConcat

连接模式由 [flatMapConcat]与 [flattenConcat]操作符实现。它们是相应序列操作符最相近的类似物。它们在等待内部流完成之前开始收集下一个值，如下面的示例所示：

al startTime = System.currentTimeMillis() // 记录开始时间
(1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字 
    .flatMapConcat { requestFlow(it) }                                                                           
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    } 

1: First at 121 ms from start
1: Second at 622 ms from start
2: First at 727 ms from start
2: Second at 1227 ms from start
3: First at 1328 ms from start
3: Second at 1829 ms from start

在输出中可以清楚地看到 [flatMapConcat] 的顺序性质：

flatMapMerge

另一种展平模式是并发收集所有传入的流，并将它们的值合并到一个单独的流，以便尽快的发射值。 它由 [flatMapMerge] 与 [flattenMerge]操作符实现。他们都接收可选的用于限制并发收集的流的个数的 concurrency 参数（默认情况下，它等于 [DEFAULT_CONCURRENCY]。

val startTime = System.currentTimeMillis() // 记录开始时间
(1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字 
    .flatMapMerge { requestFlow(it) }                                                                           
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    } 

1: First at 150 ms from start
2: First at 246 ms from start
3: First at 347 ms from start
1: Second at 651 ms from start
2: Second at 747 ms from start
3: Second at 849 ms from start

[flatMapMerge]的并发性质很明显：

flatMapLatest

与 collectLatest 操作符类似（在"处理最新值" 小节中已经讨论过），也有相对应的“最新”展平模式，在发出新流后立即取消先前流的收集。 这由 flatMapLatest 操作符来实现。

val startTime = System.currentTimeMillis() // 记录开始时间
(1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字 
    .flatMapLatest { requestFlow(it) }                                                                           
    .collect { value -> // 收集并打印
        println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
    } 

1: First at 225 ms from start
2: First at 357 ms from start
3: First at 458 ms from start
3: Second at 959 ms from start

该示例的输出很好的展示了 [flatMapLatest]的工作方式：

在一个新值到来时取消了块中的所有代码 (本示例中的 { requestFlow(it) }）。 这在该特定示例中不会有什么区别，由于调用 requestFlow 自身的速度是很快的，不会发生挂起， 所以不会被取消。然而，如果我们要在块中调用诸如 delay 之类的挂起函数，这将会被表现出来。

总结Flow 关键字

• asFlow
• flowOf
• flow{...}
• withTimeoutOrNull
• 操作符，take, map,reduce,filter,transform，toList，collect，toSet，first，single，reduce，fold，
  flowOn,conflate,collectLatest,Combine
  flatMap,flatMapConcat,flatten,flatMapMerge,flatMapLatest