Актеры Scala вместо Java Futures

После беглого взгляда я бы предложил следующее обновление:

val resultsChannel = new Channel[List[B]] // used instead of countdown latch to get the results

val dispatcher = new Actor {

  def act = loop(Nil, (0 to actorCount).map(id =>
      (id, new Worker(id).start.asInstanceOf[Worker])).toMap,
    Nil)

  @tailrec
  def loop(jobQueue: List[A], // queue, workers and results are immutable lists, passed recursively through the loop
           workers: Map[Int, Worker],
           res: List[B]):Unit = react {
    case ReportResult(id, result) =>
      val results = result :: res
      if (results.size == jobs.size) { // when the processing is finished, sends results to the output channel        
        resultsChannel ! results
      }
      loop(jobQueue, workers, results)

    case SendJob(id) =>
      if (!jobQueue.isEmpty) {
        workers(id) ! Process(jobQueue.head)
        loop(jobQueue.tail, workers, res)
      }

    case Stopped(id) =>
      loop(jobQueue, workers - id, res)
  }

}
dispatcher.start()

def results: List[B] = {
  resultsChannel.receive {
    case results => results // synchronously wait for the data in the channel
  }
}

Вот окончательный вариант, который я придумал (спасибо Василю Ременюку). Операторы println, помеченные комментарием // DEBUG, показывают прогресс, а метод main является модульным тестом:

import scala.actors.Actor
import scala.actors.Channel
import scala.actors.Scheduler
import scala.annotation.tailrec

object MultiWorker {
  private val megabyte = 1024 * 1024
  private val runtime = Runtime.getRuntime

  def main(args: Array[String]) {
    val jobs = (0 until 5).map((value: Int) => value).toList
    val multiWorker = new MultiWorker[Int, Int](jobs, 2, { value =>
        Thread.sleep(100)
        println(value)
        value
      })
    println("multiWorker.results: " + multiWorker.results)
    Scheduler.shutdown
  }
}

class MultiWorker[A, B](jobs: List[A],
                        actorCount: Int,
                        process: (A) => B) {
  import MultiWorker._

  sealed abstract class Message

  // Dispatcher -> Worker: Run this job and report results
  case class Process(job: A) extends Message

  // Worker -> Dispatcher: Result of processing
  case class ReportResult(id: Int, result: B) extends Message

  // Worker -> Dispatcher: I need work -- send me a job
  case class SendJob(id: Int) extends Message

  // Worker -> Dispatcher: I have stopped as requested
  case class Stopped(id: Int) extends Message

  // Dispatcher -> Worker: Stop working -- all jobs done
  case class StopWorking() extends Message

  /**
   * A simple logger that can be sent text messages that will be written to the
   * console. Used so that messages from the actors do not step on each other.
   */
  object Logger
  extends Actor {
    def act() {
      loop {
        react {
          case text: String => println(text)
          case StopWorking => exit()
        }
      }
    }
  }
  Logger.start()

  /**
   * A worker actor that will process jobs and return results to the
   * dispatcher.
   */
  case class Worker(id: Int)
  extends Actor{
    def act() {
      // Ask the dispatcher for an initial job
      dispatcher ! SendJob(id)

      loop {
        react {
          case Process(job) =>
            println("Worker(" + id + "): " + Process(job)) // DEBUG
            val startTime = System.nanoTime
            dispatcher ! ReportResult(id, process(job))

            val endTime = System.nanoTime
            val totalMemory = (runtime.totalMemory / megabyte)
            val usedMemory = totalMemory - (runtime.freeMemory / megabyte)
            val message = "Finished job " + job + " in " +
            ((endTime - startTime) / 1000000000.0) +
            " seconds using " + usedMemory +
            "MB out of total " + totalMemory + "MB"
            Logger ! message

            dispatcher ! SendJob(id)

          case StopWorking() =>
            println("Worker(" + id + "): " + StopWorking()) // DEBUG
            dispatcher ! Stopped(id)
            exit()
        }
      }
    }
  }

  val resultsChannel = new Channel[List[B]]
  /**
   * The job dispatcher that sends jobs to the worker until the job queue
   * (jobs: TraversableOnce[A]) is empty. It then tells the workers to
   * stop working and returns the List[B] results to the caller.
   */
  val dispatcher = new Actor {
    def act() {
      @tailrec
      def loop(jobs: List[A],
               workers: Map[Int, Worker],
               acc: List[B]) {
        println("dispatcher: loop: jobs: " + jobs + ", workers: " + workers + ", acc: " + acc) // DEBUG
        if (!workers.isEmpty) { // Stop recursion when there are no more workers
          react {
            case ReportResult(id, result) =>
              println("dispatcher: " + ReportResult(id, result)) // DEBUG
              loop(jobs, workers, result :: acc)

            case SendJob(id) =>
              println("dispatcher: " + SendJob(id)) // DEBUG
              if (!jobs.isEmpty) {
                println("dispatcher: " + "Sending: " + Process(jobs.head) + " to " + id) // DEBUG
                workers(id) ! Process(jobs.head)
                loop(jobs.tail, workers, acc)
              } else {
                println("dispatcher: " + "Sending: " + StopWorking() + " to " + id) // DEBUG
                workers(id) ! StopWorking()
                loop(Nil, workers, acc)
              }

            case Stopped(id) =>
              println("dispatcher: " + Stopped(id)) // DEBUG
              loop(jobs, workers - id, acc)
          }
        } else {
          println("dispatcher: " + "jobs: " + jobs + ", workers: " + workers + ", acc: " + acc) // DEBUG
          resultsChannel ! acc
        }
      }

      loop(jobs, (0 until actorCount).map(id => (id, new Worker(id).start.asInstanceOf[Worker])).toMap, Nil)
      exit()
    }
  }.start()

  /**
   * Get the results of the processing -- wait for the dispatcher to finish
   * before returning.
   */
  def results: List[B] = {
    resultsChannel.receive {
      case results => results
    }
  }
}