将阻塞代码转换为使用Scala Future [英] Converting blocking code to using scala futures

问题描述

我的旧代码如下所示，其中所有数据库调用均阻塞.

My old code looks something like below, where all db calls blocking.

我需要帮助将其转换为使用期货.

I need help converting this over to using Futures.

def getUserPoints(username: String): Option[Long]
    db.getUserPoints(username) match {
        case Some(userPoints) => Some(userPoints.total)
        case None => {
            if (db.getSomething("abc").isEmpty) {
                db.somethingElse("asdf") match {
                    case Some(pointId) => {
                        db.setPoints(pointId, username)
                        db.findPointsForUser(username)
                    }
                    case _ => None
                }
            } else {
                db.findPointsForUser(username)
            }
        }       
    }
}

我的新API在我返回期货的位置下方.

My new API is below where I am returning Futures.

db.getUserPoints(username: String): Future[Option[UserPoints]]
db.getSomething(s: String): Future[Option[Long]]
db.setPoints(pointId, username): Future[Unit]
db.findPointsForUser(username): Future[Option[Long]]

如何将上述内容转换为使用新的使用期货的API.

How can I go about converting the above to use my new API that uses futures.

我尝试使用for-compr，但是开始遇到诸如Future [Nothing]之类的奇怪错误.

I tried using a for-compr but started to get wierd errors like Future[Nothing].

var userPointsFut: Future[Long] = for {
  userPointsOpt <- db.getUserPoints(username)
  userPoints <- userPointsOpt
} yield userPoints.total

但是对于所有分支和if子句并尝试将其转换为期货，它会变得有些棘手.

But it gets a bit tricky with all the branching and if clauses and trying to convert it over to futures.

推荐答案

我认为这种设计的第一个问题是对 Future 的阻塞调用的端口不应包装Option类型:

I would argue that the first issue with this design is that the port of the blocking call to a Future should not wrap the Option type:

阻止呼叫: def GiveMeSoSomethingBlocking(for:Id):选项[T] 应成为: def GiveMeSoSomethingBlocking(for:Id):Future [T] 还有不是: def GiveMeSoSomethingBlocking(for:Id):Future [Option [T]]

The blocking call: def giveMeSomethingBlocking(for:Id): Option[T] Should become: def giveMeSomethingBlocking(for:Id): Future[T] And not: def giveMeSomethingBlocking(for:Id): Future[Option[T]]

阻塞调用给出的值是 Some(value)或 None ，非阻塞Future版本给出的是 Success(value)或 Failure(exception)(故障(例外))以非阻塞方式完全保留 Option 语义.

The blocking call give either a value Some(value) or None, the non-blocking Future version gives either a Success(value) or Failure(exception) which fully preserves the Option semantics in a non-blocking fashion.

考虑到这一点，我们可以在 Future 上使用组合器对有问题的流程进行建模.让我们看看如何:

With that in mind, we can model the process in question using combinators on Future. Let's see how:

首先，让我们将API重构为可以使用的东西:

First, lets refactor the API to something we can work with:

type UserPoints = Long
object db {
  def getUserPoints(username: String): Future[UserPoints] = ???
  def getSomething(s: String): Future[UserPoints] = ???
  def setPoints(pointId:UserPoints, username: String): Future[Unit] = ???
  def findPointsForUser(username: String): Future[UserPoints] = ???
}
class PointsNotFound extends Exception("bonk")
class StuffNotFound extends Exception("sthing not found")

然后，该过程将如下所示:

Then, the process would look like:

def getUserPoints(username:String): Future[UserPoints] = {
  db.getUserPoints(username)
  .map(userPoints => userPoints /*.total*/)
  .recoverWith{ 
    case ex:PointsNotFound => 
    (for {
      sthingElse <- db.getSomething("abc")
      _ <- db.setPoints(sthingElse, username)
      points <- db.findPointsForUser(username)
    } yield (points))
    .recoverWith{
      case ex: StuffNotFound => db.findPointsForUser(username)
    }
  }
}

正确检查类型.

鉴于API是一成不变的，所以处理嵌套的monadic类型的一种方法是定义MonadTransformer.简而言之，让我们将 Future [Option [T]] 变成一个新的monad，将其称为 FutureO ，它可以与其他同类组合.[1]

Given that the API is set in stone, a way to deal with nested monadic types is to define a MonadTransformer. In simple words, let's make Future[Option[T]] a new monad, let's call it FutureO that can be composed with other of its kind. [1]

case class FutureO[+A](future: Future[Option[A]]) {
  def flatMap[B](f: A => FutureO[B])(implicit ec: ExecutionContext): FutureO[B] = {
    val newFuture = future.flatMap{
      case Some(a) => f(a).future
      case None => Future.successful(None)
    }
    FutureO(newFuture)
  }

  def map[B](f: A => B)(implicit ec: ExecutionContext): FutureO[B] = {
    FutureO(future.map(option => option map f))
  }
  def recoverWith[U >: A](pf: PartialFunction[Throwable, FutureO[U]])(implicit executor: ExecutionContext): FutureO[U] = {
    val futOtoFut: FutureO[U] => Future[Option[U]] = _.future
    FutureO(future.recoverWith(pf andThen futOtoFut))
  }

  def orElse[U >: A](other: => FutureO[U])(implicit executor: ExecutionContext): FutureO[U] = {
      FutureO(future.flatMap{
        case None => other.future
        case _ => this.future
      }) 
    }
  }

现在，我们可以重新编写流程，并保留与基于未来的组合相同的结构.

And now we can re-write our process preserving the same structure as the future-based composition.

type UserPoints = Long 
object db { 
  def getUserPoints(username: String): Future[Option[UserPoints]] = ???
  def getSomething(s: String): Future[Option[Long]] = ???
  def setPoints(pointId: UserPoints, username:String): Future[Unit] = ???
  def findPointsForUser(username: String): Future[Option[Long]] = ???
}
class PointsNotFound extends Exception("bonk")
class StuffNotFound extends Exception("sthing not found")

def getUserPoints2(username:String): Future[Option[UserPoints]] = {
  val futureOpt = FutureO(db.getUserPoints(username))
  .map(userPoints => userPoints /*.total*/)
  .orElse{ 
    (for {
      sthingElse <- FutureO(db.getSomething("abc"))
      _ <- FutureO(db.setPoints(sthingElse, username).map(_ => Some(())))
      points <- FutureO(db.findPointsForUser(username))
    } yield (points))
    .orElse{
      FutureO(db.findPointsForUser(username))
    }
  }
  futureOpt.future
}

[1]，带有对 http://loicdescotte.github.io的致谢/posts/scala-compose-option-future/

这篇关于将阻塞代码转换为使用Scala Future的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！