akka actors默认邮箱介绍

简介: 1. UnboundedMailbox is the default unbounded MailboxType used by Akka Actors  ”无界邮箱“ 是akka actors默认使用的邮箱, UnboundedMailbox继承了MailboxType /** *...

 1. UnboundedMailbox is the default unbounded MailboxType used by Akka Actors

 ”无界邮箱“ 是akka actors默认使用的邮箱,

UnboundedMailbox继承了MailboxType

/**
 * MailboxType is a factory to create MessageQueues for an optionally
 * provided ActorContext.
 *
 * <b>Possibly Important Notice</b>
 *
 * When implementing a custom mailbox type, be aware that there is special
 * semantics attached to `system.actorOf()` in that sending to the returned
 * ActorRef may—for a short period of time—enqueue the messages first in a
 * dummy queue. Top-level actors are created in two steps, and only after the
 * guardian actor has performed that second step will all previously sent
 * messages be transferred from the dummy queue into the real mailbox.
 */
trait MailboxType {
  def create(owner: Option[ActorRef], system: Option[ActorSystem]): MessageQueue
}

 和ProducesMessageQueue

trait ProducesMessageQueue[T <: MessageQueue]

 MailboxType特质的create方法接受两个参数owner和system,owner表示“消息队列”的所有者,system表示 ???

UnboundedMailbox的定义如下:

/**
 * UnboundedMailbox is the default unbounded MailboxType used by Akka Actors.
 */
case class UnboundedMailbox() extends MailboxType with ProducesMessageQueue[UnboundedMailbox.MessageQueue] {

  // 使用this关键字来调用构造函数。
  // 它演示了如何从其他构造函数调用构造函数。必须确保this必须放在构造函数中的第一个语句,同时调用其他构造函数this(),否则编译器会抛出错误。
  // “=”等号右边调用的this()是指向的UnboundedMailBox()类
  def this(settings: ActorSystem.Settings, config: Config) = this()

  // 重写MailboxType的create()方法,并创建UnboundedMailbox.MessgaeQueue对象
  final override def create(owner: Option[ActorRef], system: Option[ActorSystem]): MessageQueue =
    new UnboundedMailbox.MessageQueue
}

 UnboundedMailbox中的create()方法中创建的UnboundedMailbox.MessgaeQueue对象是

object UnboundedMailbox {
  class MessageQueue extends ConcurrentLinkedQueue[Envelope] with UnboundedQueueBasedMessageQueue {
    final def queue: Queue[Envelope] = this
  }
}  

   MessageQueue继承了ConCurrentLinkedQueue,并且ConCurrentLinkedQueue是存放的Envelope

final case class Envelope private (val message: Any, val sender: ActorRef)

(信封)对象,信封对象中包含一个message消息和发送者sender的ActorRef。

此外,MessageQueue还with了一个UnboundedQueueBasedMessageQueue

/**
 * BoundedMessageQueueSemantics adds bounded semantics to a QueueBasedMessageQueue,
 * i.e. blocking enqueue with timeout.
 */
trait BoundedMessageQueueSemantics {
  def pushTimeOut: Duration
}

trait UnboundedQueueBasedMessageQueue extends QueueBasedMessageQueue with UnboundedMessageQueueSemantics {
  // 入队,这里是入队了一个信封,貌似receiver没有用到?不明白为什么要有这个receiver,可能是为了明显标记是发给那个Actor的吧
  def enqueue(receiver: ActorRef, handle: Envelope): Unit = queue add handle
  // 出队,出队就是从队列中取出一个元素(一封信)
  def dequeue(): Envelope = queue.poll()
}

  UnboundedQueueBasedMessageQueue中实现了从QueueBasedMessageQueue继承来的enqueue()和dequeue()方法。更确切的话是从MessageQueue继承来的,因为QueueBasedMessageQueue是继承了MessageQueue,代码如下:

/**
 * This is a marker trait for message queues which support multiple consumers,
 * as is required by the BalancingDispatcher.
 */
trait MultipleConsumerSemantics

/**
 * A QueueBasedMessageQueue is a MessageQueue backed by a java.util.Queue.
 */
