有人说，Kotlin Flow是Sequence更好的替代品？

前言

这段时刻，笔者在日常开发中总会对一些数据量比较大的列表进行一些操作处理，终究需求拿到一个处理后的成果，但这时分假如创建中心调集会变得十分贵重，运用普通调集性能又不是太好，怎么办呢？作为一名Kotlin开发者，Kotlin现已给出了具体的计划，那便是运用Sequences慵懒序列，不必处理一切数据，然后大大节省了运转时刻；

就目前来说，由于Kotlin Flow流呈现了，它的行为就像一个序列，Sequences能做的工作，它也能做，甚至还能做的更多，这时分有人就说了，那Flow岂不是Sequence更好的替代品了？Sequence要退出历史的舞台了？凡具有相似功用的事物存在，必有这个问题，这儿先不做定论，下面笔者和小伙伴们一同讨论下吧

Kotlin Flow 与 Sequence 对比

首要，在开端探究它们之间的区别前，咱们来看看Flow和Sequence都是怎样处理数据的，直接上代码

 (1..3).asSequence()
       .map {
        println("sequence map $it")
        it * 2
       }
       .first {
        it > 2
       }.let {
        println("sequence $it")
       }
​
    runBlocking {
       (1..3).asFlow()
         .map {
          println("flow map $it")
          it * 2
         }
         .first {
          it > 2
         }.let {
          println("flow $it")
         }
     }

以上它们的行为都是相似的，如下图所示

成果清楚明了，两种方式的成果都是一致的,在获取成果之后马上终止了该进程

和传统List比较，它们的确节省了不少时刻，能够不必处理一切的值，在这种处理工作的行为上面，Flow和Sequence都是相同的。就好像，现在Flow和Sequence是两名艺人，都是歌唱的，但这个时分风趣的工作发生了，Flow说我不当当会演戏，我还会歌唱，跳舞，Rap，篮球等等

总而言之，Flow能做的工作比Sequence多得多，接下来咱们来看看Flow和Sequence比较有哪些优点

1. Sequence是堵塞的，Flow是非堵塞的

为什么这么说呢，这儿的堵塞是堵塞主线程，请答应我娓娓道来，咱们仍是分隔来说吧，这样方便了解

• 关于Sequence

  先来生成一个序列，这儿咱们想要模仿一个慢推迟的作用，可是在序列傍边是不答应有suspend函数的，所以笔者运用Thread.sleep来替代，看下代码

  fun simple() = sequence {
     (1..3).forEach { Thread.sleep(100); yield(it) }
  }
  ​
  fun main() = runBlocking<Unit> {
    launch {
      for (k in 1..3) {
        println("From main $k")
        delay(100)
       }
     }
  ​
    simple().forEach { value -> println("From sequence $value") }
  }
  

  此刻成果展现如下

  

  能够看到咱们是先从sequence取出成果后，再去履行主线程的内容

• 关于Flow

  相同咱们运用Flow来完结上面列表，模仿一个慢推迟的作用，在Flow中能够运用delay挂起函数来进行操作，如下代码所示

  fun simple() = flow {
     (1..3).forEach {
      delay(100)
      emit(it)
     }
  }
  ​
  fun main() = runBlocking {
    launch {
      for (k in 1..3) {
        println("From main $k")
        delay(100)
       }
     }
    simple().collect { value -> println("From flow $value") }
  }
  

  此刻成果展现如下图所示：

  

  能够看到Flow流并不影响主线程的内容履行

2. Sequence不能轻易撤销，Flow能够随时撤销

在序列傍边，即使运转进程很慢，也不能被半途撤销

fun simple() = sequence {
   (1..3).forEach {
    Thread.sleep(100)
    yield(it)
   }
}
​
fun main() = runBlocking {
  //250ms后超时
  withTimeoutOrNull(250) {
    simple().forEach { value -> println(value) }
   }
  println("Done")
}

运转成果如咱们所料，一旦序列开端履行后，就算咱们设置了超时时长，但也什么都没有改动，什么都改动不了….

