scala.rx：实践指南

实际评估scala.rx时，我先确认它解决的具体问题：Scala 函数响应式编程实验库。一旦进入软件开发环节，依赖、接口和异常处理往往比主路径更影响采用会直接影响交付，这也是我最关心的风险。我建议在隔离分支完成一个可回滚的小任务，重点记录安装步骤、接口契约、测试结果和错误信息，再与现有方案比较。它更像给需要可检查开发流程而非单次演示的工程师准备的可审查方案，是否长期使用应由试跑数据决定。

Scala.Rx 0.4.1

Scala.Rx 是 Scala 的变更传播库。 Scala.Rx 为您提供反应变量（Rxs），它们是智能变量，当它们依赖的值发生变化时，它们会自动更新。底层实现是基于推送的 FRP，基于 弃用观察者模式 中的思想。

演示该行为的一个简单示例是：

import rx._
val a = Var(1); val b = Var(2)
val c = Rx{ a() + b() }
println(c.now) // 3
a() = 4
println(c.now) // 6

这个想法是，99% 的情况下，当您重新计算变量时，您会按照最初计算它的方式重新计算它。此外，只有当它所依赖的值之一发生变化时，您才需要重新计算它。 Scala.Rx 自动为您完成此操作，并为您处理所有繁琐的更新逻辑，以便您可以专注于其他更有趣的事情！

除了基本的更改传播之外，Scala.Rx 还提供了许多其他功能，例如一组用于轻松构建数据流图的组合器、用于高度正确性的编译时检查以及与现有 Scala 代码的无缝互操作。这意味着它可以轻松嵌入到现有的 Scala 应用程序中。

内容

• 入门
• ScalaJS
• 使用 Scala.Rx
  • 基本用法
  • 所有权上下文
  • 数据依赖
  • 附加操作
  • 异步组合器
  • 图形检查
  • 日志记录和调试
• 设计注意事项
• 相关工作
• Scaladoc

开始使用

Scala.Rx 在 Maven Central 上可用。首先，只需将以下内容添加到您的 build.sbt 中：

libraryDependencies += "com.lihaoyi" %% "scalarx" % "0.4.1"

之后，打开 sbt console 并将上面的示例粘贴到控制台中应该可以工作！您可以继续完成 基本用法 页面中的示例，以了解 Scala.Rx 的功能。

ScalaJS

除了在 JVM 上运行之外，Scala.Rx 还可以编译为 Scala-Js！该工件当前位于 Maven Central 上，并可通过以下 SBT 片段使用：

libraryDependencies += "com.lihaoyi" %%% "scalarx" % "0.4.1"

在 JVM 上运行 Scala.Rx 和在 Javascript 中运行 Scala.Rx 之间存在一些细微差别，特别是在 异步操作、并行模型 和 内存模型方面。不过，一般来说，下面文档中给出的所有示例在交叉编译为 javascript 并在浏览器中运行时都可以完美运行！

Scala.rx 0.4.1仅与ScalaJS 0.6.5+兼容。

使用Scala.Rx

主要操作只需要import rx._即可使用，附加操作也需要import rx.ops._。下面的一些示例也使用来自 scala.concurrent 或 scalatest 的各种导入。

基本用法

import rx._

val a = Var(1); val b = Var(2)
val c = Rx{ a() + b() }
println(c.now) // 3
a() = 4
println(c.now) // 6

上面的例子是一个可执行程序。一般来说，import rx._ 足以让您开始使用 Scala.Rx，并且在所有进一步的示例中都将假定它。这些例子都取自单元测试。

您必须关心的基本实体是 Var、Rx 和 Obs：

• Var: a 智能变量，您可以使用 a() 获取并使用 a() = ... 设置。每当其值发生变化时，它都会 ping 任何需要重新计算的下游实体。
• Rx: a 反应式定义自动捕获在其主体中调用的任何 Var 或其他 Rx，将它们标记为依赖项，并在其中之一发生更改时重新计算。与 Var 一样，您可以使用 a() 语法来检索其值，并且当值更改时它还会 ping 下游实体。
• Obs: an 观察者在一个或多个 Var 或 Rx s 上，当观察到的节点更改值并向其发送 ping 时执行一些副作用。

使用这些组件，您可以轻松构建数据流图，并在图表的输入发生变化时使数据流图中的各种值保持最新：

val a = Var(1) // 1

val b = Var(2) // 2

val c = Rx{ a() + b() } // 3
val d = Rx{ c() * 5 } // 15
val e = Rx{ c() + 4 } // 7
val f = Rx{ d() + e() + 4 } // 26

println(f.now) // 26
a() = 3
println(f.now) // 38

该程序的数据流图如下所示：

其中 Var 由正方形表示，Rx 由圆形表示，依赖关系由箭头表示。每个 Rx 都标有其名称、主体和值。

修改 a 的值会导致更改通过数据流图传播

从上面可以看出，更改 a 的值会导致更改通过 c d e 一直传播到 f。您可以在使用普通变量的任何地方使用 Var 和 Rx。

更改通过waves 中的数据流图传播。对 Var 的每次更新都会触发一次传播，将更改从 Var 推送到任何（直接或间接）依赖于其值的 Rx。在此过程中，Rx可能会被重新计算多次。

