我想实现 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()
}
})
省略的映射函数可能很复杂,涉及分支和链接 future ,重复调用
r
的方法。那个返回 future 。我想确定
r.close()
在所有 future 的延续完成(或失败)后调用。在每个调用站点手动执行此操作很容易出错。这需要 ARM 解决方案。尝试的解决方案
scala-arm 库通常是同步的。这段代码不会做正确的事情,因为 close() 会在块内的 future 完成之前被调用:
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。如果它不小心没有,那么在使用它的所有 future 完成之前,资源将再次关闭。并且不会有静态验证来捕获此错误。例如,这将是一个运行时错误:
usingAsync(MyResource.open("name")) ( myr => {
myr.foo() // Do one thing
myr.bar() // Do another
})
如何解决这个问题?
显然,我可以使用 scala-arm 的定界延续支持 (CPS)。看起来有点复杂,我怕弄错。它需要启用编译器插件。此外,我的团队对 Scala 非常陌生,我不想要求他们使用 CPS。
CPS 是唯一的出路吗?是否有一个库或设计模式可以更简单地使用 Futures 来做到这一点,或者有一个使用 Scala-arm 做到这一点的例子?
最佳答案
响应式(Reactive)扩展 (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
更多信息:
关于scala - 如何在 Futures 中使用 Scala ARM?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/20720708/