而假如运用Flow的话，咱们半途能够进行暂停

fun simple() = flow {
   (1..3).forEach {
    delay(100)
    emit(it)
   }
}
​
fun main() = runBlocking {
  //250ms后超时
  withTimeoutOrNull(250) {
    simple().collect { value -> println(value) }
   }
  println("Done")
}

这样流程在250毫秒后会被撤销，成果如下

3. Sequence自身不能轻易扩展,Flow 能够很容易地自我扩展

想象一下，此刻有一个列表2，4，6 ，咱们想要将它扩展成1，2，3，4，5，6，这时分呢能够运用transform运算符完结这个操作，来瞥一眼代码

fun main() = runBlocking {
   (2..6 step 2).asFlow().transform {
    emit(it - 1)
    emit(it)
   }.collect { println(it) }
}

输出成果清楚明了，将从本来的2, 4, 6转到1, 2, 3, 4, 5, 6

思考一下，假如是序列呢，它能这样扩展么？答案信任大家现已清楚了

4. Sequence不能独自在另一个线程中发动,Flow 能够在另一个线程中发动自己

怎么说呢？在序列傍边，假如咱们真实想要在另一个线程中运用序列的话，必须要依靠独自的工具(例如协程)，不然单靠自己自身是无法完结的。

fun main() {
  run()
  //保证其他线程完结
  Thread.sleep(100)
}
​
fun run() {
  CoroutineScope(Dispatchers.IO).launch {
     (1..3).asSequence()
       .forEach {
        println("$it ${Thread.currentThread()}")
       }
   }
}

这样的话，咱们就在子线程中发动了咱们的序列

反观Flow,它自身便是基于协程来履行的，所以咱们能够运用launchIn轻松让它在另一个线程中发动,并为其供给相应的作用域CoroutineScope

fun main() {
  run()
  // 保证其他线程完结
  Thread.sleep(100) 
}
​
fun run() {
   (1..3).asFlow()
     .onEach {
      println("$it ${Thread.currentThread()}")
     }.launchIn(CoroutineScope(Dispatchers.IO))
}

便是这么简略，简练而甜美，直接在它自身上进行操作就完结了

5. Sequence不能在两个线程中发送接纳操作,Flow则能够在另一个线程中运转

什么意思呢？假设咱们想在一个线程中触发程序并在另一个线程中运转，终究在另一个线程中搜集数据，这样的话，序列Sequence并不能满意，咱们需求运用Flow进行完结，Kotlin有供给flowOn函数

fun main() = runBlocking {
  flow {
     (1..3).forEach {
      println("Fire $it ${Thread.currentThread()}")
      emit(it)
     }
   }
     .flowOn(Dispatchers.IO)
     .transform {
      println("Operate $it ${Thread.currentThread()}")
      emit(it)
     }
     .flowOn(Dispatchers.Default)
     .collect {
      println("Collect $it ${Thread.currentThread()}")
     }
}

成果如下图所示

6. Sequence不能够并行处理，Flow能够并行处理

在序列中，履行元素是依照列表顺序履行的，只有上一个元素履行结束后才会履行下一个元素，它无法进行并行处理的；仍是用一个例子来阐明吧，假如每个元素的生成需求 100ms，每个元素的处理进程又需求 300ms，下面每一轮大约需求400ms，来撇一下代码

fun simple() = sequence {
   (1..3).forEach {
    Thread.sleep(100)
    yield(it)
   }
}
​
fun main() = runBlocking {
  val time = measureTimeMillis {
    simple().forEach {
      delay(300)
     }
   }
  println("Collected in $time ms")
}

清楚明了，由于具有3个元素，这段程序耗时大约需求 3 x 400ms = 1200ms，而实践成果也的确如咱们所料，实践耗时在1257ms,差不多

而在Flow中，咱们能够运用buffer要求发射器持续其工作而无需等候处理完结，也便是说Flow能够进行数据的并行处理，仍是依照上述要求写一段代码

