使用“回滚”代码平面映射未来

Question

我很难找到一个优雅的解决方案来链接一些期货。我试图实现的方法看起来像这样（它是缓存的一部分）：

def acquire(key: A, producer: => Future[B]): Future[B]

算法大致如下：

• 如果密钥被锁定，则立即通过运行时异常（使用 Future.failed 是否会提前？
• 否则打开一个将来的 { getOrRefresh } 块，因为它需要一些时间来检索密钥
• getOrRefresh要么返回一个直B，然后返回未来和方法的结果
• 或者它需要运行生产者

最后一步意味着我需要从未来内部对未来进行“扁平化”。也就是说，我不能做一个outer.flatMap，所以我想策略是使用等待。

现在，< code>Await有一种奇怪的分裂，即您可以使用方法< code>ready获得< code >选项[Try[B]]，或者使用< code>result获得展开的< code>B。问题是，在完成外部未来之前，我需要在< code>Failure的情况下释放一个锁，所以我想我必须坚持< code>Await.ready。

这很难看：

val fut   = producer
val prod  = Await.ready(fut, Duration.Inf).value.get
if (prod.isFailure) sync.synchronized { locked = false }
prod.get

这会有那么难看吗？必须有更好的方法来做到这一点。

因此，重复一下：从＜code＞未来[B]中运行一些同样使用＜code＞B＜code＞完成的对等未来，并返回对等结果，但如果失败，请在完成主未来之前清除锁。

Answer 1

我对您的实现做了一些更改。

首先，我已经将您最初的＜code＞抛出锁定到＜code＞未来。失败，因为期货消费者应该可以安全地认为，唯一需要注意的失败是失败的未来。

接下来，我不是在readEntry的Option[B]上调用get，而是将其作为未来的结果返回。然后我平面地图结果，以便我可以在无的情况下替换生产者产生的未来（map将导致Future[Future[B]]]）。在一些的情况下，我从值返回Future.successful，因为平面地图需要返回未来。

最后，我已经将恢复和抛出替换为然后，因为我们希望失败的Future传播，并且只想将副作用链接到未来以解锁条目。

trait Cache[A, B] {
  class Entry(var locked: Boolean = true)

  private var map  = Map.empty[A, Entry]
  private val sync = new AnyRef

  def readEntry(key: A): Option[B] = ???

  def acquire(key: A, producer: => Future[B]): Future[B] = sync.synchronized {
    map.get(key) match {
      case Some(e) =>
        if (e.locked)
          Future.failed(new IllegalStateException())
        else {
        e.locked = true
        future { readEntry(key)}.flatMap {
          case None => producer.andThen {
            case Failure(_) => sync.synchronized(e.locked = false)
          }
          case Some(value) => Future.successful(value)
        }
      }

      case _ => producer.map { value =>
        sync.synchronized {
          map += key -> new Entry()
        }
        value
      }
    }
  }
}

Answer 2

这里尝试使用 recover（With），从而避免等待。不过，它看起来很笨拙，因为我需要重新抛出这个例外。

import concurrent._

trait Cache[A, B] {
  class Entry(var locked: Boolean = true)

  private var map  = Map.empty[A, Entry]
  private val sync = new AnyRef

  implicit def exec: ExecutionContext

  def readEntry(key: A): Option[B]

  def acquire(key: A, producer: => Future[B]): Future[B] = sync.synchronized {
    map.get(key) match {
      case Some(e) =>
        if (e.locked) throw new IllegalStateException()
        e.locked     = true
        val existing = future { readEntry(key).get }
        val refresh  = existing.recoverWith {
          case _: NoSuchElementException => producer
        }
        refresh.recover {
          case t => sync.synchronized(e.locked = false); throw t
        }

      case _ => producer.map { value =>
        sync.synchronized {
          map += key -> new Entry()
        }
        value
      }
    }
  }
}

如果您有建议，请将它们作为单独的答案发布。