内存 – Actors邮箱溢出.斯卡拉

从这里可以看出：scala.actors.MQueue,一个actor的邮箱被实现为链表,因此只受堆大小的限制.

尽管如此,如果您担心生产者非常快且消费者非常慢,我建议您探索一种限制机制.但我不建议从问题scala mailbox size limit的接受答案中采用这种方法.

通常,在系统受到严重压力时尝试发送过载消息似乎不是一个好主意.如果您的系统太忙而无法检查过载怎么办？如果过载消息的接收者太忙而无法对其进行操作怎么办？另外,删除消息对我来说听起来不是一个好主意.我认为您希望所有工作项目都可靠地处理.

另外,我不会依赖mailboxSize来确定负载.您无法区分不同的消息类型,您只能从消费者本身而不是生产者那里进行检查.

我建议使用消费者要求更多工作的方法,当他知道他可以处理它时.

下面是一个如何实现它的简单示例.

import scala.actors._
import Actor._

object ConsumerProducer {
  def main(args: Array[String]) {
    val producer = new Producer(Iterator.range(0,10000))
    val consumer = new Consumer(producer)
  }
}

case class Produce(count: Int)
case object Finished

class Producer[T](source: Iterator[T]) extends Actor {

  start

  def act() {
    loopWhile(source.hasNext) {
      react {
        case Produce(n: Int) => produce(n)
      } 
    }
  }

  def produce(n: Int) {
    println("producing " + n)
    var remaining = n
    source takeWhile(_ => remaining > 0) foreach { x => sender ! x; remaining -= 1 }
    if(!source.hasNext) sender ! Finished
  }
}

class Consumer(producer: Actor) extends Actor {

  start

  private var remaining = 0

  def act() {
    requestWork()
    consume()
  }

  def consume(): Nothing = react {
    case Finished => println("Finished")
    case n: Int => work(n); requestWork(); consume()
  }

  def requestWork() = if(remaining < 5) { remaining += 10; producer ! Produce(10) }

  def work(n: Int) = {
    println(n + ": " + (0 until 10000).foldLeft(0) { (acc,x) => acc + x * n })
    remaining -= 1
  }
}