Я недавно тестирую разницу в модуле обработки сообщений Akka, и я обнаружил интересное явление. В основном я сравнил две ситуации:
1) Один актер с Future для обработки сообщения.
2) Несколько актеров с одной нитью для каждого.
С точки зрения перформанса, я не нашел особой разницы. Но я обнаружил, что первое решение имеет большую вероятность потери данных, если емкость почтового ящика очень ограничена. Например, у меня 8 потоков в пуле и 16 параллельных отправок сообщений, а емкость очереди сообщений равна 1, для первого решения я потеряю большинство из них из вторых 8 сообщений, но для второго решения 8 участников могут обрабатывать все 16 сообщений (иногда потерял только 1 или 2).
Означает ли это, что субъект будет кэшировать следующее сообщение во время обработки текущего?
import java.util.Calendar
import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, DeadLetter, Props, Terminated}
import akka.routing.{ActorRefRoutee, RoundRobinRoutingLogic, Router, RoutingLogic}
import com.typesafe.config.ConfigFactory
import scala.concurrent.{Await, ExecutionContext, Future}
import scala.concurrent.duration._
object Example_6_Backpressure extends App{
lazy val akkaSystemConfiguration = ConfigFactory.parseString(
"""
|akka.actor.default-mailbox {
| mailbox-type = "akka.dispatch.UnboundedMailbox"
|}
|
|akka.actor.bounded-mailbox {
| mailbox-type = "akka.dispatch.BoundedMailbox"
| mailbox-capacity = 1
| mailbox-push-timeout-time = 100ms
|}
|
|akka.actor.default-dispatcher {
| type = Dispatcher
| throughput = 100
| executor = "fork-join-executor"
|
| fork-join-executor {
| parallelism-min = 1
| parallelism-factor = 1 # 8 core cpu
| parallelism-max = 8
| }
|}
""".stripMargin)
final case class PayLoad[T](msg:T)
final case class Shutdown()
object RouterActor {
def apply(childProps: => Props, instance:Int = 1, rl: RoutingLogic = RoundRobinRoutingLogic() ) = {
Props(new RouterActor(childProps, instance, rl))
}
}
class RouterActor(childProps: => Props, instance:Int, rl: RoutingLogic ) extends Actor with ActorLogging {
override def preStart() = log.debug(s"${self.path}: Pre-Start")
override def postStop() = log.debug(s"${self.path}: Post-Stop")
var router = Router(rl, Vector.fill(instance) {
val actor = context.actorOf(childProps)
addWatcher(actor)
ActorRefRoutee(actor)
})
def addWatcher(actor:ActorRef): Unit = {
val watcher = context.actorOf(Props(classOf[Watcher], actor))
context.system.eventStream.subscribe(watcher, classOf[DeadLetter])
}
def receive: Actor.Receive = {
case t:Shutdown =>
router.routees.foreach { r =>
context.stop(r.asInstanceOf[ActorRefRoutee].ref)
router.removeRoutee(r)
}
context.system.terminate()
case p:PayLoad[_] =>
log.debug(s"${self.path}: route to child actor")
router.route(p, sender())
}
}
class Watcher(target: ActorRef) extends Actor with ActorLogging {
private val targetPath = target.path
override def preStart() {
context watch target
}
def receive: Actor.Receive = {
case d: DeadLetter =>
if(d.recipient.path.equals(targetPath)) {
log.info(s"Timed out message: ${d.message.toString}")
// TODO: ...
}
}
}
object ChildActor{
def apply() = Props[ChildActor]
}
class ChildActor() extends Actor with ActorLogging {
override def preStart() = log.debug(s"${self.path}: Pre-Start")
override def postStop() = log.debug(s"${self.path}: Post-Stop")
override def receive: Receive = {
case msg => {
Future {
println(s"${Calendar.getInstance.getTimeInMillis} - [Thread-${Thread.currentThread.getId}] - ${self.path}: $msg")
Thread.sleep(1000)
}(context.dispatcher)
}
}
}
object BackPressureExample {
def apply(): Unit = {
val system = ActorSystem("testSystem", akkaSystemConfiguration)
val rootRef = system.actorOf(
RouterActor( ChildActor().withMailbox("akka.actor.bounded-mailbox"), instance = 1), "actor-router"
)
rootRef ! PayLoad("Hello-1!")
rootRef ! PayLoad("Hello-2!")
rootRef ! PayLoad("Hello-3!")
rootRef ! PayLoad("Hello-4!")
rootRef ! PayLoad("Hello-5!")
rootRef ! PayLoad("Hello-6!")
rootRef ! PayLoad("Hello-7!")
rootRef ! PayLoad("Hello-8!")
rootRef ! PayLoad("Hello-9!")
rootRef ! PayLoad("Hello-10!")
rootRef ! PayLoad("Hello-11!")
rootRef ! PayLoad("Hello-12!")
rootRef ! PayLoad("Hello-13!")
rootRef ! PayLoad("Hello-14!")
rootRef ! PayLoad("Hello-15!")
rootRef ! PayLoad("Hello-16!")
Thread.sleep(5100)
rootRef ! new Shutdown
Await.result(system.terminate(), 10 second)
}
}
BackPressureExample()
}
В коде показан один актер со сценарием с несколькими потоками, он может закомментировать «Будущее» из ChildActor и установить для параметра экземпляров RouterActor значение 8, чтобы использовать нескольких акторов один.