Scala 异步/等待和并行化

Scala async/await and parallelization

提问by Sanete

I'm learning about the uses of async/await in Scala. I have read this in https://github.com/scala/async

我正在学习 Scala 中 async/await 的用法。我已经在https://github.com/scala/async 中阅读了这个

Theoretically this code is asynchronous (non-blocking), but it's not parallelized:

理论上，这段代码是异步的（非阻塞），但它不是并行化的：

def slowCalcFuture: Future[Int] = ...             
def combined: Future[Int] = async {               
   await(slowCalcFuture) + await(slowCalcFuture)
}
val x: Int = Await.result(combined, 10.seconds)    

whereas this other one is parallelized:

而另一个是并行化的：

def combined: Future[Int] = async {
  val future1 = slowCalcFuture
  val future2 = slowCalcFuture
  await(future1) + await(future2)
}

The only difference between them is the use of intermediate variables. How can this affect the parallelization?

它们之间的唯一区别是中间变量的使用。这如何影响并行化？

回答by Patryk ?wiek

Since it's similar to async & awaitin C#, maybe I can provide some insight. In C#, it's a general rule that Taskthat can be awaited should be returned 'hot', i.e. already running. I assume it's the same in Scala, where the Futurereturned from the function does not have to be explicitly started, but is just 'running' after being called. If it's notthe case, then the following is pure (and probably not true) speculation.

由于它类似于async & await在 C# 中，也许我可以提供一些见解。在 C# 中Task，可以等待的一般规则应该返回“热”，即已经在运行。我假设它在 Scala 中是一样的，Future从函数返回的函数不必显式启动，而只是在被调用后“运行”。如果情况并非如此，那么以下纯属（可能不是真的）推测。

Let's analyze the first case:

先分析第一种情况：

async {
    await(slowCalcFuture) + await(slowCalcFuture)
}

We get to that block and hit the first await:

我们到达那个块并点击第一个等待：

async {
    await(slowCalcFuture) + await(slowCalcFuture)
    ^^^^^
}

Ok, so we're asynchronously waiting for that calculation to finish. When it's finished, we 'move on' with analyzing the block:

好的，所以我们异步等待计算完成。完成后，我们“继续”分析块：

async {
    await(slowCalcFuture) + await(slowCalcFuture)
                            ^^^^^
}

Second await, so we're asynchronously waiting for second calculation to finish. After that's done, we can calculate the final result by adding two integers.

第二次等待，所以我们异步等待第二次计算完成。完成后，我们可以通过将两个整数相加来计算最终结果。

As you can see, we're moving step-by-step through awaits, awaiting Futures as they come one by one.

正如您所看到的，我们正在逐步完成等待，等待Futures 一个一个出现。

Let's take a look at the second example:

我们来看第二个例子：

async {
  val future1 = slowCalcFuture
  val future2 = slowCalcFuture
  await(future1) + await(future2)
}

OK, so here's what (probably) happens:

好的，这就是（可能）发生的事情：

async {
  val future1 = slowCalcFuture // >> first future is started, but not awaited
  val future2 = slowCalcFuture // >> second future is started, but not awaited
  await(future1) + await(future2)
  ^^^^^
}

Then we're awaiting the first Future, but both of the futuresare currently running. When the first one returns, the second might have already completed (so we will have the result available at once) or we might have to wait for a little bit longer.

然后我们正在等待第一个Future，但两个期货目前都在运行。当第一个返回时，第二个可能已经完成（因此我们将立即获得结果）或者我们可能需要等待更长的时间。

Now it's clear that second example runs two calculations in parallel, then waits for both of them to finish. When both are ready, it returns. First example runs the calculations in a non-blocking way, but sequentially.

现在很明显，第二个示例并行运行两个计算，然后等待它们都完成。当两者都准备好时，它返回。第一个示例以非阻塞方式运行计算，但按顺序运行。

回答by Mikha

the answer by Patryk is correct if a little difficult to follow. the main thing to understand about async/await is that it's just another way of doing Future's flatMap. there's no concurrency magic behind the scenes. all the calls inside an async block are sequential, including await which doesn't actually block the executing thread but rather wraps the rest of the async blockin a closure and passes it as a callback on completionof the Futurewe're waiting on. so in the first piece of code the second calculation doesn't start until the first await has completed since no one started it prior to that.

如果有点难以理解，Patryk 的答案是正确的。关于 async/await 的主要理解是它只是另一种做Future's 的方式flatMap。幕后没有并发魔法。异步块内的所有调用都是连续的，包括 await，它实际上并不阻塞正在执行的线程，而是将异步块的其余部分包装在一个闭包中，并在我们正在等待的完成时将其作为回调传递Future。所以在第一段代码中，第二次计算直到第一个等待完成后才开始，因为在此之前没有人启动它。

回答by Vadym Chekan

In first case you create a new thread to execute a slow future and wait for it in a single call. So invocation of the second slow future is performed after the first one is complete.

在第一种情况下，您创建一个新线程来执行一个慢未来并在一次调用中等待它。因此，在第一个完成后执行第二个慢未来的调用。

In the second case when val future1 = slowCalcFutureis called, it effectively create a new thread, pass pointer to "slowCalcFuture" function to the thread and says "execute it please". It takes as much time as it is necessary to get a thread instance from thread pool, and pass a pointer to a function to the thread instance. Which can be considered instant. So, because val future1 = slowCalcFutureis translated into "get thread and pass pointer" operations, it is complete in no time and the next line is executed without any delay val future2 = slowCalcFuture. Feauture 2 is scheduled to execution without any delay too.

在第二种情况下，当val future1 = slowCalcFuture被调用时，它有效地创建了一个新线程，将指向“slowCalcFuture”函数的指针传递给线程并说“请执行它”。从线程池中获取线程实例并将指向函数的指针传递给线程实例所需的时间与需要的时间一样长。这可以被认为是即时的。所以，因为val future1 = slowCalcFuture被翻译成“获取线程和传递指针”操作，它很快就完成了，并且没有任何延迟地执行下一行val future2 = slowCalcFuture。特征 2 也计划立即执行。

Fundamental difference between val future1 = slowCalcFutureand await(slowCalcFuture)is the same as between asking somebody to make you coffee and waiting for your coffee to be ready. Asking takes 2 seconds: which is needed to say phrase: "could you make me coffee please?". But waiting for coffee to be ready will take 4 minutes.

val future1 = slowCalcFuture和之间的根本区别与await(slowCalcFuture)请某人为您煮咖啡和等待您的咖啡准备好是相同的。询问需要 2 秒钟：这需要说短语：“请给我煮咖啡吗？”。但是等待咖啡准备好需要 4 分钟。

Possible modification of this task could be waiting for 1st available answer. For example, you want to connect to any server in a cluster. You issue requests to connect to every server you know, and the first one which responds, will be your server. You could do this with: Future.firstCompletedOf(Array(slowCalcFuture, slowCalcFuture))

此任务的可能修改可能是等待第一个可用答案。例如，您想连接到集群中的任何服务器。您发出连接到您知道的每台服务器的请求，第一个响应的将是您的服务器。你可以这样做： Future.firstCompletedOf(Array(slowCalcFuture, slowCalcFuture))