Twitter的Future.collect无法同时工作(Scala)

发布时间：2020-12-16 09:51:38 所属栏目：安全来源：网络整理

导读：来自node.js背景,我是 Scala的新手,我尝试使用Twitter的Future.collect来执行一些简单的并发操作.但我的代码显示顺序行为而不是并发行为.我究竟做错了什么？这是我的代码, import com.twitter.util.Futuredef waitForSeconds(seconds: Int,container:String

来自node.js背景,我是 Scala的新手,我尝试使用Twitter的Future.collect来执行一些简单的并发操作.但我的代码显示顺序行为而不是并发行为.我究竟做错了什么？

这是我的代码,

import com.twitter.util.Future

def waitForSeconds(seconds: Int,container:String): Future[String]  = Future[String] {
  Thread.sleep(seconds*1000)
  println(container + ": done waiting for " + seconds + " seconds")
  container + " :done waiting for " + seconds + " seconds"
}

def mainFunction:String = {
  val allTasks = Future.collect(Seq(waitForSeconds(1,"All"),waitForSeconds(3,waitForSeconds(2,"All")))
  val singleTask = waitForSeconds(1,"Single")

  allTasks onSuccess  { res =>
    println("All tasks succeeded with result " + res)
  }

  singleTask onSuccess { res =>
    println("Single task succeeded with result " + res)
  }

  "Function Complete"
}

println(mainFunction)

这是我得到的输出,

All: done waiting for 1 seconds
All: done waiting for 3 seconds
All: done waiting for 2 seconds
Single: done waiting for 1 seconds
All tasks succeeded with result ArraySeq(All :done waiting for 1 seconds,All :done waiting for 3 seconds,All :done waiting for 2 seconds)
Single task succeeded with result Single :done waiting for 1 seconds
Function Complete

我期望的输出是,

All: done waiting for 1 seconds
Single: done waiting for 1 seconds
All: done waiting for 2 seconds
All: done waiting for 3 seconds
All tasks succeeded with result ArraySeq(All :done waiting for 1 seconds,All :done waiting for 2 seconds)
Single task succeeded with result Single :done waiting for 1 seconds
Function Complete

解决方法

Twitter的期货比Scala标准库期货更明确地执行计算的位置.特别是,Future.apply将安全地捕获异常(如scFuture),但它没有说明计算将运行在哪个线程.在你的情况下,计算在主线程中运行,这就是为什么你是看到你看到的结果.

与标准库的未来API相比,这种方法有几个优点.首先,它使方法签名更简单,因为没有必须在任何地方传递的隐式ExecutionContext.更重要的是,它可以更容易避免上下文切换(由Brian Degenhardt编写的here’s a classic explanation).在这方面,Twitter的未来更像是Scalaz的任务,并且具有基本相同的性能优势(例如在this blog post中描述).

更明确地计算运行位置的缺点是你必须更明确地计算运行的位置.在你的情况下你可以写这样的东西：

import com.twitter.util.{ Future,FuturePool }

val pool = FuturePool.unboundedPool

def waitForSeconds(seconds: Int,container:String): Future[String] = pool {
  Thread.sleep(seconds*1000)
  println(container + ": done waiting for " + seconds + " seconds")
  container + " :done waiting for " + seconds + " seconds"
}

这不会产生您要求的输出(“函数完成”将首先打印,并且所有Task和singleTask不会相互排序),但它将在不同的线程上并行运行任务.

(作为脚注：上面示例中的FuturePool.unboundedPool是一种为演示创建未来池的简单方法,并且通常很好,但它不适合CPU密集型计算 – 请参阅the FuturePool API docs了解其他方法创建一个将使用您提供的ExecutorService并可以自行管理的未来池.)

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!