使用Monifu的rx可观察对象

Question

我只是想掌握冷热可观察对象之间的概念，并尝试Monifu库。我的理解是，下面的代码应该导致只有一个订阅者获得由Observable发出的事件，但它不是!

scala> :paste
// Entering paste mode (ctrl-D to finish)
import monifu.reactive._
import scala.concurrent.duration._
import monifu.concurrent.Implicits.globalScheduler
val obs = Observable.interval(1.second).take(10)
val x = obs.foreach(a => println(s"from x ${a}"))
val y = obs.foreach(a => println(s"from y ${a}"))
// Exiting paste mode, now interpreting.
from x 0
from y 0
import monifu.reactive._
import scala.concurrent.duration._
import monifu.concurrent.Implicits.globalScheduler
obs: monifu.reactive.Observable[Long] = monifu.reactive.Observable$$anon$5@2c3c615d
x: Unit = ()
y: Unit = ()
scala> from x 1
from y 1
from x 2
from y 2
from x 3
from y 3
from x 4
from y 4
from x 5
from y 5
from x 6
from y 6
from x 7
from y 7
from x 8
from y 8
from x 9
from y 9

所以，在我看来，这看起来像是Observable向所有感兴趣的订阅者发布事件?

Answer 1

我是Monifu的主要作者。

冷可观察对象意味着它的订阅函数为每个订阅者(每次subscribe()呼叫)启动一个新的数据源，而热可观察对象在多个订阅者之间共享相同的数据源。

作为一个例子，假设一个文件是数据源。让我们建立一个简单的Observable模型，它从文件中发出以下行:

def fromFile(file: File): Observable[String] = {
  // this is the subscribe function that
  // we are passing to create ;-)
  Observable.create { subscriber =>
    // executing things on our thread-pool
    subscriber.scheduler.execute {
      val source = try {
        Observable.fromIterable(scala.io.Source
          .fromFile(file).getLines().toIterable)
      } 
      catch {
        // subscribe functions must be protected
        case NonFatal(ex) =>
          Observable.error(ex)
      }
      source.unsafeSubscribe(subscriber)
    }
  }
}

这个函数创建一个冷观察对象。这意味着它将为每个订阅的观察者打开一个新的文件句柄，然后读取并发出每个订阅的观察者的行。

但是我们可以把它变成一个热可观察对象:

// NOTE: publish() turns a cold observable into a hot one
val hotObservable = fromFile(file).publish()

当你这样做的时候，区别就在于:

val x = observable.subscribe()
val y = observable.subscribe()

如果可观察对象是hot:

1. 这个可观察对象在你调用connect()之前不会做任何事情
2. 在connect()之后，打开相同的文件，两者将收到完全相同的事件
3. 在发出该文件中的所有行之后，那么新的订阅者将得不到任何，因为(共享的)数据源已经耗尽

如果可观察对象是冷的:

1. 在每次订阅时，打开一个新的文件句柄并读取
2. 元素在subscribe()之后立即释放，因此不需要等待connect()
3. 所有订阅的观察者都将从该文件中接收所有行，无论他们何时订阅

一些同样适用于Monifu的引用:

1. Rx简介:冷热观测
2. RxJava wiki中的主题