How are Scala Futures chained together with flatMa

问题:

I'm working on using Futures for the first time in Scala and am working through an example of using the flatMap combinator; I've been following this discussion:

http://docs.scala-lang.org/overviews/core/futures.html

Specifically, this example:

val usdQuote = future { connection.getCurrentValue(USD) }
val chfQuote = future { connection.getCurrentValue(CHF) }
val purchase = for {
    usd <- usdQuote
    chf <- chfQuote
      if isProfitable(usd, chf)
} yield connection.buy(amount, chf)

purchase onSuccess {
    case _ => println("Purchased " + amount + " CHF")
}

is translated to this:

val purchase = usdQuote flatMap {
    usd =>
         chfQuote
        .withFilter(chf => isProfitable(usd, chf))
        .map(chf => connection.buy(amount, chf))
}

What I'm having a bit of trouble grasping is how and when this is flatMap executed?

I understand that usdQuote and chfQuote are executed by "some thread" at "some time" and their registered callback functions called, questions are:

a) Are usdQuote and chfQuote executed concurrently? (I'm pretty sure they are).

b) How does flatMap assign the value of the Future useQuote to usd? As in, does it get called when the operation usdQuote completes?

c) What thread is executing the 'flatMap' and 'map' operation (probably more of a follow-on from the last question).

Cheers.

回答1:

• a) When you created them you've already started them executing against the implicit ExecutionContext in scope, so they're potentially running concurrently as it depends on how that is executing them.

• b) It doesn't really assign the value as such, but the implementation uses the onComplete method to cause the function you've passed to be triggered once a result has been reached. At the current time this should link to that flatMap method I'm referring to: https://github.com/scala/scala/blob/v2.11.2/src/library/scala/concurrent/Future.scala#L246

• c) Those are running via the ExecutionContext previously mentioned, consider also that if those Future instances can be running on different ExecutionContexts, so parts of the for-comprehension can be running on different thread pools.

回答2:

I'm face the same question... And i found useful this general explanation about for-comprehesion. May be this helps:

For-Comprehensions

A for-comprehension is syntactic sugar for map, flatMap and filter operations on collections.

The general form is for (s) yield e

• s is a sequence of generators and filters
• p <- e is a generator
• if f is a filter
• If there are several generators (equivalent of a nested loop), the last generator varies faster than the first
• You can use { s } instead of ( s ) if you want to use multiple lines without requiring semicolons
• e is an element of the resulting collection

Example 1:

  // list all combinations of numbers x and y where x is drawn from
  // 1 to M and y is drawn from 1 to N
  for (x <- 1 to M; y <- 1 to N)
    yield (x,y)

is equivalent to

(1 to M) flatMap (x => (1 to N) map (y => (x, y)))

Translation Rules

A for-expression looks like a traditional for loop but works differently internally

• for (x <- e1) yield e2 is translated to e1.map(x => e2)
• for (x <- e1 if f) yield e2 is translated to for (x <- e1.filter(x => f)) yield e2
• for (x <- e1; y <- e2) yield e3 is translated to e1.flatMap(x => for (y <- e2) yield e3)

This means you can use a for-comprehension for your own type, as long as you define map, flatMap and filter

Example 2:

for {  
  i <- 1 until n  
  j <- 1 until i  
  if isPrime(i + j)  
} yield (i, j)

is equivalent to

for (i <- 1 until n; j <- 1 until i if isPrime(i + j))
    yield (i, j)

is equivalent to

(1 until n).flatMap(i => (1 until i).filter(j => isPrime(i + j)).map(j => (i, j)))

回答3:

You also have a good example of concurrent Future execution in "Scala notes – Futures – 3 (Combinators and Async)" from Arun Manivannan.

Our Futures need to run in parallel.
In order to achieve this, all we need to do is to extract the Future block out and declare them separately.

Code:

val oneFuture: Future[Int] = Future {
  Thread.sleep(1000)
  1
}

val twoFuture: Future[Int] = Future {
  Thread.sleep(2000)
  2
}

val threeFuture: Future[Int] = Future {
  Thread.sleep(3000)
  3
}

for-comprehension:

def sumOfThreeNumbersParallelMapForComprehension(): Future[Int] = for {  
    oneValue <- oneFuture
    twoValue <- twoFuture
    threeValue <- threeFuture
} yield oneValue + twoValue + threeValue

flatmap:

def sumOfThreeNumbersParallelMap(): Future[Int] = oneFuture.flatMap { oneValue =>  
    twoFuture.flatMap { twoValue =>
      threeFuture.map { threeValue =>
        oneValue + twoValue + threeValue
      }
    }
}

Test:

describe("Futures that are executed in parallel") {
  it("could be composed using for comprehensions") {
    val futureCombinators = new FutureCombinators
    val result = timed(Await.result(futureCombinators.sumOfThreeNumbersParallel(), 4 seconds))
      result shouldBe 6
  }
}

It does illustrate that:

1. Future is a container of a value(s) of some type (i.e it accepts a type as an argument and it can’t exist without it).
   You can have a Future[Int] or Future[String] or Future[AwesomeClass] – you can’t just have a plain Future.
   A fancy term for this is type-constructor.
   To compare, a List is a type constructor (and a Monad as well).
   A List is a container of values that are of type Int, String or any of other types. A List/Future without a contained type does not exist.
2. Future has flatMap and unit functions (and consequentially a map function too).