观察员

如前所述， Obs 可以从 Rx 或 Var 创建，并用于在它们更改时执行副作用：

val a = Var(1)
var count = 0
val o = a.trigger {
  count = a.now + 1
}
println(count) // 2
a() = 4
println(count) // 5

这将创建一个如下所示的数据流图：

当a被修改时，观察者o将执行副作用：

Rx 的主体应该没有副作用，因为它们每次传播可能运行多次。您应该使用 Obs 来执行副作用，因为在所有 Rx 的值稳定后，它们保证每次传播仅运行一次。

Scala.Rx 提供了一个方便的 .foreach() 组合器，它提供了从 Rx 创建 Obs 的替代方法：

val a = Var(1)
var count = 0
val o = a.foreach{ x =>
  count = x + 1
}
println(count) // 2
a() = 4
println(count) // 5

此示例与上面的代码执行相同的操作。

请注意，Obs 的主体在声明时最初运行一次。这与每个 Rx 在最初声明时计算一次的方式相匹配。但可以想象，您想要一个 Obs，仅当它正在的 Rx 更改时才第一次触发。您可以使用备用 triggerLater 语法来执行此操作：

val a = Var(1)
var count = 0
val o = a.triggerLater {
  count = count + 1
}
println(count) // 0
a() = 2
println(count) // 1

Obs 用于封装它运行的回调。它们可以传递、存储在变量中等。当 Obs 被垃圾回收时，回调将停止触发。因此，Obs应该存储在它影响的对象中：如果回调只影响该对象，那么Obs本身何时被垃圾收集并不重要，因为它只会在持有它的对象变得不可访问之后发生，在这种情况下无论如何都无法观察到它的影响。如果需要更强的保证，也可以主动关闭 Obs：

val a = Var(1)
val b = Rx{ 2 * a() }
var target = 0
val o = b.trigger {
  target = b.now
}
println(target) // 2
a() = 2
println(target) // 4
o.kill()
a() = 3
println(target) // 4

手动调用.kill()后，Obs不再触发。除了 .kill()ing Obs 之外，您还可以杀死 Rx，这会阻止进一步更新。

一般来说，Scala.Rx 围绕构建数据流图来自动保持事物同步，您可以轻松地从外部命令性代码与之交互。这涉及使用：

• Var 作为命令式世界数据流图的输入
• Rx作为数据流图中的中间节点
• Obs 作为数据流图的输出回到命令式世界

复杂反应物

Rx不限于Ints。 Strings、Seq[Int]s、Seq[String]s，任何东西都可以进入 Rx：

val a = Var(Seq(1, 2, 3))
val b = Var(3)
val c = Rx{ b() +: a() }
val d = Rx{ c().map("omg" * _) }
val e = Var("wtf")
val f = Rx{ (d() :+ e()).mkString }

println(f.now) // "omgomgomgomgomgomgomgomgomgwtf"
a() = Nil
println(f.now) // "omgomgomgwtf"
e() = "wtfbbq"
println(f.now) // "omgomgomgwtfbbq"

如图所示，您可以使用 Scala.Rx 的反应变量对任意复杂的问题进行建模，而不仅仅是涉及原始数字的琐碎问题。

错误处理

由于 Rx 的主体可以是任意 Scala 代码，因此它可以抛出异常。将异常向上传播到调用堆栈没有多大意义，因为评估 Rx 的代码可能无法控制其失败的原因。相反，任何异常都会被 Rx 本身捕获并在内部存储为 Try。

这可以在以下单元测试中看到：

val a = Var(1)
val b = Rx{ 1 / a() }
println(b.now) // 1
println(b.toTry) // Success(1)
a() = 0
intercept[ArithmeticException]{
  b()
}
assert(b.toTry.isInstanceOf[Failure])

最初，a 的值为 1，因此 b 的值为 1。您还可以使用 b.toTry 提取内部 Try，最初是 Success(1)。

然而，当a的值变成0时，b的主体会抛出ArithmeticException。如果您尝试使用 b() 从 b 中提取值，则会被 b 捕获并重新抛出。您可以使用 toTry 提取整个 Try 并对其进行模式匹配，以处理 Success 情况以及 Failure 情况。

当您将许多 Rx 链接在一起时，异常会按照依赖关系图向前传播，正如您所期望的那样。下面的代码：

val a = Var(1)
val b = Var(2)

val c = Rx{ a() / b() }
val d = Rx{ a() * 5 }
val e = Rx{ 5 / b() }
val f = Rx{ a() + b() + 2 }
val g = Rx{ f() + c() }

inside(c.toTry){case Success(0) => () }
inside(d.toTry){case Success(5) => () }
inside(e.toTry){case Success(2) => () }
inside(f.toTry){case Success(5) => () }
inside(g.toTry){case Success(5) => () }

b() = 0

inside(c.toTry){case Failure(_) => () }
inside(d.toTry){case Success(5) => () }
inside(e.toTry){case Failure(_) => () }
inside(f.toTry){case Success(3) => () }
inside(g.toTry){case Failure(_) => () }

创建如下所示的依赖关系图：

