使用 Monifu 的 Scala Rx Observable

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 在多个订阅者之间共享相同的数据源。

作为示例，将文件视为数据源。让我们建模一个简单的 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()

如果可观察值很热:

1. 在您调用 connect() 之前，可观察对象不会执行任何操作
2. 在 connect() 之后，打开同一个文件，并且两者都将收到完全相同的事件
3. 发出该文件中的所有行后，新订阅者将不会得到任何东西，因为(共享)数据源已经耗尽

如果可观测值是冷的:

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

一些也适用于 Monifu 的引用资料:

1. Connectable Observable from RxJava's wiki
2. Intro to Rx: Hot and Cold Observables
3. Subjects from RxJava's wiki