使用 onComplete 时限制 Scala Future 块

Question

我正在尝试使用 Scala Futures 生成许多 CPU 密集型作业。因为有这么多，我需要限制这些作业（线程）的创建。为此，我使用：

import java.util.concurrent.ArrayBlockingQueue
import scala.concurrent._

val numThread = sys.runtime.availableProcessors

import java.util.concurrent.ExecutorService
import java.util.concurrent.ArrayBlockingQueue

implicit val context = ExecutionContext.fromExecutorService(
    new ThreadPoolExecutor(
      numThread, numThread,
      0L, TimeUnit.SECONDS,
      new ArrayBlockingQueue[ Runnable ]( numThread ) {
        override def offer( e: Runnable ) = {
          put( e ); // Waiting for empty room
          true
        }
      })
     )

为了测试这一点，我创建了 2 个非常简单的函数：

import scala.util.{ Try, Success, Failure }
import scala.util.Random

def longComputation() = {
  val id = Thread.currentThread().getId
  //blocking {
    println( s"Started thread: $id" )
    Thread.sleep( 500 )
    println( s"Finished thread: $id" )
  //}
  id
}

def processResult[T](r : Try[T]) = {
  blocking {
      r match {
        case Success( id ) => println( s"Thread result: $id" )
        case Failure( t )  => println( "An error has occured: " + t.getMessage )
       }
  }

}

然后我执行测试以通过多线程执行任务：

def main( args: Array[ String ] ) {


   val s = Stream.from( 0 )
   //s.foreach { x => println(x) ;  val f = Future( longComputation ) ; f.onComplete{ processResult } }

   s.foreach { x => 
     println(x) 
     val f = Future( longComputation )  
     val p = Promise[Long]()
     p completeWith f
     p.future.onComplete{ processResult } 
   }

   println("Finished")
   context.shutdown
 } 

当我执行此操作时，启动了 16 个线程（CPU 计数为 8）。该程序打印了“完成”消息。然后系统锁定，不执行任何其他操作。但是，如果我删除回调，则线程将按预期执行 ad infinitum。

上面我已经尝试了blocking，也使用了Promise。行为没有变化。所以我的问题是：如何在不阻塞回调的情况下限制任务执行？如果这不可能，在线程中进行 I/O 是否可行（未来）？

感谢任何指针。

Answer 1

程序在死锁中运行。提供的threadPool 具有固定大小，因此会发生以下情况： Future(longComputation) 从线程池中分配一个线程并开始工作。完成后，onComplete 从池中分配一个Thread 来执行提供的函数。

鉴于完成工作比完成工作需要更长的时间，在某些时候，所有线程都在忙于工作。它们中的任何一个都完成了，onComplete 也需要一个线程，因此它向执行程序请求一个线程。工作无法完成，因为所有线程都忙，机器陷入死锁。

我们可以通过给消费者预留资源来解决这个生产者-消费者的死锁。这样，固定大小的线程池会限制工作，但我们确保可以进一步处理任何已完成的工作。

我将context 重命名为fixedContext 的这个sn-p 显示了使用单独的上下文来处理结果，从而解决了死锁。我也去掉了Promise，除了代理未来之外，它并没有起到真正的作用。

val fixedContext = // same as in question
val singleThreadContext = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(1))
...
...
def main( args: Array[ String ] ) {

   val s = Stream.from( 0 )

   s.foreach { x => 
     println(x)
     val f = Future( longComputation )(fixedContext)  
     f.onComplete{ processResult }(singleThreadContext)
   }

   println("Finished")
   fixedContext.shutdown
 } 
}

Answer 2

当一个线程完成longComputation 时，它会尝试将作业放入队列以执行回调，并被阻塞，因为队列已满。因此，最终，第一批“作业”完成了，但所有线程仍然很忙，在队列中等待调度回调，并且没有任何东西可以退出队列。

解决方案？从队列中删除限制。这样，尝试提交回调的线程不会被阻塞，并且可以用于接收下一个任务。

您可能希望在您的生产者循环中插入一些东西以减慢它的速度，这样您的无限队列就不会耗尽所有内存。可能是Semaphore？

val sem = new Semaphore(numThread*2)
def processResult[T](r : Try[T]) = blocking {
  r match {
    case Success( id ) => println( s"Thread result: $id" )
    case Failure( t )  => println( "An error has occured: " + t.getMessage )
  }
  sem.release
}

Stream.from(0).foreach { _ => 
  sem.acquire
  new Future(longComputation).onComplete(processResult)
}

您不需要自定义执行上下文 - scala 的默认设置实际上更适合您想要做的事情