在此示例中，最初 a、b、c、d、e、f 和 g 的所有值均已明确定义。但是，当 b 设置为 0 时：

c 和 e 都会导致异常，并且来自 c 的异常传播到 g。例如，尝试使用 g.now 从 g 中提取值将重新抛出 ArithmeticException。同样，使用 toTry 也可以。

嵌套

Rx 可以包含任意深度的其他 Rx。此示例显示 Rx 嵌套两层：

val a = Var(1)
val b = Rx{
    (Rx{ a() }, Rx{ math.random })
}
val r = b.now._2.now
a() = 2
println(b.now._2.now) // r

在此示例中，我们可以看到，虽然我们修改了 a，但这仅影响左内 Rx，右内 Rx（每次重新计算时都会采用不同的随机值）或外 Rx（这将导致整个事物重新计算）都不会受到影响。一个稍微不那么做作的例子可能是：

var fakeTime = 123
trait WebPage{
    def fTime = fakeTime
    val time = Var(fTime)
    def update(): Unit  = time() = fTime
    val html: Rx[String]
}
class HomePage(implicit ctx: Ctx.Owner) extends WebPage {
    val html = Rx{"Home Page! time: " + time()}
}
class AboutPage(implicit ctx: Ctx.Owner) extends WebPage {
    val html = Rx{"About Me, time: " + time()}
}

val url = Var("www.mysite.com/home")
val page = Rx{
    url() match{
        case "www.mysite.com/home" => new HomePage()
        case "www.mysite.com/about" => new AboutPage()
    }
}

println(page.now.html.now) // "Home Page! time: 123"

fakeTime = 234
page.now.update()
println(page.now.html.now) // "Home Page! time: 234"

fakeTime = 345
url() = "www.mysite.com/about"
println(page.now.html.now) // "About Me, time: 345"

fakeTime = 456
page.now.update()
println(page.now.html.now) // "About Me, time: 456"

在本例中，我们定义一个具有 html 值（Rx[String]）的网页。但是，根据 url，它可能是 HomePage 或 AboutPage，因此我们的 page 对象是 Rx[WebPage]。

拥有一个 Rx[WebPage]，其中 WebPage 内部有一个 Rx[String]，看起来自然而明显，而 Scala.Rx 让您可以简单自然地做到这一点。当以面向对象的方式建模问题时，这种对象内对象的情况非常自然地出现。 Scala.Rx 能够优雅地处理 Rx 内相应的 Rx 的能力，使其能够优雅地适应这种范式，这是我调查的大多数 相关工作 中所缺乏的。

这里的大多数示例都取自 单元测试，它提供了有关如何使用此库的更多示例。

所有权背景

在上面的最后一个示例中，我们必须引入 所有权 的概念，其中使用了 Ctx.Owner。事实上，如果我们忽略 (implicit ctx: Ctx.Owner)，我们会得到以下编译时错误：

错误：此 Rx 可能会泄漏！要么显式地将其标记为不安全（Rx.unsafe），要么确保隐式 RxCtx 在范围内！
           val html = Rx{"Home Page! time: " + time()}

要了解 ownership，重要的是要了解它修复的问题：leaks。作为一个例子，请考虑对第一个示例的轻微修改：

var count = 0
val a = Var(1); val b = Var(2)
def mkRx(i: Int) = Rx.unsafe { count += 1; i + b() }
val c = Rx{ 
  val newRx = mkRx(a()) 
  newRx() 
}
println(c.now, count) //(3,1)

在这个版本中，增加了函数mkRx，但c的计算值保持不变。修改 a 似乎表现符合预期：

a() = 4
println(c.now, count) //(6,2)

但是如果我们修改 b 我们可能会开始注意到一些不太正确的地方：

b() = 3 
println(c.now, count) //(7,5) -- 5??

(0 to 100).foreach { i => a() = i }
println(c.now, count) //(103,106)

b() = 4
println(c.now, count) //(104,211) -- 211!!!

在此示例中，即使 b 仅更新了几次，但随着 a 的修改，计数值开始飙升。这是 mkRx 泄漏！也就是说，每次重新计算 c 时，它都会构建一个全新的 Rx ，即使在它不再作为数据依赖项可访问并被遗忘之后，它也会保留并继续评估。因此，运行 (0 to 100).foreach 语句后，每次 b 更改时，都会触发超过 100 个 Rxs。这显然是不可取的。

但是，通过添加显式的 owner （并删除 unsafe），我们可以修复泄漏：

var count = 0
val a = Var(1); val b = Var(2)
def mkRx(i: Int)(implicit ctx: Ctx.Owner) = Rx { count += 1; i + b() }
val c = Rx{ 
  val newRx = mkRx(a()) 
  newRx() 
}
println(c.now,count) // (3,1)
a() = 4
println(c.now,count) // (6,2)
b() = 3
println(c.now,count) // (7,4)
(0 to 100).foreach { i => a() = i }
println(c.now,count) //(103,105)
b() = 4
println(c.now,count) //(104,107)

