scala - 为什么 Scala 中的生产者-消费者解决方案不起作用？

Question

我目前正在通过《Scala 编程》第 2 版一书学习 Scala。我尝试使用演员来实现生产者-消费者问题，但我的代码只打印：

Producer: new value: 0
Consumer #2: read value: 0
Consumer #4: read value: 0
Consumer #1: read value: 0
Consumer #3: read value: 0
Consumer #5: read value: 0

然后它挂起。我不知道我做错了什么。非常感谢任何帮助！

生产者.scala：

package cz.zaoral.scala.actors.prodcons

import scala.actors.Actor
import scala.actors.Actor.{react, actor, self}

class Producer(
  val cycles: Int,
  val productionTime: Int,
  val consumers: List[Actor])
extends Actor {
  private var consumersConsumed = 0
  private var currentValue = 0

  case class ProduceNewValue(cycleNum: Int)
  case class NewValueProduced(value: Int)

  private val producer = actor {
    loop {
      self.react {
        case ProduceNewValue(x) =>
          Thread.sleep(productionTime)
          Producer.this ! NewValueProduced(x)
        case x => println("Unhandled message in producer: "+ x)
      }
    }
  }

  def act() {
    producer ! ProduceNewValue(currentValue)
    loop {
      if (consumersConsumed == consumers.length - 1) {
        consumersConsumed = 0
        if (currentValue < cycles) {
          currentValue += 1
          producer ! ProduceNewValue(currentValue)
        } else {
          sendToAllConsumers(EndOfInput())
          exit()
        }
      }
      receiveMessage()
    }
  }

  private def receiveMessage() {
    react {
      case ValueRead() => consumersConsumed += 1
      case NewValueProduced(x) =>
        println("Producer: new value: "+ x)
        sendToAllConsumers(ValueProduced(x, self))
      case msg => println("unhandled message in Producer: "+ msg)
    }
  }

  private def sendToAllConsumers(msg: Message) {
    for (consumer <- consumers)
      consumer ! msg
  }
}

消费者.scala

package cz.zaoral.scala.actors.prodcons

import scala.actors.Actor

class Consumer(number: Int) extends Actor {
  def act() {
    loop {
      react {
        case ValueProduced(x, producer) =>
          println(this + ": read value: "+ x)
          producer ! ValueRead()
        case EndOfInput => exit()
        case x => println("Unhandled message in Consumer: "+ x)
      }
    }
  }

  override def toString =
    "Consumer #"+ number

  start()
}

消息.scala：

package cz.zaoral.scala.actors.prodcons

import scala.actors.Actor

abstract class Message

case class ValueProduced(newValue: Int, producer: Actor) extends Message
case class ValueRead extends Message
case class EndOfInput extends Message

Demo.scala

package cz.zaoral.scala.actors.prodcons

import scala.actors.Actor

object Demo {
  def main(args: Array[String]) {
    val consumers = for {
      i <- 1 to 5
      consumer = new Consumer(i)
    } yield consumer

    val producer = new Producer(10, 1000, consumers.toList)
    producer.start()
  }
}

编辑：

我自己找到了解决方案 :) 我以错误的方式使用了 actor 方法。scala.actors.Actor.actor 方法是一个工厂方法，它创建并立即启动新的actor。

新的Producer.scala：

package cz.zaoral.scala.actors.prodcons

import scala.actors.Actor
import scala.actors.Actor.{react, actor, self}


class Producer(
  val cycles: Int,
  val productionTime: Int,
  val consumers: List[Actor])
extends Actor {
  private var remainingConsumers = consumers.toSet
  private var currentValue = 0

  case class NewValueProduced(value: Int)

  private def produce() = actor {
    Thread.sleep(productionTime)
    Producer.this ! NewValueProduced(currentValue + 1)
  }

  def act() {
    produce()
    loop {
      if (remainingConsumers.isEmpty) {
        if (currentValue < cycles) {
          remainingConsumers ++= consumers
          produce()
        } else {
          sendToAllConsumers(EndOfInput())
          exit()
        }
      }
      receiveMessage()
    }
  }

  private def receiveMessage() {
    react {
      case ValueRead(consumer) =>
        assert(remainingConsumers.contains(consumer))
        remainingConsumers -= consumer
      case NewValueProduced(x) =>
        currentValue = x
        println("Producer: new value: "+ currentValue)
        sendToAllConsumers(ValueProduced(currentValue, self))
    }
  }

  private def sendToAllConsumers(msg: Message) {
    for (consumer <- consumers)
      consumer ! msg
  }
}

请注意，如果有人想在示例的其余部分中适应这个新定义，则必须在其他文件中进行一些小的更改，但这应该没什么大不了的。

有美好的一天！