如何使用中断器实现自定义演员邮箱？

Question

我对@9​​87654321@ 有一些经验，我真的很想使用disruptor 实现一个自定义actor 邮箱。

有什么指导方针吗？甚至可能吗？ Akka的actor邮箱有什么限制？

Answer 1

正如here 所说，您只需要实现一些方法 - 当然您应该使用指向环形缓冲区的指针直接写入/读取消息。您还应该记住：

• disruptor 通常会预先分配大量内存，因此每个actor 使用一个disruptor 是个坏主意，您可以结合BalancingPool 使用一个路由器actor（内部带有disruptor）。

1234563不同消息类型的不同起始值），但仍使用一个 Disruptor。所以不同的邮箱会指向一个破坏者。

• 您将失去对消息创建、提取等的低级控制，因此默认情况下不会进行批量分配。

• 您还可以使用 ring 中的历史记录来恢复失败 actor 的状态（在 preRestart 或主管中）。

LMAX 是怎么说的：

它的工作方式与更传统的方法不同，所以你 使用它与您可能习惯的略有不同。例如， 将模式应用于您的系统并不像替换所有 您的队列与魔术环缓冲区。我们有代码示例 引导您，越来越多的博客和文章提供概述 关于它是如何工作的，技术论文会像你一样详细介绍 期望，并且性能测试给出了如何使用的示例 破坏者 http://mechanitis.blogspot.com/2011/06/dissecting-disruptor-whats-so-special.html

而here 是一个简短的队列/干扰器/参与者比较

在pseudo-scala-code中它会是这样的：

object MyUnboundedMailbox {
  val buffer = new RingBuffer()

  class MyMessageQueue(val startPointer: Pointer, readerPointer: Pointer, writerPointer: Pointer) extends MessageQueue {

    // these should be implemented; queue used as example
    def enqueue(receiver: ActorRef, handle: Envelope): Unit = {
      writerPointer.allocate(() => handle) //allocate one element and set, if you want different message types - you should allocate big amount of data before and block when it ends (to not interfere with another messages), so it has to be bounded queue then  

    }
    def dequeue(): Envelope = readerPointer.poll()
    def numberOfMessages: Int = writerPointer - readerPointer //should be synchronized
    def hasMessages: Boolean = readerPointer == writerPointer //should be synchronized
    def cleanUp(owner: ActorRef, deadLetters: MessageQueue) { }
  }

  trait MyUnboundedMessageQueueSemantics 

}

class MyUnboundedMailbox(settings: ActorSystem.Settings, config: Config) extends MailboxType
  with ProducesMessageQueue[MyUnboundedMailbox.MyMessageQueue] {

  import MyUnboundedMailbox._
  final override def create(owner: Option[ActorRef],
                            system: Option[ActorSystem]): MessageQueue = {

    val pointer = ring.newPointer
    val read = pointer.copy
    val write = pointer.copy
    new MyMessageQueue(pointer, read, write) 
  }
    // you may use another strategy here based on owner (you can access name and path here), 
    // so for example may allocate same pointers for same prefixes in the name or path 
}

在故障恢复过程中可以使用不变的 MyMessageQueue.startPointer 来访问消息日志（你也可以看看 akka 的Event Sourcing 进行类比）。

在这里使用 UnboundedQueue 方法并不能保证消息传递，因为如果环“结束”，很旧的未传递消息可能会被新版本覆盖，因此您可能需要 BoundedQueue，例如 here。