所有权通过继续允许父 Rx 跟踪其“拥有的”嵌套 Rx 来修复泄漏。也就是说，每当 Rx 重新计算时，它都会首先杀死其拥有的所有依赖项，确保它们不会泄漏。在此示例中，c 是在 mkRx 中创建的所有 Rxs 的所有者，并在每次 c 重新计算时自动杀死它们。

数据上下文 使用 () （又名 apply）给定 Rx 或 Var 会解开当前值并将其自身添加为当前正在评估的任何 Rx 的依赖项。或者，.now 可用于简单地解开该值并跳过成为数据依赖项：

val a = Var(1); val b = Var(2)
val c = Rx{ a.now + b.now } //not a very useful `Rx`
println(c.now) // 3
a() = 4
println(c.now) // 3 
b() = 5
println(c.now) // 3 

要了解 Data 上下文的需求以及 Data 上下文与 Owner 上下文有何不同，请考虑以下示例：

def foo()(implicit ctx: Ctx.Owner) = {
  val a = rx.Var(1)
  a()
  a
}

val x = rx.Rx{val y = foo(); y() = y() + 1; println("done!") }

有了所有权的概念，如果允许 a() 创建对其所有者的数据依赖，它将进入无限递归并炸毁堆栈！相反，上面的代码给出了这个编译时错误：

<console>:17: error: No implicit Ctx.Data is available here!
        a()

我们可以通过显式允许数据依赖性来“修复”错误（并看到堆栈爆炸）：

def foo()(implicit ctx: Ctx.Owner, data: Ctx.Data) = {
  val a = rx.Var(1)
  a()
  a
}
val x = rx.Rx{val y = foo(); y() = y() + 1; println("done!") }
...
at rx.Rx$Dynamic$Internal$$anonfun$calc$2.apply(Core.scala:180)
  at scala.util.Try$.apply(Try.scala:192)
  at rx.Rx$Dynamic$Internal$.calc(Core.scala:180)
  at rx.Rx$Dynamic$Internal$.update(Core.scala:184)
  at rx.Rx$.doRecalc(Core.scala:130)
  at rx.Var.update(Core.scala:280)
  at $anonfun$1.apply(<console>:15)
  at $anonfun$1.apply(<console>:15)
  at rx.Rx$Dynamic$Internal$$anonfun$calc$2.apply(Core.scala:180)
  at scala.util.Try$.apply(Try.scala:192)
...

Data 上下文是 Rx 用于决定何时重新计算的机制。 Ownership修复了泄漏问题。混合两者可能会导致无限递归：当某些东西既拥有又具有同一父 Rx 的数据依赖性时。

幸运的是，几乎总是只需要一个或另一个上下文的情况。在处理动态图时，几乎总是只需要所有权上下文，即函数通常具有以下形式：

def f(...)(implicit ctx: Ctx.Owner) = Rx { ... }

Data 上下文的需要较少，并且在需要 DRY 某些重复的 Rx 代码的情况下很有用。这样的函数将具有以下形式：

def f(...)(implicit data: Ctx.Data) = ...

这将允许将一些共享数据依赖性从每个 Rx 的主体中拉出并放入共享函数中。

通过分解 ownership 和 data 依赖关系的正交概念，如上所述的无限递归问题得到极大限制。当使用 data 意味着数据依赖性而不仅仅是简单读取当前值（即 .now）时，显式 data 依赖性也使其更加清晰。如果没有这种区分，就更容易引入意料之外的“意外”数据依赖关系。

附加操作

除了Var/Rx/Obs的基本构建块之外，Scala.Rx还提供了一组组合器，可以让您轻松地转换Rx；这使得程序员可以避免不断地重写构建数据流图的常见方法的逻辑。五个基本组合器：map()、flatMap、filter()、reduce() 和 fold() 均以 scala 集合库为模型，并提供了一种转换来自 Rx 的值的简单方法。

地图

val a = Var(10)
val b = Rx{ a() + 2 }
val c = a.map(_*2)
val d = b.map(_+3)
println(c.now) // 20
println(d.now) // 15
a() = 1
println(c.now) // 2
println(d.now) // 6

map 执行您所期望的操作，使用通过某个函数转换的旧 Rx 的值创建一个新的 Rx。例如，a.map(_*2)本质上等同于Rx{ a() * 2 }，但写起来更方便一些。

FlatMap

val a = Var(10)
val b = Var(1)
val c = a.flatMap(a => Rx { a*b() })
println(c.now) // 10
b() = 2
println(c.now) // 20

flatMap 类似于集合库中的 flatMap，因为它允许将 Rx[Rx[_]] 类型的嵌套 Rx 合并到单个 Rx[_] 中。

这与映射组合器相结合，允许 scala 的 for 理解语法与 Rx 和 Var 一起使用：

val a = Var(10)
val b = for {
  aa <- a
  bb <- Rx { a() + 5}
  cc <- Var(1).map(_*2)
} yield {
  aa + bb + cc
}

过滤器

val a = Var(10)
val b = a.filter(_ > 5)
a() = 1
println(b.now) // 10
a() = 6
println(b.now) // 6
a() = 2
println(b.now) // 6
a() = 19
println(b.now) // 19

filter 忽略对谓词失败的 Rx 值的更改。

