Scala Actors：如果队列太长，是否会丢弃消息？

Question

我希望能够在演员的邮箱中删除消息，如果它变得太满。例如，如果队列大小达到1000条消息，则应删除最旧的消息。

- Kim Stebel

Answer 1

您不能直接使用邮箱，但可以在现有库的基础上实现消息过期模式。

发送每个消息时都附带一个创建日期：

case class ExpirableMessage(msg: String, createdAt: Long) 

使用reactWithin(0)扫描邮箱，并过滤掉过期的邮件：

react{ 
  case msg: ExpirableMessage => 
    // handle the message
    // clean the mailbox with nested react
    reactWithin(0){
        case ExpirableMessage(_, createdAt) if(currentTimeMillis - createdAt > INTERVAL) =>
        case TIMEOUT =>
    }
}

- Vasil Remeniuk

Answer 2

您可以通过使用代理actor将一个actor的队列实例化到堆上并限制其利用率。然后您可以编写类似以下内容的代码：

// adder actor with a bounded queue size of 4
val adder = boundActor(4) {
  loop {
    react {
      case x: Int => reply(x*2)
    }
  }
}

// test the adder
actor {
  for (i <- 1 to 10) {
    adder !! (i, { case answer: Int => println("Computed " + i + " -> " + answer) })
  }
}

这是boundedActor的实现。请注意，一个boundedActor必须始终回复其发送者，否则就无法跟踪其队列大小，而且boundedActor会冻结并拒绝接受任何进一步的消息。

object ActorProxy extends scala.App {

  import scala.actors._
  import scala.actors.Actor._
  import scala.collection.mutable._

  /**
   * Accepts an actor and a message queue size, and 
   * returns a proxy that drops messages if the queue
   * size of the target actor exceeds the given queue size.
   */
  def boundActorQueue(target: Actor, maxQueueLength: Int) = actor {
    val queue = new Queue[Tuple2[Any, OutputChannel[Any]]]
    var lastMessageSender: Option[OutputChannel[Any]] = None

    def replyHandler(response: Any) {
      if (lastMessageSender.get != null) lastMessageSender.get ! response
      if (queue.isEmpty) {
        lastMessageSender = None
      } else {
        val (message, messageSender) = queue.dequeue
        forwardMessage(message, messageSender)
      }
    }

    def forwardMessage(message: Any, messageSender: OutputChannel[Any]) = {
      lastMessageSender = Some(messageSender)
      target !! (message, { case response => replyHandler(response) })
    }

    loop {
      react {
        case message =>
          if (lastMessageSender == None) {
            forwardMessage(message, sender)
          } else {
            queue.enqueue((message, sender))
            // Restrict the queue size
            if (queue.length > maxQueueLength) {
                val dropped = queue.dequeue
                println("!!!!!!!! Dropped message " + dropped._1)
            }
          }
      }
    }
  }

  // Helper method
  def boundActor(maxQueueLength: Int)(body: => Unit): Actor = boundActorQueue(actor(body), maxQueueLength)

}

- Nikolay Botev