10

我想实现 ARM（自动资源管理）模式，其中资源是异步使用的。

问题

假设我的资源如下所示：

class MyResource { 
  def foo() : Future[MyResource] = ???
  // Other methods returning various futures
  def close() : Unit = ???
}
object MyResource { 
  def open(name: String): Future[MyResource] = ???
}

所需的使用模式是：

val r : Future[MyResource] = MyResource.open("name")
r flatMap (r => {
  r.foo() /* map ... */ andThen {
    case _ => r.close()
  }
})

省略的映射函数可能很复杂，涉及分支和链接期货，这些期货重复调用r返回期货的方法。

我想确保在所有未来的延续完成（或失败）r.close()之后被调用。在每个呼叫站点手动执行此操作很容易出错。这需要 ARM 解决方案。

尝试的解决方案

scala-arm 库通常是同步的。这段代码不会做正确的事情，因为 close() 将在块内的期货完成之前被调用：

for (r <- managed(MyResource.open("name"))) {
  r map (_.foo()) // map ...
}

我虽然使用这个包装器：

def usingAsync[T](opener: => Future[MyResource]) (body: MyResource => Future[T]) : Future[T] =
  opener flatMap { 
    myr => body(myr) andThen { case _ => myr.close() } }

然后呼叫站点将如下所示：

usingAsync(MyResource.open("name")) ( myr => {
  myr.foo // map ...
})

但是，块内的代码将负责返回一个 Future，该 Future 在该块创建的所有其他期货完成时完成。如果它不小心没有，那么在所有使用它的期货完成之前，该资源将再次关闭。并且不会有静态验证来捕获此错误。例如，这将是一个运行时错误：

usingAsync(MyResource.open("name")) ( myr => {
  myr.foo() // Do one thing
  myr.bar() // Do another
})

如何解决这个问题？

显然，我可以使用 scala-arm 的定界延续支持 (CPS)。它看起来有点复杂，我害怕弄错。它需要启用编译器插件。另外，我的团队对 scala 非常陌生，我不想要求他们使用 CPS。

CPS 是唯一的出路吗？是否有一个库或设计模式可以使用 Futures 更简单地做到这一点，或者有一个使用 scala-arm 的例子？

4

1 回答 1

2

反应式扩展 (Rx) 可能是一种替代解决方案。围绕这种编程范式的势头越来越大，现在可以在包括 Scala 在内的许多语言中使用。

Rx 的基础是创建一个 Observable，它是异步事件的来源。Observable 可以以复杂的方式链接起来，这就是它的力量。你订阅一个 Observable 来监听 onNext、onError 和 onComplete 事件。您还会收到一个允许您取消的订阅。

我认为您可能会在 onCompleted 和/或 onError 处理程序中添加一个 resource.close() 调用。

请参阅 RxScala 文档了解：

Observable.subscribe(
    onNext: (T) ⇒ Unit, 
    onError: (Throwable) ⇒ Unit, 
    onCompleted: () ⇒ Unit): Subscription

更多信息：

于 2013-12-21T18:45:44.493 回答