java.util.concurrent.Future 的 scala.concurrent.Future 包装器

Question

我将 Play Framework 2.1.1 与一个生成 java.util.concurrent.Future 结果的外部 java 库一起使用。我使用的是 scala 未来而不是 Akka，我认为从 Play 2.1 开始这是正确的做法。如何将 java.util.concurrent.Future 包装到 scala.concurrent.Future 中，同时保持代码非阻塞？

def geConnection() : Connection = {
  // blocking with get
  connectionPool.getConnectionAsync().get(30000, TimeUnit.MILLISECONDS)
}

上面的代码返回一个连接，但是使用了一个get，所以它变成了阻塞

def getConnectionFuture() : Future[Connection] = {
  future {
    // how to remove blocking get and return a scala future?
    connectionPool.getConnectionAsync().get(30000, TimeUnit.MILLISECONDS)
  }
}

理想情况下，我想要一个 scala 函数，它可以像上面的代码一样将连接作为未来返回，但不会通过 get 阻塞代码。我还需要在函数中添加什么以使其非阻塞。

任何指针都会很棒。

Answer 1

import java.util.concurrent.{Future => JFuture}
import scala.concurrent.{Future => SFuture}

你不能在不阻塞的情况下用SFuture 包装JFuture，因为SFuture (onComplete) 中有一个回调，而JFuture 中只有阻塞get。

你所能做的就是创建额外的线程并用get阻止它，然后用get的结果完成Promise。

val jfuture: JFuture[T] = ???
val promise = Promise[T]()
new Thread(new Runnable { def run() { promise.complete(Try{ jfuture.get }) }}).start
val future = promise.future

你可以在无限循环中检查isDone，但我认为它并不比阻塞更好。

Answer 2

Future {
  blocking {
    jfuture.get
  }
}

这让 ExecutionContext 知道你正在做的事情将被阻塞，让它有机会分配更多的线程。如果你不包含blocking { }，那么你可能会用完线程。

Answer 3

scala-java8-compat 库提供了 java8 和 Scala Futures 之间的转换器。

具体可以使用FutureConverters.toScala(connectionPool.getConnectionAsync())将java.util.concurrent.Future转换为scala.concurrent.Future

Answer 4

     import java.util.concurrent.{Future => JFuture}
     import scala.concurrent.ExecutionContext.Implicits.global
     import scala.concurrent.Future
     import scala.util.Try

     object JFuture2SFuture {
        val jFuture: JFuture[Int] = ???
        val promise = Promise[Int]()
        Future { promise.complete(Try(jFuture.get)) } //it is non blocking 
        val sFuture:Future[Int] = promise.future

     }