akka 演员向邮箱的负责人发布消息答案

【问题标题】：akka actor post a message to head of the MailBoxakka 演员向邮箱的负责人发布消息
【发布时间】：2013-03-11 11:22:14
【问题描述】：

生产者演员可以将消息发布给另一个演员以立即处理吗？即向消费者邮箱的头部而不是消费者邮箱的尾部发布消息？

我知道 akka 提供了一种配置我自己定义的邮箱类型的方法，但是如何控制是否需要将某些类型的消息发布在邮箱的头部而不是尾部。例如TimerMessages。我想要一个时间窗口实现的精确计时器控制。消息必须仅保留 1000 毫秒（例如），如果消息处理消耗时间并且邮箱中有许多待处理的消息，我不希望将计时器消息附加到同一个队列。

我可以使用PriorityMailBox，但PriorityMailBox 的问题在于，即使它可以将更高优先级的消息（计时器消息）放在 MailBox 的头部，对于相同优先级的消息，消息在邮箱不保证与到货顺序一致。所以我也不能使用priorityMailBox。

有人可以告诉我如何实现这种行为吗？

【问题讨论】：

标签： scala akka actor

【解决方案1】：

您可以使用自己的PriorityMailBox，它可以处理消息的到达时间并将其用作附加优先级（对于具有相同“主要”优先级的消息）。

类似这样的东西（未测试）：

import akka.dispatch._
import com.typesafe.config.Config
import akka.actor.{ActorRef, PoisonPill, ActorSystem}
import java.util.Comparator
import java.util.concurrent.PriorityBlockingQueue

class MyTimedPriorityMailbox(settings: ActorSystem.Settings, config: Config)
  extends UnboundedTimedPriorityMailbox(
    TimedPriorityGenerator {
      case 'highpriority ⇒ 0

      case 'lowpriority  ⇒ 2

      case PoisonPill    ⇒ 3

      case otherwise     ⇒ 1
    })

case class TimedEnvelope(envelope: Envelope) {
  private val _timestamp = System.nanoTime()
  def timestamp = _timestamp
}

class UnboundedTimedPriorityMailbox( final val cmp: Comparator[TimedEnvelope], final val initialCapacity: Int) extends MailboxType {
  def this(cmp: Comparator[TimedEnvelope]) = this(cmp, 11)
  final override def create(owner: Option[ActorRef], system: Option[ActorSystem]): MessageQueue =
    new PriorityBlockingQueue[TimedEnvelope](initialCapacity, cmp) with TimedQueueBasedMessageQueue with TimedUnboundedMessageQueueSemantics {
      override def queue: java.util.Queue[TimedEnvelope] = this
    }
}

trait TimedQueueBasedMessageQueue extends MessageQueue {
  def queue: java.util.Queue[TimedEnvelope]
  def numberOfMessages = queue.size
  def hasMessages = !queue.isEmpty
  def cleanUp(owner: ActorRef, deadLetters: MessageQueue) {
    if (hasMessages) {
      var envelope = dequeue()
      while (envelope ne null) {
        deadLetters.enqueue(owner, envelope)
        envelope = dequeue()
      }
    }
  }
}

trait TimedUnboundedMessageQueueSemantics extends TimedQueueBasedMessageQueue {
  def enqueue(receiver: ActorRef, handle: Envelope) { queue add TimedEnvelope(handle) }
  def dequeue(): Envelope = Option(queue.poll()).map(_.envelope).getOrElse(null)
}


object TimedPriorityGenerator {
  def apply(priorityFunction: Any ⇒ Int): TimedPriorityGenerator = new TimedPriorityGenerator {
    def gen(message: Any): Int = priorityFunction(message)
  }
}


abstract class TimedPriorityGenerator extends java.util.Comparator[TimedEnvelope] {
  def gen(message: Any): Int

  final def compare(thisMessage: TimedEnvelope, thatMessage: TimedEnvelope): Int = {
    val result = gen(thisMessage.envelope.message) - gen(thatMessage.envelope.message)
    // Int.MaxValue / Int.MinValue check omitted
    if(result == 0) (thisMessage.timestamp - thatMessage.timestamp).toInt else result
  }

}

【讨论】：

谢谢，您在上面的示例中使用了哪个 akka 版本？我正在使用 akka 2.0.4，并且出现了一些编译错误。 UnboundedTimedPriorityMailbox.create 的方法参数是Option[ActorContext] 和方法cleanup(ActorContext,MessageQueue) 也需要定义。我已经进行了更改以使其编译，但我最好使用与您相同的 akka 版本。
我用过 Akka 2.1.1 和 Scala 2.10.0

【解决方案2】：

上面的代码运行正常。

只有一个细节。避免使用 System.getTimeNano()。它在多核机器中存在问题，因为它是由 per-cpu 逻辑定义的

Here another post

然后，我们在消息顺序中有一个奇怪的行为取决于哪个 cpu enque it.

我用经典的 System.currentTimeMillis() 来改变它。它不太精确，但在我们的例子中，如果两条消息具有相同的优先级和相同的毫秒生成时间，则不必关心它们的处理顺序。

感谢代码！

【讨论】：