fun simple(): Flow<Int> = flow {
  for (i in 1..3) {
    delay(100)
    emit(i)
   }
}
​
fun main() = runBlocking {
  val time = measureTimeMillis {
    simple().buffer().collect {
      delay(300)
     }
   }
  println("Collected in $time ms")
}

这段程序的处理耗时是1083ms

这样的话也便是说，只有第一个元素将花费大约 100ms + 300 ms，列表元素搜集只需求 300ms，由于 100ms 的触发时刻是与处理并行完结的。

7. Sequence无法除去快慢元素，Flow能够消除快慢元素

结合上一个观念，序列Sequence无法进行并行处理，不管元素处理快慢与否，都得依照先后顺序来，Flow能够并行处理，基于此，咱们能够在搜集元素之前做一些操作，能够除去发射快的元素，也能够除去比较慢的元素，经过这些来进步程序的运转效率。

在Flow中，一般咱们能够经过Conflate函数和CollectLatest函数完结

• Conflate

  运用Conflate函数，咱们能够挑选经过仅搜集最新元从来消除快速发出的元素，如下代码所示

  fun simple(): Flow<Int> = flow {
    for (i in 1..3) { delay(100); emit(i) }
  }
  ​
  fun main() = runBlocking<Unit> {
    val time = measureTimeMillis {
      simple().conflate().collect {
        delay(300)
        println(it)
       }
     }
    println("Collected in $time ms")
  }
  

  元素 2 被兼并，由于元素 3 在元素 1 彻底处理之前就已预备就绪。结合下图会更好地了解这个流程。

（在这种状况下，元素 2 将被兼并，由于元素 3 在元素 1 完结处理之前发出）

CollectLatest

相同的，咱们运用CollectLatest函数，假如有最新的元素发出，这时分咱们就能够挑选消除去之前缓慢的元素，终究接纳的元素便是最新的元素，如下代码所示

 fun simple(): Flow<Int> = flow {
   for (i in 1..3) { delay(100); emit(i) }
  }
 
 fun main() = runBlocking {
   val time = measureTimeMillis {
     simple().collectLatest {
       println("get $it")
       delay(300)
       println("done $it")
      }
    }
   println("Collected in $time ms")
  }

搜集了一切元素 1、2 和 3。但元素 1 和 2 被消除，由于下一个元素在彻底处理之前就抵达了，结合下图会更好地了解这个流程。

8. Sequence只能同步组合，Flow能够异步组合

这是什么意思呢，往常开发咱们都会碰到两个以上的序列或者流需求一同履行的状况，这个时分就需求用到组合运算符了，当然序列和流都有相应的组合的运算符，但它们之间亦有不同，不管是序列仍是流，都有zip运算符来进行组合，当然流还有另外的运算符，这个之后会说，咱们先来看看zip运算符

• zip运算符(序列和流)

  关于序列，即使元素是以不同的速度进行生成，但终究的成果也是同步的，即逐一按元素排序,来先上代码

  fun firstSeq() = sequence {
     (1..3).forEach {
      Thread.sleep(100)
      yield(it)
     }
  }
  ​
  fun secondSeq() = sequence {
     (4..6).forEach {
      Thread.sleep(300)
      yield(it)
     }
  }
  fun main() = runBlocking {
    val time = measureTimeMillis {
      firstSeq().zip(secondSeq()).forEach {
        println(it)
       }
     }
    println("Collected in $time ms")
  }
  

  接着咱们运用Flow，相同完结上面这个流程

  fun firstFlow() = flow {
     (1..3).forEach {
      delay(100)
      emit(it)
     }
  }
  ​
  fun secondFlow() = flow {
     (4..6).forEach {
      delay(300)
      emit(it)
     }
  }
  ​
  fun main() = runBlocking {
    val time = measureTimeMillis {
      firstFlow().zip(secondFlow()) { first, second ->
        Pair(first, second)
       }.collect {
        println(it)
       }
     }
    println("Collected in $time ms")
  }
  

  代码有点多，现在咱们来看下两者的成果

