Scala的Future和ExecutionContext执行

Question

假设我有以下一组代码在未来做了一些事情:

1 to 10 foreach {
  case x => Future { x + x }
}

假设我给这段代码提供了默认的ExecutionContext,我知道后台会发生什么,但我想知道的是如何实现Future的处理呢？我的意思是应该有一些线程或一组线程可能等待Future完成？这些线程被阻止了吗？在他们真正等待未来完成的意义上被阻止了？

现在在以下场景中:

val x: Future[MyType] = finishInSomeFuture()

假设x有一个超时,我可以像这样调用:

Future {
  blocking {
    x.get(3, TimeOut.SECONDS)
  }
} 

我真的在阻止吗？是否有更好的异步超时方法？

编辑:以下Timeout比我上面定义的阻塞上下文有多么不同或有多好？

object TimeoutFuture {
  def apply[A](timeout: FiniteDuration)(block: => A): Future[A] = {

    val prom = promise[A]

    // timeout logic
    Akka.system.scheduler.scheduleOnce(timeout) {
      prom tryFailure new java.util.concurrent.TimeoutException
    }

    // business logic
    Future { 
      prom success block
    }

    prom.future
  } 
}

Answer 1

1. 假设我有以下一组代码在未来做了一些事情:

   1 to 10 foreach {
     case x => Future { x + x }
   }
   

   ...

您的一段代码创建了十个Futures,使用隐式ExecutionContext提供的线程立即设置执行.由于您没有存储对期货的引用,并且不等待它们的执行,因此您的主线程(在您foreach定义的位置)不会阻塞并立即继续执行.如果那段代码在main方法的最后,那么,取决于生成的守护程序线程程序是否可以ThreadFactory在不等待Futures完成的情况下退出.ExecutionContext

2. 现在在以下场景中:

   val x: Future[MyType] = finishInSomeFuture()
   

   假设x有一个超时,我可以像这样调用:

   Future {
     blocking {
       x.get(3, TimeOut.SECONDS)
     }
   } 
   

   我真的在阻止吗？是否有更好的异步超时方法？

你可能意味着Await.result代替x.get:

def inefficientTimeoutFuture[T](f:Future[T], x:Duration) = Future { Await.result(f, x) }

在这种情况下,将来f将在单独的线程中计算,而另外的线程将被阻塞等待计算f.

使用调度程序创建TimeoutFuture 更有效,因为调度程序通常共享固定数量的线程(通常是一个),而阻塞Await.result将总是需要额外的线程来阻止.

3. 我想知道如何在不阻塞的情况下超时？

使用调度程序创建TimeoutFuture允许您在不阻塞的情况下超时运行.你正在将你的Future包装在超时助手中,新的Future要么成功完成,要么因超时而失败(无论什么是第一次).新的Future具有相同的异步性,它取决于你如何使用它(注册onComplete回调或同步等待结果,阻塞主线程).

UPD我将尝试澄清有关多线程和阻塞的一些基本知识.

现在异步非阻塞方法是趋势,但您必须了解阻塞意味着什么以及为什么应该避免它.

Java中的每个线程都需要付出代价.首先,创建新线程(这就是存在线程池的原因)相对昂贵,其次,它会消耗内存.为什么不用CPU？因为您的CPU资源受到您拥有的核心数量的限制.无论您拥有多少活动线程,您的并行度级别始终都会受到核心数量的限制.如果线程处于非活动状态(被阻止),则它不会占用CPU.

在当代Java应用程序中,您可以创建相当多的线程(数千个).问题是在某些情况下你无法预测你需要多少线程.这就是异步方法发挥作用的时候.它说:而不是阻塞当前线程,而其他一些线程做他们的工作让我们在回调中包装我们的后续步骤并将当前线程返回到池,因此它可以做一些其他有用的工作.因此,几乎所有线程都在忙于实际工作,而不仅仅是等待和消耗内存.

现在以计时器为例.如果您使用基于netty的HashedWheelTimer,则可以使用单线程支持它并安排数千个事件.当您创建Future被阻止等待超时时,每个"计划"占用一个线程.因此,如果您计划了数千次超时,那么最终会有数千个被阻塞的线程(这会再次消耗内存,而不是cpu).

现在你的"主要"未来(你希望在超时中包装)也不必阻塞线程.例如,如果您将来执行同步http请求,您的线程将被阻止,但如果您使用基于netty AsyncHttpClient(例如),则可以使用不占用该线程的基于承诺的未来.在这种情况下,您可以使用少量固定数量的线程来处理任意数量的请求(数十万).

UPD2

但是应该有一些线程应该阻塞,即使在Timer的情况下,因为它必须等待Timeout millis.那么好处是什么以及在哪里？我仍然阻止,但可能是我在Timer情况下阻止或？

这仅适用于一种特定情况:当您拥有等待异步任务完成的主线程时.在这种情况下你是对的,没有阻塞主线程就无法在超时中包装操作.在这种情况下使用Timers没有任何意义.您只需要额外的线程来执行操作,而主线程等待结果或超时.

但通常Futures用于更复杂的场景,其中没有"主"线程.例如,假设异步webserver,请求进来,你创建Future来处理它并注册回调来回复.没有"主"线程等待任何事情.

或者另一个示例,您希望通过单个超时向外部服务发出1000个请求,然后将所有结果收集到一个位置.如果您有该服务的异步客户端,则创建1000个请求,将它们包装在异步超时中,然后组合成一个Future.您可以阻止主线程等待该未来完成或注册回调以打印结果,但您不必创建1000个线程只是为了等待每个单独的请求完成.

所以,重点是:如果你已经有了同步流,并且想要在超时中包含它的某些部分,那么你唯一能做的就是阻止当前线程,直到其他线程执行该工作.如果要避免阻塞,则需要从一开始就使用异步方法.