Почему мой печатный актер не был перезапущен его опекуном? - PullRequest
Новая
       108

Почему мой печатный актер не был перезапущен его опекуном?

0 голосов
/ 05 июня 2018

Я экспериментирую с набранным Аккой.У меня есть фиктивный актер, который эмулирует нестабильного работника: 

import akka.actor.typed.Behavior
import akka.actor.typed.receptionist.{Receptionist, ServiceKey}
import akka.actor.typed.scaladsl.Behaviors

import scala.util.Random

object DummyActor {
  def behavior[T](serviceKey: ServiceKey[T]): Behavior[Any] = Behaviors.setup { ctx =>

    ctx.system.receptionist ! Receptionist.Register(serviceKey, ctx.self)

    ctx.log.info("Woohoo, I'm alive!")

    Thread.sleep(1000)

    if (Random.nextInt(10) > 5)
      throw new IllegalStateException("Something bad happened!")

    Behaviors.empty
  }
}

и его опекуна с маршрутизатором: 

import akka.actor.typed.{Behavior, SupervisorStrategy}
import akka.actor.typed.receptionist.ServiceKey
import akka.actor.typed.scaladsl.Behaviors

object MyCluster {

  val serviceKey: ServiceKey[String] = ServiceKey[String]("cluster")

  val strategy = SupervisorStrategy.restart

  val behavior: Behavior[String] = Behaviors.setup { ctx =>


    (1 to 5).foreach { i =>
      ctx.log.info(s"Spawning actor #$i")
      ctx.spawn(
        Behaviors.supervise(DummyActor.behavior(serviceKey))
                 .onFailure[Throwable](strategy),
        s"actor-$i"
      )
    }

    val router = ctx.spawn(RandomRouter.clusterRouter(serviceKey), "router")

    Behaviors.empty
  }
}

Мой маршрутизатор прослушивает события администратора: 

import java.util.concurrent.ThreadLocalRandom

import akka.actor.Address
import akka.actor.typed.{ActorRef, Behavior}
import akka.actor.typed.receptionist.{Receptionist, ServiceKey}
import akka.actor.typed.scaladsl.Behaviors
import akka.cluster.ClusterEvent.{ReachabilityEvent, ReachableMember, UnreachableMember}
import akka.cluster.typed.{Cluster, Subscribe}

object RandomRouter {
  private final case class WrappedReachabilityEvent(event: ReachabilityEvent)

  // same as above, but also subscribes to cluster reachability events and
  // avoids routees that are unreachable
  def clusterRouter[T](serviceKey: ServiceKey[T]): Behavior[T] =
    Behaviors.setup[Any] { ctx ⇒
      ctx.system.receptionist ! Receptionist.Subscribe(serviceKey, ctx.self)

      val cluster = Cluster(ctx.system)
      // typically you have to map such external messages into this
      // actor's protocol with a message adapter
      val reachabilityAdapter: ActorRef[ReachabilityEvent] = ctx.messageAdapter(WrappedReachabilityEvent.apply)

      cluster.subscriptions ! Subscribe(reachabilityAdapter, classOf[ReachabilityEvent])

      def routingBehavior(routees: Vector[ActorRef[T]], unreachable: Set[Address]): Behavior[Any] =
        Behaviors.receive { (ctx, msg) ⇒
          msg match {
            case serviceKey.Listing(services) ⇒
              if (services.isEmpty) {
                ctx.log.info("Found no services")
              } else {
                ctx.log.info(s"Found services: ${services.map(_.path).mkString(", ")}")
              }
              routingBehavior(services.toVector, unreachable)
            case WrappedReachabilityEvent(event) => event match {
              case UnreachableMember(m) =>
                ctx.log.warning(s"Member ${m.uniqueAddress.address} has become unreachable")
                routingBehavior(routees, unreachable + m.address)
              case ReachableMember(m) =>
                ctx.log.info(s"Member ${m.uniqueAddress.address} has become reachable again")
                routingBehavior(routees, unreachable - m.address)
            }

            case other: T @unchecked ⇒
              if (routees.isEmpty)
                Behaviors.unhandled
              else {
                val reachableRoutes =
                  if (unreachable.isEmpty) routees
                  else routees.filterNot { r => unreachable(r.path.address) }

                val i = ThreadLocalRandom.current.nextInt(reachableRoutes.size)
                reachableRoutes(i) ! other
                Behaviors.same
              }
          }
        }

      routingBehavior(Vector.empty, Set.empty)
    }.narrow[T]
}

Но я обнаружил, что когда я запускаю свой кластер, в конечном итоге я обнаружил, что некоторые актеры умерли (как и ожидалось), но не перезапущены, и у меня остались такие журналы: 

[INFO] [06/01/2018 18:11:14.242] [cluster-system-akka.actor.default-dispatcher-4] [akka://cluster-system/user/router] Found services: akka://cluster-system/user/actor-4, akka://cluster-system/user/actor-3, akka://cluster-system/user/actor-1, akka://cluster-system/user/actor-5

Почему нет MyCluster#strategy перезапуск неудачных актеров?

1 Ответ

0 голосов
/ 21 июня 2018

Код и связанные комментарии для SupervisorStrategy # restart содержат ответ.

Если есть исключение во время действия актера setup , перезапуск будет опасным из-за потенциальной возможностисоздать бесконечный цикл перезапуска.Вместо этого рекомендуется использовать restartWithBackoff для контроля запуска.

...