尽管输出成果是相同的，但很明显Flow要快上一些，原因就在于序列履行的元素是串联摆放的，第一个和第二个元素被紧缩在一同，所耗费的时刻也是两者相加，而Flow是两者并排履行的，所耗费的时刻是两边较长耗时的一方；简略来说，前者是串行的，后者是并行的，结合下图会更好的了解

• combine运算符(仅支持流)

  在Flow中，咱们还能够运用combine运算符来组合两个具有不同速率的流，但它仅仅只能用在Flow中，相同它不会堵塞其间任何一个，这便是Flow为什么能够异步组合的原因了，老规矩，先上代码

  fun main() = runBlocking {
    val time = measureTimeMillis {
      firstFlow().combine(secondFlow()) { first, second ->
        Pair(first, second)
       }.collect { println(it) }
     }
    println("Collected in $time ms")
  }
  

  这儿咱们将上文的zip运算符换成combine运算符，此刻成果会不会有所不同呢？一同来看下

  

  如上图成果所示，第一个流持续发出元素，当第二个流呈现时，它会采纳任何可用的元素（在这种状况下2并与4它所具有的结合起来）,这点和zip运算符是不同的，由于篇幅原因，就不细说它们之间的区别了，咱们只需知道zip运算符会等候一切流回来新的元素才会进行组合，而combine不会等候流中是否有新的元素，它会马上调用转换函数；简略一点，zip要等候新元素才进行组合，combine不等候，只需有元素就组合，具体想要了解的同学能够看Kotlin的官方文档

  好像上面的例子中， 在元素2和4之间发出的3，因而它会产生(2, 4)和(3, 4)

9. Sequence只做同步展平，Flow能够进行异步展平

所谓展平呢，简略来说，便是序列/流在接纳元素的时分，或许需求另一个序列/流的元素，这两者进行交互的操作。对此它们都有供给相应的运算符FlatMap，而Flow流展平一般有三种形式，连接形式FlatMapConcat，兼并形式FlatMapMerge，最新展平形式flatMapLatest，咱们就不作过多讨论了，这儿仅仅简略将序列和流的展平做一个比较，想要了解流展平的同学，能够看一下这篇文章:

Flow 流展平 ( 连接形式 flatMapConcat | 兼并形式 flatMapMerge | 最新展平形式 flatMapLatest )

• 回归主题，先来看看序列的展平，运用FlatMap，咱们来看下代码

  fun requestSequence(i: Int): Sequence<String> = sequence {
    yield("$i: First")
    Thread.sleep(300)
    yield("$i: Second")
  }
  ​
  fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
     (1..3).asSequence().onEach {
      Thread.sleep(100)
     }.flatMap {
      requestSequence(it)
     }.forEach {
      println("$it at ${System.currentTimeMillis() - startTime} ms from start")
     }
  }
  

  运转成果如下图所示:

  

  在序列中，悉数元素都是串行完结的，所以必须按顺序处理，很明显咱们现已知道了终究的成果，但相对来说所需求的时刻是最长的，由于每个序列都是依照供给的顺序连续处理的。结合下图会更好的了解该流程

  

