我怎样才能让 scala 的阻塞与 tailrec 一起工作？答案

【问题标题】：How can I get scala's blocking to work with tailrec?我怎样才能让 scala 的阻塞与 tailrec 一起工作？
【发布时间】：2013-08-24 07:01:28
【问题描述】：

我正在为 Futures 编写 scala java 互操作包装器，但我不知道实现 scala.concurrent.Future.onComplete (http://www.scala-lang.org/api/current/index.html#scala.concurrent.Future) 的正确方法。这可能有效：

def onComplete[U](func: Try[T] => U)(implicit executor: ExecutionContext): Unit = {
  executor.execute(new Runnable {
    @tailrec
    def run = value match {
      case Some(t) => func(t)
      case None => { Thread.sleep(100); run }
    }
  })
}

但是Asynchronous IO in Scala with futures 建议当我必须阻止时，我应该将代码的相关部分传递给 scala.concurrent.blocking 让 ExecutionContext 知道发生了什么。问题是当我用阻塞 {} 包围值 match{...} 时，它不再是尾调用。

众所周知的正确方法是什么？

编辑：为了完整起见，这里是整个包装类：

class JavaFutureWrapper[T](val jf: java.util.concurrent.Future[T]) extends scala.concurrent.Future[T] {
  def isCompleted = jf.isDone

  def result(atMost: Duration)(implicit permit: CanAwait): T =
    atMost match { case Duration(timeout, units) => jf.get(timeout, units) }

  def onComplete[U](func: Try[T] => U)(implicit executor: ExecutionContext): Unit = {
    executor.execute(new Runnable {
      @tailrec
      def run = value match {
        case Some(t) => func(t)
        case None => { Thread.sleep(100); run }
      }
    })
  }

  def ready(atMost: Duration)(implicit permit: CanAwait): this.type = atMost match {
    case Duration(timeout, units) => {
      jf.get(timeout, units)
      this
    }
  }

  def value: Option[Try[T]] = (jf.isCancelled, jf.isDone) match {
    case (true, _) => Some(Failure(new Exception("Execution was cancelled!")))
    case (_, true) => Some(Success(jf.get))
    case _ => None
  }
}

【问题讨论】：

你能指出什么是Java API吗？例如，value 来自哪里？为什么要让线程休眠 100 毫秒？换句话说，您的潜在阻塞代码在哪里？
我已经为你粘贴了整个包装类。睡眠是为了防止紧密循环，因为我轮询 Java 未来是否已经完成。整个 run 方法将阻塞，直到 jf 决定它是完成还是取消。
如果你只用阻塞包围 Thread.sleep() 怎么办？另一个想法是使用一段时间而不是递归调用。
哎呀，我没想到只围着睡觉，我想我们可能有一个赢家。 ExecutionContexts 应该足够聪明，可以处理快速退出然后又回到阻塞部分，对吧？

标签： scala java.util.concurrent

【解决方案1】：

我会等待 Java 的未来完成：

import scala.util.{Try, Success, Failure}
import scala.concurrent._
import java.util.concurrent.TimeUnit

class JavaFutureWrapper[T](val jf: java.util.concurrent.Future[T])
  extends scala.concurrent.Future[T] {
  ...

  def onComplete[U](func: Try[T] => U)(implicit executor: ExecutionContext): Unit =
    executor.execute(new Runnable {
      def run: Unit = {
        val result = Try(blocking(jf.get(Long.MaxValue, TimeUnit.MILLISECONDS)))
        func(result)
      }
    })
  ...
}

【讨论】：

哦，对，是的，这显然是正确的答案，谢谢！我已向您发送了一个简化代码的编辑，一旦完成，我会将其标记为答案。
编辑改变了含义。您将blocking 放在func 的执行周围。但是func 有责任声明它是否正在做一些阻塞或不做的事情。所以我认为这个版本是正确的，只有在等待jf完成时才阻止。
好点，这是另一个没有这个问题的编辑版本

【解决方案2】：

嗯，我对 0__ 答案的编辑没有得到批准，所以为了未来的读者，这是我要使用的解决方案（从 0__ 简化）

def onComplete[U](func: Try[T] => U)(implicit executor: ExecutionContext): Unit = {
  executor.execute(new Runnable {
    def run = func(Try( blocking { jf.get } ))
  })
}

【讨论】：

是的，.get 没有超时似乎无限期地等待。
无限期而不是大约 3 亿年？是的，不知怎的，我可以接受这种权衡。另外，如果 3 亿年后未来仍未到来，我希望 onComplete 不会表示完成。