trait QueueBasedMessageQueue extends MessageQueue with MultipleConsumerSemantics {
  def queue: Queue[Envelope]
  // 消息的数量 (队列的元素个数)
  def numberOfMessages = queue.size
  // 是否包含消息 (队列是否为空)
  def hasMessages = !queue.isEmpty
  // 清空消息 (清空队列)
  def cleanUp(owner: ActorRef, deadLetters: MessageQueue): Unit = {
    if (hasMessages) {
      // 获取第一个消息
      var envelope = dequeue
      while (envelope ne null) {
        // 如果消息不为空,则不断向deadLetters发送消息
        // deadLetters也是一个消息队列actor.dispatcher.mailboxes.deadLetterMailbox.messageQueue
        deadLetters.enqueue(owner, envelope)
        envelope = dequeue
      }
    }
  }
}

  从QueueBasedMessageQueue的定义来看,其中包含了几个重要的方法。还定义了一个queue,回顾一下UnboundedMailbox的定义

object UnboundedMailbox {
  class MessageQueue extends ConcurrentLinkedQueue[Envelope] with UnboundedQueueBasedMessageQueue {
    final def queue: Queue[Envelope] = this
  }
}  

   其中的queue是重写了QueueBasedMessageQueue中定义的queue,def queue: Queue[Envelope]。

 

3. 邮箱是akka里面是怎么使用的? 

如果我们是使用的ActorRef去创建一个actor,例如:

    val testActor: ActorRef = context.actorOf(
      TestActor.props().withDispatcher("test-actor-dispatcher"),
      TestActor.Name)

   

目录
相关文章
|
Java Spring
spring data elasticsearch: 设置保活策略|长时间不连接es,报错超时连接
java client长时间没有连接es后,再次调用访问接口,报错连接超时
1881 0
|
3月前
|
消息中间件 Java Kafka
【Azure 事件中心】开启 Apache Flink 制造者 Producer 示例代码中的日志输出 (连接 Azure Event Hub Kafka 终结点)
【Azure 事件中心】开启 Apache Flink 制造者 Producer 示例代码中的日志输出 (连接 Azure Event Hub Kafka 终结点)
|
5月前
|
消息中间件 设计模式 网络安全
消息队列 MQ操作报错合集之broker启用controller配置时,遇到报错,是什么导致的
在使用消息队列MQ时,可能会遇到各种报错情况。以下是一些常见的错误场景、可能的原因以及解决建议的汇总:1.连接错误、2.消息发送失败、3.消息消费报错、4.消息重试与死信处理、5.资源与权限问题、6.配置错误、7.系统资源限制、8.版本兼容性问题。
103 1
|
消息中间件 缓存 Kafka
彻底搞懂 Kafka 消息大小相关参数设置的规则
Kafka 消息大小的设置还是挺复杂的一件事,而且还分版本,需要注意的参数巨多,而且每个都长得差不多,不但分版本,还需要注意生产端、broker、消费端的设置,而且还要区分 broker 级别还是 topic 级别的设置,而且还需要清楚知道每个配置的含义。 本文通过相关参数的解析说明,再结合实战测试,帮助你快速搞明白这些参数的含义以及规则。
1666 0
彻底搞懂 Kafka 消息大小相关参数设置的规则
|
消息中间件 网络安全 RocketMQ
这个错误可能是由于RocketMQ Proxy的配置或者运行状态出现了问题
这个错误可能是由于RocketMQ Proxy的配置或者运行状态出现了问题
207 1
|
消息中间件 监控
十五、.net core(.NET 6)搭建RabbitMQ消息队列生产者和消费者的简单方法
搭建RabbitMQ简单通用的直连方法 如果还没有MQ环境,可以参考上一篇的博客: https://www.cnblogs.com/weskynet/p/14877932.html
720 0
十五、.net core(.NET 6)搭建RabbitMQ消息队列生产者和消费者的简单方法
|
消息中间件
8、RabbitMQ教程-Direct Exchange类型的基本使用demo
8、RabbitMQ教程-Direct Exchange类型的基本使用demo
135 0
8、RabbitMQ教程-Direct Exchange类型的基本使用demo
|
消息中间件 网络架构
9、RabbitMQ教程-Topic Exchange类型的基本使用demo
9、RabbitMQ教程-Topic Exchange类型的基本使用demo
137 0
9、RabbitMQ教程-Topic Exchange类型的基本使用demo
|
分布式计算 网络协议 Scala
Spark worker 注册功能完成 | 学习笔记
快速学习 Spark worker 注册功能完成
Spark worker 注册功能完成 | 学习笔记
|
消息中间件 监控
轻松搞定RabbitMQ5:主题机制与RPC调用
轻松搞定RabbitMQ(六)——主题 翻译地址:http://www.rabbitmq.com/tutorials/tutorial-five-java.html 在上一篇博文中,我们进一步改良了日志系统。