在深入探讨Akka框架中的Mailbox和Dispatcher之前,先来了解它们是什么以及它们之间的关系。
在Akka框架中,Mailbox是一个队列,用于存储发送给Actor的消息。通常情况下,每个Actor都有一个对应的Mailbox。虽然在某些情况下,如果涉及到路由,可能会有多个Actor共享一个Mailbox,但目前先假设Mailbox和Actor之间是一一对应的关系。
在Akka框架中,Dispatcher是Actor系统的心脏,负责分发消息。Akka允许配置特定的Dispatcher,并且Dispatcher可以进一步配置使用特定的Mailbox类型。
以下是一个配置Actor使用自定义Dispatcher的示例代码:
import akka.actor.Props
val myActor = context.actorOf(Props[MyActor].withDispatcher(
"my-dispatcher"
), "myactor1")
在系统的配置中,可能会有如下自定义Dispatcher的配置:
my-dispatcher {
# Type of mailbox to use for the Dispatcher
mailbox-requirement = org.example.MyInterface
# Dispatcher is the name of the event-based dispatcher
type = Dispatcher
# What kind of ExecutionService to use
executor = "thread-pool-executor"
# Configuration for the thread pool
thread-pool-executor {
# minimum number of threads to cap factor-based core number to
core-pool-size-min = 2
# No of core threads ... ceil(available processors * factor)
core-pool-size-factor = 2.0
# maximum number of threads to cap factor-based number to
core-pool-size-max = 10
}
# Throughput defines the maximum number of messages to be
# processed per actor before the thread jumps to the next actor.
# Set to 1 for as fair as possible.
throughput = 100
}
如上所示,可以通过配置文件中的一行代码来为Dispatcher配置Mailbox类型:
# Type of mailbox to use for the Dispatcher
mailbox-requirement = org.example.MyInterface
实际上,Akka提供了多种内置的Dispatcher类型,可以在创建自定义Dispatcher时使用。
Akka自带了多种Mailbox实现:
默认情况下,UnboundedMailbox是默认的Mailbox类型。但是,可以通过以下配置更改默认Mailbox,但需要确保所选的默认Mailbox类型与使用的Dispatcher类型相匹配。
akka.actor.default-mailbox {
mailbox-type = "akka.dispatch.SingleConsumerOnlyUnboundedMailbox"
}
可以通过混合RequiresMessageQueue特性来为特定类型的Actor关联特定类型的Mailbox。
import akka.dispatch.RequiresMessageQueue
import akka.dispatch.BoundedMessageQueueSemantics
class SomeActor extends Actor with RequiresMessageQueue[BoundedMessageQueueSemantics]
以下配置用于配置Mailbox:
bounded-mailbox {
mailbox-type = "akka.dispatch.BoundedMailbox"
mailbox-capacity = 1000
mailbox-push-timeout-time = 10s
}
akka.actor.mailbox.requirements {
"akka.dispatch.BoundedMessageQueueSemantics" = bounded-mailbox
}
值得注意的是,这个设置可能会被代码或Dispatcher的Mailbox配置部分覆盖。
https://github.com/sachabarber/SachaBarber.AkkaExamples