请注意，filter 方法都无法过滤掉 Rx 的第一个初始值，因为没有“旧”值可供回退。因此：

val a = Var(2)
val b = a.filter(_ > 5)
println(b.now)

将打印出“2”。

减少

val a = Var(1)
val b = a.reduce(_ * _)
a() = 2
println(b.now) // 2
a() = 3
println(b.now) // 6
a() = 4
println(b.now) // 24

reduce 运算符从初始值开始将 Rx 的后续值组合在一起。对原始 Rx 的每次更改都会与先前存储的值相结合，并成为减少的 Rx 的新值。

折叠

val a = Var(1)
val b = a.fold(List.empty[Int])((acc,elem) => elem :: acc)
a() = 2
println(b.now) // List(2,1)
a() = 3
println(b.now) // List(3,2,1)
a() = 4
println(b.now) // List(4,3,2,1)

Fold 能够以与减少类似的方式进行累积，但可以累积到源 Rx 之外的其他类型。

这五个组合器中的每一个在 .all 命名空间中都有一个对应项，它在 Try[T]s 而不是 Ts 上运行，以防您需要增加灵活性以某种特殊方式处理 Failures。

异步组合器 这些组合器的作用不仅仅是将一个值从一个值转换为另一个值。它们具有异步效果，并且可以自发地修改数据流图并开始传播周期，而无需任何外部触发。虽然这可能听起来有些令人不安，但这些组合器提供的功能通常是必要的，并且围绕 去抖动 等内容手动编写逻辑比简单地使用提供的组合器更容易出错。

请注意，这些组合器都没有做任何通过 Obs 和 Var 的组合无法完成的事情；它们只是封装了常见的模式，从而节省了您一遍又一遍地手动编写它们的麻烦，并减少了出现错误的可能性。

未来

import scala.concurrent.Promise
import scala.concurrent.ExecutionContext.Implicits.global
import rx.async._

val p = Promise[Int]()
val a = p.future.toRx(10)
println(a.now) //10
p.success(5)
println(a.now) //5

toRx 组合器仅适用于 Future[_]s。它需要一个初始值，该值将是Rx的值，直到Future完成，此时该值将成为Future的值。

这个async可以根据需要多次创建Futures。此示例显示它创建两个不同的 Futures：

import scala.concurrent.Promise
import scala.concurrent.ExecutionContext.Implicits.global
import rx.async._

var p = Promise[Int]()
val a = Var(1)

val b: Rx[Int] = Rx {
  val f =  p.future.toRx(10)
  f() + a()
}
println(b.now) //11
p.success(5)
println(b.now) //6

p = Promise[Int]()
a() = 2
println(b.now) //12

p.success(7)
println(b.now) //9

随着 Futures 系列完成（在本例中，手动使用 Promises），b() 的值将按照您的预期进行更新。

如果您的依赖关系图包含一些异步元素，这会很方便。例如，您可能有一个 Rx，它依赖于另一个 Rx，但需要异步 Web 请求来计算其最终值。使用 async，当 Future 完成时，异步 Web 请求的结果将自动推回到数据流图中，开始另一次传播运行，并根据新结果方便地更新图表的其余部分。

计时器

import rx.async._
import rx.async.Platform._
import scala.concurrent.duration._

val t = Timer(100 millis)
var count = 0
val o = t.trigger {
    count = count + 1
}

println(count) // 3
println(count) // 8
println(count) // 13

Timer 是定期生成事件的 Rx。在上面的示例中，在控制台中使用 println 显示值 t() 随着时间的推移而增加。

当Timer对象变得不可访问时，计划任务会自动取消，因此可以进行垃圾收集。这意味着您不必担心管理 Timer 的生命周期。另一方面，这意味着程序员应该确保对 Timer 的引用与保存侦听它的任何 Rx 的对象相同。这将确保 Timer 被垃圾收集的确切时刻并不重要，因为到那时持有它的对象（以及它可能影响的任何 Rx）都是无法访问的。

延迟

import rx.async._
import rx.async.Platform._
import scala.concurrent.duration._

val a = Var(10)
val b = a.delay(250 millis)

a() = 5
println(b.now) // 10
eventually{
  println(b.now) // 5
}

a() = 4
println(b.now) // 5
eventually{
  println(b.now) // 4
}

delay(t) 组合器创建 Rx 的延迟版本，其值滞后于原始值持续时间 t。当Rx改变时，延迟版本不会改变，直到延迟t过去。

此示例显示了应用于 Var 的延迟，但它也可以轻松应用于 Rx。

去抖动

import rx.async._
import rx.async.Platform._
import scala.concurrent.duration._

val a = Var(10)
val b = a.debounce(200 millis)
a() = 5
println(b.now) // 5

a() = 2
println(b.now) // 5

eventually{
  println(b.now) // 2
}

a() = 1
println(b.now) // 2

eventually{
  println(b.now) // 1
}

debounce(t) 组合器创建 Rx 的版本，每个时间段 t 更新不会超过一次。

如果在短时间内（间隔小于 t）发生多次更新，则第一次更新将立即发生，第二次更新将在 t 时间过去后发生。例如，这可用于限制重新计算昂贵结果的速率：如果它可以让您节省每隔几秒多次执行昂贵计算的费用，您可能愿意让计算值过时几秒钟。

