为什么多个Akka Actors会阻止消息丢失？

Question

我最近测试了不同的Akka消息处理模块，我发现了一个有趣的现象。基本上我比较了两种情况：

1）使用Future的单个actor来处理消息。 2）每个单线程的多个actor。

从性能的角度来看，我没有发现太大的差异。但我发现，如果邮箱容量非常有限，第一种解决方案丢失数据的可能性就更高。例如，我在池中有8个线程，并且16个消息并行发送，消息队列容量为1，对于第一个解决方案，我将从第二个8消息中丢失大部分，但对于第二个解决方案，8个actor可以处理所有16条消息（有时只丢失1或2条）。

这是否意味着actor在处理当前消息时会缓存下一条消息？

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()
}

代码显示具有多线程场景的单个actor，可以自由地从ChildActor注释掉“Future”并将RouterActor的实例参数设置为8以体验多个actor。

Answer 1

您的测试有点奇怪，我无法理解您在这里尝试使用大小为1的邮箱进行测试。

每个actor当时可以处理一条消息，因此预期场景（1）丢失消息，因为当新消息到达时，actor将忙于该消息，如果队列被填满，它将丢弃消息。

您在方案1中的预期行为是什么？