java - Akka 是否有一个 ExecutorCompletionService 等价物，其中 Futures 按完成时间排队？

Question

使用 java，我可以创建一个 ExecutorCompletionService，其中包含一个执行程序和一系列任务。此类安排提交的任务在完成后放置在可使用 take 访问的队列中。 https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorCompletionService.htmlAkka 是否有类似的东西来管理 Actor 返回的 future ？

Answer 1

此答案仅适用于 Scala。在 scala 中有 sequence/firstCompletedOf 到 compose futures ，它会在 all/one 基础 future 完成后返回您新的 future 完成(相当于 CompletionService 的 api docs 中的示例) ).这种解决方案比 ecs.take().get() 更安全，因为如果您使用 onComplete 监听器，则不会阻塞；但是，如果您仍然需要一些阻塞式服务员 - 使用 Await.result。因此，不需要 CompletionService，因为 future 列表足够灵活并且更安全。第一个例子的等价物:

  val solvers: List[() => Int] = ...
  val futures = solvers.map(s => Future {s()}) //run execution
  (Future sequence futures) onComplete { results: Seq[Int] =>
      results.map(use)
  }

另一个例子是cancelling the task :

  val solvers: List[Future => Int] = ... //some list of functions(tasks), Future is used to check if task was interrupted
  val (futures, cancels): solvers.map(cancellableFuture) //see https://stackoverflow.com/questions/16020964/cancellation-with-future-and-promise-in-scala

  (Future firstCompletedOf futures) onComplete { result: Int =>
        cancels.foreach(_())
        use(result)
  }

谈到 Java，Akka 对 scala 的 future 进行了改编:http://doc.akka.io/docs/akka/snapshot/java/futures.html

如果您只想在结果完成时按顺序处理结果，您可以为此使用 actor:

  val futures: List[Future]
  futures.map(_ pipeTo actor) //actor's mailbox is used as queue

模拟完成队列的行为(不推荐):

  import scala.concurrent._
  import duration._
  import scala.concurrent.ExecutionContext.Implicits.global //some execution context

  class Queue[T](solvers: Seq[() => T]) extends Iterator[T]{
     case class Result(f: Future[Result], r: T)
     var futures: Set[Future[Result]] = solvers map {s => 
        lazy val f: Future[Result]  = Future{Result(f, s())}
        f
     } toSet

     def hasNext() = futures.nonEmpty        

     def next() = {         
        val result = Await.result((Future firstCompletedOf futures.toSeq), Duration.Inf)
        futures -= result.f
        result.r
     }
  }

scala> val q = new Queue(List(() => 1, () => 2, () => 3, () => 4))
q: Queue[Int] = non-empty iterator

scala> q.next
res14: Int = 2

scala> q.next
res15: Int = 1

scala> q.foreach(println)
4
3