scala - 如何在 Monix 中全局订购多个有序的 observables

Question

假设我有多个有序的迭代器。如果我想合并这些迭代器，同时[(1,3,4), (2,4,5)] -> [1,2,3,4,4,5]使用monix对它们进行全局排序（例如） ，我会怎么做？

Answer 1

不使用 Monix，但我不确定这是否相关

import scala.collection.BufferedIterator


def merge[A:Ordering](xs: Seq[Iterator[A]]) = 
  new Iterator[A] {
    val its = xs.map(_.buffered)
    def hasNext = its.exists(_.hasNext)
    def next = its.filter{ _.hasNext}
                  .minBy(_.head)
                  .next
  }


val ys = merge(Seq(List(1,3,5).toIterator, List(2,4,6).toIterator, List(10,11).toIterator))

ys.toList  //> res0: List[Int] = List(1, 2, 3, 4, 5, 6, 10, 11)

Answer 2

由于 observable 是一个项目流，因此可以概括为两种类型：

• 有限流
• 无限流

请注意，为了正确排序，您需要所有项目。所以，没有简单的方法可以做到这一点。

对于有限流，您必须累积所有项目然后排序。你可以用Observable.fromIterable.

val items = List((1,3,4), (2,4,5))

val sortedList = Observable
  .fromIterable(items)
  .flatMap(item => Observable.fromIterable(List(item._1, item._2, item._3)))
  .toListL // Flatten to an Observable[Int]
  .map(_.sorted) 

对于无限流，您唯一能做的就是将项目缓冲到特定时间或大小。我看不到任何方法，因为您不知道流何时结束。

例如，

val itemsStream: Observable[(Int, Int, Int)] = ???

itemsStream
  .bufferIntrospective(10)
  .flatMap((itemList: List[(Int, Int, Int)]) => // You'll have to sort this
     ???
  )