设计考虑因素

使用简单 这意味着以依赖跟踪方式编写程序的语法必须尽可能轻量，使用 FRP 编写的程序必须“看起来”像它们正常的、老式的命令式对应程序。这意味着使用 DynamicVariable 而不是隐式自动传递参数，牺牲适当的词法范围来获得良好的语法。

我排除了使用纯粹的单子风格（如 反应式网络），因为虽然以这种方式实现该库会容易得多，但实际使用它会更加痛苦。我也不想手动声明依赖项，因为当您两次声明依赖项时，这违反了 DRY：一次在 Rx 的标头中，一次在主体中使用它时。

目标是能够编写代码，散布一些 Rx{}s 并让依赖项跟踪和更改传播正常工作。总的来说，我相信它在这方面非常成功！

易于推理 这意味着很多事情，但最重要的是它意味着没有全局变量。对于使用该库的人来说，这极大地简化了许多事情，因为您不再需要推理程序的不同部分通过该库进行交互。在大型程序的不同部分使用 Scala.Rx 是完全没问题的；他们是完全独立的。

该领域的另一个设计决策是将并行性和传播调度主要留给隐式 ExecutionContext，并默认在更新数据流图的任何线程上简单地运行传播波。

• 前者意味着任何习惯在 Scala/Akka 中编写并行程序的人都已经熟悉如何处理并行化 Scala.Rx
• 后者使得更容易推断传播何时发生，至少在默认情况下：它只是“立即”发生，并且当 Var.update() 函数返回时，传播已完成。

总体而言，限制副作用的范围并删除全局状态使 Scala.Rx 易于推理，并且意味着开发人员可以专注于使用 Scala.Rx 构建数据流图，而不用担心不可预测的深远交互或性能瓶颈。

易于互操作 这意味着程序员必须能够轻松进出 FRP 世界。我在准备 Scala.Rx 时读到的许多论文都描述了一些系统，这些系统本身运行得非常出色，并且具有一些令人惊奇的特性，但要求整个程序用一种晦涩语言的晦涩变体编写。根本没有考虑与现有语言或范例的互操作性，这使得不可能将 FRP 逐步引入现有代码库。

有了 Scala.Rx，我决定以不同的方式做事。因此，Scala.Rx：

• 用 Scala 编写：一种不常见，但可能比 Haskell 或 更晦涩的语言 方案
• 是一个库：它是普通的旧式 scala。没有源到源的转换，使用 Scala.Rx 不需要特殊的运行时。将源代码下载到 Scala 项目中，然后开始使用它
• 允许您在 Rx 中使用任何编程语言构造或库功能：Scala.Rx 将找出依赖关系，而程序员不必担心它，而不会将自己限制在该语言的某些不方便的子集上
• 允许您在较大的项目中使用 Scala.Rx，而不会有太大的痛苦。您可以轻松地将数据流图嵌入到更大的面向对象的宇宙中，并通过设置 Var 和 Obs 与它们交互

所审查的许多论文都展示了一个美丽的新 FRP 宇宙，只要您将所有代码移植到 FRP-Haskell 并将自己限制在用于创建数据流图的一小组组合器中，我们就可以在其中进行编程。另一方面，通过让您在现有代码中的任何位置嵌入 FRP 片段，在现有项目中使用 FRP 想法而无需完全承诺，并允许您在 FRP 和非 FRP 代码之间轻松互操作，Scala.Rx 旨在带来好处FRP 进入我们今天正在编程的肮脏、混乱的宇宙。

局限性 Scala.Rx 具有许多重大限制，其中一些来自设计中的权衡，另一些则来自底层平台的限制。

没有“空”反应

Scala.Rx 中 Rxs 的 API 尝试尽可能遵循集合 API：您可以对 Rxs 进行映射、过滤和缩减，就像对集合进行映射、过滤和缩减一样。然而，目前不可能像集合为空那样让 Rx 为空：过滤掉 Rx 中的所有值仍然至少会留下初始值（即使它未通过谓词），并且需要为异步 Rx 提供一个初始值才能启动。

这种限制是由于难以将可能为空的 Rx 与良好的用户体验连接在一起。例如，如果我有一个数据流图：

val a = Var()
val b = Var()
var c = Rx{
    .... a() ...
    ... some computation ...
    ... b() ...
    result
}

如果 a 和 b 最初是空的，我基本上有四个选项：

• 阻塞当前正在计算c的线程，等待a然后b变得可用。
• 当请求 a() 和 b() 时抛出异常，中止 c 的计算，但将其注册为在 a() 或 b() 可用时重新启动。
• 使用 for 推导式以一元样式重写此代码。
• 使用 delimited continuations 插件将上述代码自动转换为 Monadic 代码。

第一个选项是性能问题：线程在大多数操作系统上通常都非常重。您无法合理地创建超过几千个线程，与您可以创建的对象数量相比，这只是一个很小的数字。因此，虽然阻塞是最简单的，但它在许多系统中都不受欢迎（Akka 中的 e.g.，Scala.Rx 是基于它构建的），并且似乎不是一个好的解决方案。