• 接着简略看下Flow中的展平，这儿就以兼并形式FlatMapMerge，最新展平形式flatMapLatest为例，它们都是并行处理数据的，前者是假如任何一个元素完结处理，能够先展平为成果，无需等候，后者顾名思义，当发射了新值之后，上个 flow 就会被撤销，假如只想要最新的值，而且消除前面比它慢的元素,能够运用FlatMapLatest

  • 兼并形式FlatMapMerge

    仍是老规矩，用一个例子来辅助阐明

    fun otherFlow(i: Int): Flow<String> = flow {
      emit("$i:First")
      delay(300)
      emit("$i:Second")
    }
    ​
    @OptIn(FlowPreview::class)
    fun main() = runBlocking {
      val startTime = System.currentTimeMillis()
       (1..3).asFlow().onEach { delay(100) }
         .flatMapMerge {
          otherFlow(it)
         }
         .collect { value ->
          println("$value 从开端耗时 ${System.currentTimeMillis() - startTime} ms ")
         }
    }
    

    模棱两可，成果如咱们预期相同，它是并行处理的，即在一个流处理完结之前，另一个流也能够开端，所以它所耗费的时刻基本是由其间耗时最长的那个流决议的，这样也能更快的产生成果，进步效率

    

    再结合下图进行了解该流程

    

  • 最新展平形式FlatMapLatest

    仍是相同运用上面的流程，这儿咱们将运算符改为FlatMapLatest，看看成果会有什么不同，先瞄一眼代码

    fun otherFlow(i: Int): Flow<String> = flow {
      emit("$i:First")
      delay(300)
      emit("$i:Second")
    }
    ​
    @OptIn(ExperimentalCoroutinesApi::class)
    fun main() = runBlocking {
      val startTime = System.currentTimeMillis()
       (1..3).asFlow().onEach { delay(300) }
         .flatMapLatest { otherFlow(it) }
         .collect { value ->
          println("$value 从开端耗时 ${System.currentTimeMillis() - startTime} ms ")
         }
    }
    

    成果如下所示：

    

这时分信任各位小伙伴都现已发现了，第二个flow中第一个和第二个值都不见了，只展平了第三个值，这便是FlatMapLatest的共同地点了，它和咱们这儿没有提到的FlatMapConcat相似，都是后面发射的值会撤销之前正在处理的值，这儿就不展开了，能够结合下图进行尝试了解该流程

10. Sequence反常处理运用Try-Catch，Flow有封装好的反常运算符

不管是Sequence仍是Flow，关于反常的处理都能够运用try-catch

fun simple(): Sequence<Int> = sequence {
  for (i in 1..3) {
    println("Generating $i")
    yield(i)
   }
}
​
fun main() = runBlocking {
  try {
    simple().forEach { value ->
      check(value <= 1) { "Crash on $value" }
      println("Got $value")
     }
   } catch (e: Throwable) {
    println("Caught $e")
   } finally {
    println("Done")
   }
}

以上是Sequence的反常处理，成果如下所示

Flow有封装好的操作符Catch和OnCompletion，能够直接在链上履行，以此能够简化代码，如下代码所示

fun simple(): Flow<Int> = flow {
  for (i in 1..3) { println("Generating $i"); emit(i) }
}
​
fun main() = runBlocking {
  simple().onEach { value ->
    check(value <= 1) { "Crash on $value" }
    println("Got $value")
   }.catch { e ->
    println("Caught $e")
   }.onCompletion {
    println("Done")
   }.collect()
}

成果和上面彻底一致的，可是代码看起来简练了些，高雅了些

小结

经过以上观念来看的话，大家是不是觉得，显然Flow比Sequence好太多了，Flow便是像是加强版的序列Sequence Plus，能够做更多的工作，例如线程切换、撤销、异步和并行处理等等。

那么这个时分就有人会说了，那你的意思便是意味着咱们只运用Flow就能够满意日常开发需求咯，那还要Sequence有什么用呢？这时分Sequence听到两眼一黑，说罢便预备退出表演的舞台，一边自言自语，算了，我不拍了，是金子总会发光的……

这时分Kotlin制片人老大哥大喊一声，不是这样的，我知道你很急，但你先别急，Sequence仍旧有自己的简略性和优势，它在有些状况仍然比Flow好，这部电影你的角色仍旧不可或缺，即使是小角色，这时分Sequence停下了脚步，急忙跑到Kotlin身边，展颜欢笑，”不如说说我都能担任什么角色吧”

话都提到这个份上了，接下来让咱们一同来看看吧

什么状况下运用Sequence而不是Kotlin Flow？

上面咱们说了Flow很多优点，但也不是一切场景都适用于Flow的，有些特定的场景反而运用Sqeuence作用会更佳，Sequence此刻搬过来一个小板凳做到了旁边，用力的撇撇嘴，大声说道：”尽管我只会演戏，即使仅仅一个小角色，但我也要做到最好”

