Scala 中的 TraversableOnce、Future 和 Option 用于理解答案

【问题标题】：TraversableOnce, Future, and Option in a Scala for comprehensionScala 中的 TraversableOnce、Future 和 Option 用于理解
【发布时间】：2013-02-15 17:58:49
【问题描述】：

我有一个代表数据库记录的字符串 id 列表。我想从数据库异步加载它们，然后将每条记录异步上传到远程服务器，然后当所有上传完成后，记录上传记录的ID。

由于我使用的是 Scala 2.9.2，因此我使用的是 Twitter 的 core-util Future 实现，但它在 Monadic 转换方面应该与 2.10 的 future 完全一样。

一般概念是这样的：

def fetch(id: String): Future[Option[Record]]
def upload(record: Record): Future[String]
def notifyUploaded(ids: Seq[String]): Unit

val ids: Seq[String] = ....

我正在尝试通过 for 理解来做到这一点，但 fetch 返回 Option 的 Future 的事实使它变得晦涩难懂并且代码无法编译：

for {
  id <- ids
  maybeRecord <- fetch(id)
  record <- maybeRecord
  uploadedId <- upload(record)
} yield uploadedId

编译它会导致以下错误：

scala: type mismatch;
found   : com.twitter.util.Future[String]
required: Option[?]
    uploadedId <- upload(record)
                  ^

我错过了什么？为什么编译器期望 uploadId 是一个选项？有什么漂亮的方法可以解决这个问题吗？

【问题讨论】：

Type Mismatch on Scala For Comprehension 的可能重复项
Monad 是 Endo 类中的 Monoid。只是说'
@folone：我担心不是每个人都会得到joke。只是说'
@folone 但问题是自然转换是从两个内函子到自身的乘积，而不是它们本身就是内函子。
@folone 因此，每个 monad 都是 Scala 类型类别上的 endofunctors 类别中的一个对象。在该类别中，态射是函子之间的自然转换，而被破坏的是“flatMap/bind”的自然转换来自M x M -> M，而组合它就像M x T -> TM（例如函子。）问题是不能保证（而且通常不是真的）组合本身就是一个单子。

标签： scala monads future

【解决方案1】：

考虑flatMap（或绑定）函数的签名：

trait Monad[M[_]] {
  def flatMap[A](a : M[A], f : A => M[B]) : M[B]
  ....

在您的情况下，您尝试在Option 上使用flatMap，给它一个f，生成一个Future。但就像上面的签名一样，f 应该在它被调用的同一个 monad 中生成一些东西。

Scala 在这方面不一定很有帮助，因为它非常擅长转换周围的事物（例如，转换为 Seqs），这样您就会得到可以链接任意 flatMap 调用的印象一起，不管容器。

你可能想要的是一个“Monad 转换器”，它可以让你有一些编写 monad 的能力。 Debasish Ghosh 有一篇关于使用 Scalaz monad 转换器的帖子here。

【讨论】：

【解决方案2】：

您不能将所有不同的类型混合在一起以进行理解，我发现您可能会混合 Seq 和 Option 并且结果将是 Seq 或 Option 取决于第一个。不能混合 Future 和 Seq 或 Option。如果您想使用理解功能，则必须将它们级联一些。在这种情况下，使用 map/flatMap 可能会更好。我以两种方式实现了您的问题，并为少数中间结果添加了类型，以便您看到在使用所有不同类型时产生的混乱。

object TestClass {

  import scala.concurrent.Future
  import scala.concurrent.ExecutionContext.Implicits.global
  import scala.concurrent._
  import scala.concurrent.duration._

  case class Record(id: String)


  def fetch(id: String): Future[Option[Record]] = Future {
    Thread.sleep(1000);
    Some(Record(id))
  }

  def upload(record: Record): Future[String] = Future {
    Thread.sleep(3000);
    record.id + "_uploaded"
  }

  def notifyUploaded(ids: Seq[String]): Unit = println("notified" + ids)

  val ids: Seq[String] = Seq("a", "b", "c")

  def main(args: Array[String]): Unit = {
    forComprehensionImpl()
    mapAndFlatMapImpl()
  }

  def forComprehensionImpl() = {
    val result: Seq[Future[Option[Future[String]]]] = for {
      id <- ids
    } yield {
      for {
        maybeRecord <- fetch(id)
      } yield {
        for {
          record <- maybeRecord
        } yield {
          for {
            uploadedId <- upload(record)
          } yield uploadedId
        }
      }
    }
    val result2: Future[Seq[Option[Future[String]]]] = Future.sequence(result)
    val result3: Future[Unit] = result2.flatMap { x: Seq[Option[Future[String]]] =>
      Future.sequence(x.flatten).map(notifyUploaded)
    }
    Await.result(result3, Duration.Inf)
  }


  def mapAndFlatMapImpl() = {
    val res: Seq[Future[Iterable[String]]] = ids.map { id =>
      fetch(id).flatMap { maybeRecord =>
        val res1: Option[Future[Seq[String]]] = maybeRecord.map { record =>
          upload(record) map (Seq(_))
        }
        res1 match {
          case Some(a) => a
          case None => Future(Seq())
        }
      }
    }
    val res3: Future[Unit] = Future.sequence(res) map (a => notifyUploaded(a.flatten))
    Await.result(res3, Duration.Inf)
  }
}

【讨论】：