第二个选项是不同方式的性能问题：对于 n 不同的依赖项，所有依赖项都可能从空开始，c 的计算可能需要启动和中止 n 次，甚至在完成一次之前。虽然这不会阻塞任何线程，但它看起来确实非常昂贵。

从用户体验的角度来看，第三个选项是不行的：它需要对代码库和编码风格进行深远的更改，以便从更改传播中受益，但我不愿意要求这样做。

由于分隔延续插件的错误，最后一个选项是有问题的。尽管从理论上讲它应该能够解决所有问题，但大量的小错误（搞乱类型推断、干扰隐式解析）以及一些基本问题意味着即使在小型项目（少于 1000 行响应式代码）中，使用它也会变得很痛苦。

开始时没有自动并行化

如前所述，Scala.Rx 可以对数据流图中发生的更新执行自动并行化：只需提供适当的 ExecutionContext，独立的 Rx 就会将其更新分布在多个内核上。

然而，这仅适用于更新，而不是在最初定义数据流图时有效：在这种情况下，每个 Rx 都会评估其主体一次以获得其默认值，并且这一切都在同一线程上串行发生。这种限制是由于我们没有一个好的方法来处理“空”Rxs，并且在第一次评估 Rxs 依赖项之前我们不知道它是什么。

因此，我们无法开始并行评估所有 Rx，因为有些 Rx 可能会先于它们所依赖的其他 Rx 完成，然后这些 Rx 将为空，但它们的初始值仍在计算中。我们也不能选择并行化那些彼此不依赖的对象，因为在执行之前我们不知道依赖项是什么！

因此，我们别无选择，只能让 Rx 的初始定义连续发生。如有必要，程序员可以使用 Futures 并行手动创建独立的 Rxs 。

故障和冗余计算

在 FRP 的上下文中，故障是数据流图中的暂时不一致。由于更新不会立即发生，而是需要时间来计算，因此 FRP 系统内的值在更新过程中可能会暂时不同步。此外，根据 FRP 系统的性质，节点可能在传播中更新多次。

这可能是问题，也可能不是问题，具体取决于应用程序对偶尔出现的陈旧不一致数据的容忍程度。在单线程系统中，可以通过多种方式来避免

• 使数据流图静态，并执行拓扑排序以按照要更新的顺序对节点进行排名。这意味着节点总是在其依赖项之后*更新，这意味着它们永远不会看到任何过时的数据
• 当节点尝试调用尚未更新的依赖项时，暂停节点的更新。例如，这可以通过阻塞线程并仅在依赖项更新后恢复来完成。

然而，这两种方法都存在问题。第一种方法非常严格：静态数据流图意味着大量有用的行为，e.g。禁止在运行时动态创建和销毁图的部分。这违背了 Scala.Rx 的目标，即允许程序员不受限制地“正常”编写代码，并让 FRP 系统自行解决。

第二种情况对于不允许暂停计算的语言来说是一个问题。在 Java 以及扩展的 Scala 中，使用的线程是操作系统 (OS) 线程，这是非常昂贵的。因此，阻塞 OS 线程是不受欢迎的。协程和延续也可以用于此目的，但 Scala 缺乏这两种功能。

最后一个问题是，这两种模型仅在单线程、顺序代码的情况下才有意义。正如并发和并行部分提到的，Scala.Rx 允许您使用多个线程并行化传播，并允许多个线程同时启动传播。这意味着严格禁止故障是不可能的。

Scala.Rx 维护稍微宽松的模型：每个 Rx 的主体每次传播可能会被评估多次，并且 Scala.Rx 只承诺做出“尽力而为”的尝试来减少冗余更新的数量。假设每个 Rx 的主体是纯的，这意味着冗余更新应该只影响传播完成所需的时间和计算，而不会影响传播完成后每个节点的值。

此外，Scala.Rx 提供了Obs，它们是特殊的终端节点，保证每次传播仅更新一次，旨在产生一些副作用。这意味着，虽然传播可能会导致数据流图中的 Rx 的值暂时不同步，但只有在整个传播完成且 Obs 全部激发其副作用时，传播的最终副作用才会发生。

如果多个传播并行发生，Scala.Rx 保证每个 Obs 每次传播“最多”一次，并且总体“至少”一次。此外，在整个数据流图稳定且传播完成后，每个 Obs 将至少触发一次。这意味着，例如，如果您依靠 Obs 通过网络向远程客户端发送更新，则可以确保不会通过网络传输任何不必要的信息，并且当系统处于静止状态时，远程客户端将获得代表数据流图最新版本的更新。

相关工作

Scala.Rx 并不是凭空产生的，它借鉴了一系列现有项目的想法和灵感。

Scala.React Scala.React，如 弃用观察者模式 中所述，包含反应式更改传播部分（称为 Signals），这与 Scala.Rx 的做法类似。然而，它的作用远不止于此：它包含使用事件流的实现，以及使用分隔延续的多个 DSLs，以便轻松编写异步工作流程。