减少开销，Sequence比Flow更轻量

好像前面所说，Flow承载了太多太多东西了，它环绕协程进行开发的，具有异步履行事物、线程、并行处理等的能力等等，这无疑将带来额定的开销。咱们来做个对比吧，用现实来说话

• 运用序列来加载一亿个数据，咱们来看看耗时

  val sequence = (1..100000000).asSequence()
  val startTime = System.currentTimeMillis()
  fun main() = runBlocking {
    val result = sequence
       .map { it * 3 }
       .filter { it % 2 == 0 }
    println("Start")
    result.reduce { ac, it ->
      ac + it
     }.run { println(this) }
    println("Done in ${System.currentTimeMillis() - startTime}ms")
  }
  

成果如下，大约需求1秒左右

• 相同运用Flow，来履行相似的操作，咱们再来看看耗时

val flows = (1..100000000).asFlow()
val startTime = System.currentTimeMillis()
fun main() = runBlocking {
  val result = flows.map { it * 3 }
     .filter { it % 2 == 0 }
  println("Start")
  result.reduce { ac, value ->
    ac + value
   }.run { println(this) }
  println("Done in ${System.currentTimeMillis() - startTime}ms")
}

成果如下所示，能够看到耗时比序列慢了差不多1.5秒

换句话说，假如咱们不需求Flow带来的这些功用，能够挑选序列，这样也能进步程序的效率

Sequence能够遍历每个元素，Flow不能直接循环遍历

假如咱们由于某种原因想操控每个元从来履行某种算法，但又不想用操作运算符，只想经过for-loop来完结它，那么序列是支持这种做法

for (i in (1..3).asSequence()) {
    println(i)
   }

可是Flow并不能做到这一点，它只能经过操作符onEach而且要collect搜集它才干完结上面操作，不支持for-loop操作，假如咱们想这么操作的话，那么运用序列才干完结

序列比较Flow具有更多可用的运算符

目前来说，有大量的Kotlin调集运算符适用于序列，但有些或许不适用于Flow,比如chunk ，windowed, zipWithNext,toIterator，ifEmpty等等。假如你需求它们，容器要么是一个list，要么是一个sequence，这儿笔者就不详细展开了，之后会专门整理出一篇文章阐明这些操作符用法和运用场景。

值得一提的是，Flow和Sequence能够轻松相互转换，真实做到你中有我，我中有你

• 从Flow转换到Sequence

  sequenceOf((1..3).asFlow())
  

• 从Sequence转换到Flow

  (1..3).asSequence().asFlow()
  

序列能够直接履行，Flow必须依靠协程

关于Flow，它是在协程的环境中运转的：

• 在协程规模内

   runBlocking {
       (1..3).asFlow().collect()
     }
  

• 在挂起函数中

  suspend fun main() {
       (1..3).asFlow().collect()
     }
  

• 运用launchIn设置协程规模

  (1..3).asFlow().launchIn(CoroutineScope(Dispatchers.IO))
  

  而关于Sequence，它能够自行运转，不管你在哪里设置…

小结

听君一席话，如听一席话，Sequence此刻握紧Kotlin制片人的手，仍是您懂我啊，我只想做一个朴实的艺人啊…..

好的，回顾一下，假如咱们遇到以下状况的话，就能够运用序列Sequence了

• 不需求任何Flow流供给的强大的功用

• 需求自己手动For循环遍历元素的时分

• 需求拜访更多可用的调集操作符

• 不想在协程的环境中

• 不需求去异步请求数据的时分

  …..

终究想说的话

综上所述，Sequence实践上并不是Flow的替代品，物尽善其美，任何事物的存在都有它的价值，文章关于它们之间的讨论就到这儿吧，可是它们之间的故事仍然在持续……

参阅文章

• 这一次，让Kotlin Flow 操作符真实好用起来
• Kotlin Flow flatMapLatest
• 关于kotlin中的Collections、Sequence、Channel和Flow (一)
• Kotlin中的慵懒操作容器——Sequence