0

使用monix我试图通过构建 Observable[Node] 并使用广度优先算法来遍历图形。但是，我有一点递归问题。这是一个说明我的问题的片段：

package gp

import monix.eval.Task
import monix.execution.Scheduler.Implicits.global
import monix.reactive._


object HelloObservable {

  type Node = Int

  //real case fetch next node across the network so the signature 
  //has to be Node -> List[Task[Node]]
  def nexts(i : Node) : List[Task[Node]] =
    List(Task(i), Task(i+1))

  def go(i :Node) : Task[Iterator[List[Node]]] =
    Task.sequence(nexts(i).sliding(100,100).map(Task.gatherUnordered))

  def explore(r: Node): Observable[Node] = {
    val firsts = for {
      ilr <- Observable.fromTask(go(r))
      lr <- Observable.fromIterator(ilr)
      r <- Observable.fromIterable(lr)
    } yield r

    firsts ++ firsts.flatMap(explore)
  }


  def main(args : Array[String]) : Unit = {

    val obs = explore(0)

    val cancelable = obs
      .dump("O")
      .subscribe()

    scala.io.StdIn.readLine()

  }

}

第一次迭代后的可观察停止。谁能提示我为什么？

4

1 回答 1

2

我认为这个问题与递归无关。我认为这是因为您使用slidingwhich 返回一个Iterator. Iterator和之间的主要区别在于Iterable您只能消费Iterator一次，然后剩下的就是一个空的Iterator。这意味着当你这样做时firsts.flatMap，没有任何东西留在里面Observable.fromIterator(ilr)，所以什么也没有产生。

从根本上说，如果您不能在内存中保留（大部分）前缀，我认为您无法进行广度优先搜索。但是由于您nexts已经返回List，我假设您可以在内存中拥有该列表的两个副本。而第二个副本是 的物化结果sliding。所以你的固定代码是这样的：

object HelloObservable {

    import monix.eval.Task
    import monix.execution.Scheduler.Implicits.global
    import monix.reactive._

    type Node = Int

    //real case fetch next node across the network so the signature
    //has to be Node -> List[Task[Node]]
    def nexts(i: Node): List[Task[Node]] = List(Task(i), Task(i + 1))

    def go(i: Node): Task[List[List[Node]]] =
      Task.sequence(nexts(i).sliding(100, 100).toList.map(Task.gatherUnordered))


    def explore(r: Node): Observable[Node] = {
      val firsts = for {
        ilr <- Observable.fromTask(go(r))
        lr <- Observable.fromIterable(ilr)
        r <- Observable.fromIterable(lr)
      } yield r
      firsts ++ firsts.flatMap(explore)
    }


    def main(args: Array[String]): Unit = {

      val obs = explore(0)

      val cancelable = obs
        .dump("O")
        .subscribe()

      scala.io.StdIn.readLine()

    }
}
于 2017-12-26T05:43:04.087 回答