2

標準のscala.actorsパッケージを使用して、Scalaのディスパッチャー-ワーカーアクターパターンを設計しようとしています。

ディスパッチャはaから作業を受け取り、java.util.concurrent.LinkedBlockingQueueそれをワーカーアクターに送信して処理します。すべての作業が完了したら、ディスパッチャは各ワーカーに終了するように指示し、次に終了する必要があります。これが私が思いついたコードですが、すべての作業が完了するとハングします（'GiveMeWorkディスパッチャーのキューに保留中のメッセージがあると思います）：

import java.util.concurrent.LinkedBlockingQueue
import scala.actors.Actor

object Dispatcher
extends Actor {
  println("Dispatcher created")

  def act() {
    val workers = (1 to 4).map(id => (new Worker(id)).start())

    loop {
      react {
        case 'GiveMeWork =>
          // println("Worker asked for work")
          val (time, i) = workQueue.take()
          if (time == 0) {
            println("Quitting time")
            workers.foreach(_ !? 0L)
          } else {
            println("Arrival at dispatcher: i: " + i + " dispatch time: " +
                    time + ", elapsed: " + (System.nanoTime() - time))
            sender ! time
          }
        case 'Quit =>
          println("Told to quit")
          sender ! 'OffDuty
          exit()
      }
    }
  }
}

class Worker(id: Int)
extends Actor {
  println("Worker(" + id + ") created")
  var jobs = 0

  def act() {
    Dispatcher ! 'GiveMeWork

    loop {
      react {
        case time: Long =>
          if (time == 0) {
            println("Worker(" + id + ") completed " + jobs + " jobs")
            sender ! 'OffDuty
            exit()
          } else {
            println("Arrival at worker(" + id + "): dispatch time: " +
                    time + ", elapsed: " + (System.nanoTime() - time))
            Thread.sleep(id)
            jobs += 1
            Dispatcher ! 'GiveMeWork
          }
      }
    }
  }
}

val workQueue = new LinkedBlockingQueue[(Long, Int)](1000)

Dispatcher.start()

for (i <- 0 until 5000) {
  Thread.sleep(1)
  workQueue.put((System.nanoTime(), i))
}

workQueue.put((0L, 0))

println("Telling Dispatcher to quit")
Dispatcher !? 'Quit
4

1 に答える 1

3

レースがあります：

val (time, i) = workQueue.take()

を含むすべての作業が完了しているworkQueue.put((0L, 0))ため、永遠に待機します。

異なるタイプの並行性を同時に使用することはお勧めできません。

ディスパッチャは、タスクの制限についてタスクソースに通知できます。

import scala.actors.{Actor, OutputChannel}
import scala.collection.mutable.Queue

case class Task(time: Long, i: Int)
case object GiveMeWork
case object Quit
case object OffDuty

object Dispatcher extends Actor {
  println("Dispatcher created")

  def act() {
    val workers = (1 to 4).map(id => (new Worker(id)).start())
    val waitingWorkers = Queue[OutputChannel[Any]](workers: _*)
    val tasks = Queue[Task]()
    var workSender: Option[OutputChannel[Any]] = None

    loop {
      react {
        case GiveMeWork =>
          if (!tasks.isEmpty) sender ! tasks.dequeue()
          else waitingWorkers enqueue sender

          workSender map { _ ! GiveMeWork }
          workSender = None
        case t: Task =>
          if (!waitingWorkers.isEmpty) waitingWorkers.dequeue() ! t
          else tasks enqueue t

          if (tasks.length < 1000) sender ! GiveMeWork
          else workSender = Some(sender)
        case Quit =>
          println("Told to quit")
          workers.foreach{ _ ! Quit }
          sender ! OffDuty
          exit()
      }
    }
  }
}

class Worker(id: Int)
extends Actor {
  var jobs = 0

  def act() {
    loop {
      react {
        case t: Task =>
          Thread.sleep(id)
          jobs += 1
          Dispatcher ! GiveMeWork
        case Quit =>
          println("Worker(" + id + ") completed " + jobs + " jobs")
          sender ! OffDuty
          exit()
      }
    }
  }
}

Dispatcher.start()

for (i <- 0 until 5000) {
  Thread.sleep(1)
  Dispatcher !? Task(System.nanoTime(), i)
}

println("Telling Dispatcher to quit")
Dispatcher !? Quit
于 2012-12-06T14:12:01.797 に答える