我使用过这个库，我的经验是它的设置和上手极其困难。它需要大量的全局配置，由全局引擎进行调度和传播，甚至运行自己的线程池。这使得推理程序各部分之间的交互变得极其困难：完全独立的数据流图是否能够通过这个全局引擎相互影响？随着线程数量的增加，多线程代码的性能是否会开始变慢，因为引擎成为瓶颈？我从未找到其中许多问题的答案，也未能联系到作者。

全球传播引擎也让上手变得困难。花了几天时间才得到一个基本的数据流图（类似于本文档顶部的示例）。这是经过大量的努力、阅读相关论文数十次并以我不理解的方式破解源代码之后的结果。不用说，这些都不是我有信心建立的基础。

反应式网络 反应网络 是另一个灵感。它与 Scala.Rx 有点正交，更多地关注事件流以及与 Web 框架的集成，而 Scala.Rx 纯粹关注时变值。

然而，reactive-web 带有自己的时变值（称为信号），这些值是使用类似于 Scala.Rx 中的组合器（map、filter、flatMap 等）进行操作的。然而，reactive-web 并没有提供一种简单的方法来组合这些信号：程序员必须完全依赖 map 和 flatMap，可能使用 Scala 的 for 推导式。

我不喜欢这样的事实：您必须以一元风格进行编程（i.e。始终生活在 .map() 和 .flatMap() 和 for{} 理解中）才能利用更改传播。这在嵌套Rxs的情况下尤其麻烦，其中Scala.Rx的

// a b and c are Rxs
x = Rx{ a() + b().c() }

变成

x = for {
  va <- a
  vb <- b
  vc <- vb.c
} yield (va + vc)

正如您所看到的，在反应式网络中使用 for 推导式会导致代码明显更长且更加混乱。

Knockout.js Knockout.js 对 Javascript 做了类似的事情，还有一些其他额外的好处，比如 DOM- 绑定。事实上，自动依赖跟踪的设计和实现以及开发人员体验几乎是相同的。这个：

this.firstName = ko.observable('Bob');
this.lastName = ko.observable('Smith');
fullName = ko.computed(function() {
    return this.firstName() + " " + this.lastName();
}, this);

在语义上等价于以下 Scala.Rx 代码：

val firstName = Var("Bob")
val lastName = Var("Smith")
fullName = Rx{ firstName() + " " + lastName() }

ko.observable 直接映射到 Var，kocomputed 直接映射到 Rx。除了更长的变量名和 Javascript 的冗长之外，语义几乎是相同的。

除了提供Var和Rx的等价物之外，Knockout.js还致力于不同的方向。它缺少 Scala.Rx 提供的大部分有用组合器，但提供了大量其他功能，例如与浏览器的 DOM 集成，这是 Scala.Rx 所缺少的。

其他 这种变化传播的想法，具有随时间变化的值，当发生变化时通知任何依赖于它们的值，这是 功能反应式编程 领域的一部分。这是一个经过充分研究的领域，已经完成了大量研究。 Scala.Rx 以此研究为基础，并融合了以下项目的想法：

• FlapJax
• 冰沙
• 弗兰

所有这些项目都充满了好主意。然而，一般来说，它们通常是非常多的研究项目：为了换取 FRP 的好处，它们要求您用一种晦涩语言的晦涩变体编写整个程序，与现有的非 FRP 代码进行互操作的希望很小。

以不熟悉的范例（例如 FRP）编写生产软件已经是一个很大的风险。最重要的是，用不熟悉的语言编写生产软件是一个额外的变量，并且用不熟悉的语言以不熟悉的范式编写生产软件与现有代码没有互操作性是彻头彻尾的鲁莽。因此，这些库没有得到大量使用也就不足为奇了。 Scala.Rx 旨在通过以熟悉的语言提供 FRP 的优点以及 FRP 和更传统的命令式或面向对象代码之间的无缝互操作来解决这些问题。

版本历史

0.4.0

• 放弃 Scala 2.10 支持
• 添加了双向 Vars 和朋友
• 修复了多个接收“故障”

0.3.2

• 升级到 Scala 2.12.0。

0.3.1

• 修复了观察者的泄漏（他们还需要拥有上下文）。

• 修复了 flatMap 的类型问题

0.3.0

• 引入了 Owner 和 Data 上下文。这是依赖关系和生命周期管理的完全不同的实现，允许安全构建运行时动态图。

• 更多默认组合器：fold 和 flatMap 现在默认实现。

制作人员

版权所有 (c) 2013，李浩仪（haoyi.sg at gmail.com）

特此免费授予获得本软件及相关文档文件（“软件”）副本的任何人无限制地处理本软件，包括但不限于使用、复制、修改、合并、发布、分发、再许可、and/or 销售本软件副本的权利，并允许向其提供本软件的人员这样做，但须满足以下条件：

上述版权声明和本许可声明应包含在本软件的所有副本或主要部分中。

THE SOFTWARE IS PROVIDED“AS IS”，WITHOUT WARRANTY OF ANY KIND、EXPRESS OR IMPLIED、INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY、FITNESS FOR PARTICULAR PURPOSE AND NONINFRINGEMENT。 IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM，DAMAGES OR OTHER LIABILITY，WHETHER IN AN ACTION OF CONTRACT，TORT OR OTHERWISE、ARISING FROM、OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE。