akka-typed(0) - typed-actor, typed messages

akka 2.6.x正式发布以来已经有好一段时间了。核心变化是typed-actor的正式启用，当然persistence,cluster等模块也有较大变化。一开始从名称估摸就是把传统any类型的消息改成强类型消息，所以想拖一段时间看看到底能对我们现有基于akka-classic的应用软件有什么深层次的影响。不过最近考虑的一些系统架构逼的我不得不立即开始akka-typed的调研，也就是说akka-classic已经无法或者很困难去实现新的系统架构，且听我道来：最近在考虑一个微服务中台。作为后台数据服务调用的唯一入口，平台应该是个分布式软件，那么采用akka-cluster目前是唯一的选择，毕竟前期搞过很多基于akka-cluster的应用软件。但是，akka-cluster-sharding只能支持一种entity actor。毕竟，由于akka-classic的消息是没有类型的，只能在收到消息后再通过类型模式匹配的方式确定应该运行的代码。所以，这个actor必须包括所有的业务逻辑处理运算。也就是说对于一个大型应用来说这就是一块巨型代码。还有，如果涉及到维护actor状态的话，比如persistenceActor，或者综合类型业务运算，那么又需要多少种类的数据结构，又怎样去维护、管理这些结构呢？对我来说这基本上是mission-impossible。实际上logom应该正符合这个中台的要求：cluster-sharding, CQRS... 抱着一种好奇的心态了解了一下lagom源码，忽然恍然大悟：这个东西是基于akka-typed的！想想看也是：如果我们可以把actor和消息类型绑在一起，那么我们就可以通过消息类型对应到某种actor。也就是说基于akka-typed，我们可以把综合性的业务划分成多个actor模块，然后我们可以指定那种actor做那些事情。当然，经过了功能细分，actor的设计也简单了许多。现在这个新的中台可以实现前台应用直接调用对应的actor处理业务了。不用多想了，这注定就是akka应用的将来，还等什么呢？

先从一个最简单的hello程序开始吧：基本上是两个actor相互交换消息。先用第一个来示范标准的actor构建过程：

  object HelloActor {

    sealed trait Request

    case class Greeting(fromWhom: String, replyTo: ActorRef[Greeter.Greeted]) extends Request

    def apply(): Behavior[Greeting] = {

      Behaviors.receive { (ctx, greeter) =>

        ctx.log.info("receive greeting from {}", greeter.fromWhom)

        greeter.replyTo ! Greeter.Greeted(s"hello ${greeter.fromWhom}!")

        Behaviors.same

      }

    }

  }

akka-typed的actor构建是通过定义它的Behavior行为实现的。特别的是类型参数Behavior[Greeting]，代表这个actor只处理Greeting类型的消息，因而是个typed-actor。akka-typed已经不支持sender()了，在消息里自带，如Greeting.replyTo。Behavior定义是通过工厂模式Behaviors实现的，看看Behaviors的定义：

/**

 * Factories for [[akka.actor.typed.Behavior]].

 */

object Behaviors {

  def setup[T](factory: ActorContext[T] => Behavior[T]): Behavior[T] 

  def withStash[T](capacity: Int)(factory: StashBuffer[T] => Behavior[T]): Behavior[T] 

  def same[T]: Behavior[T] 

  def unhandled[T]: Behavior[T] 

  def stopped[T]: Behavior[T] 

  def stopped[T](postStop: () => Unit): Behavior[T]

  def empty[T]: Behavior[T]

  def ignore[T]: Behavior[T] 

  def receive[T](onMessage: (ActorContext[T], T) => Behavior[T]): Receive[T]

  def receiveMessage[T](onMessage: T => Behavior[T]): Receive[T]

  def receivePartial[T](onMessage: PartialFunction[(ActorContext[T], T), Behavior[T]]): Receive[T] 

  def receiveMessagePartial[T](onMessage: PartialFunction[T, Behavior[T]]): Receive[T] 

  def receiveSignal[T](handler: PartialFunction[(ActorContext[T], Signal), Behavior[T]]): Behavior[T] 

  def supervise[T](wrapped: Behavior[T]): Supervise[T] 

  def withTimers[T](factory: TimerScheduler[T] => Behavior[T]): Behavior[T] 

 ...

}

上面的构建函数除返回Behavior[T]外还有Receive[T]和Supervise[T]，这两个类型是什么？它们还是Behavior[T]：

  trait Receive[T] extends Behavior[T] {

    def receiveSignal(onSignal: PartialFunction[(ActorContext[T], Signal), Behavior[T]]): Behavior[T]

  }

  def supervise[T](wrapped: Behavior[T]): Supervise[T] =

    new Supervise[T](wrapped)

  private final val ThrowableClassTag = ClassTag(classOf[Throwable])

  final class Supervise[T] private[akka] (val wrapped: Behavior[T]) extends AnyVal {

    /** Specify the [[SupervisorStrategy]] to be invoked when the wrapped behavior throws. */

    def onFailure[Thr <: Throwable: ClassTag](strategy: SupervisorStrategy): Behavior[T] = {

      val tag = classTag[Thr]

      val effectiveTag = if (tag == ClassTag.Nothing) ThrowableClassTag else tag

      Supervisor(Behavior.validateAsInitial(wrapped), strategy)(effectiveTag)

    }

  }

注意，Supervise.onFailure返回了Behavior[T]。

helloActor的Behavior是通过Behaviors.receive构建的。还可以用setup,receiveMessage来构建。注意：构建函数的入参数也是Behavior[T]，所以这些构造器可以一层层嵌套着使用。setup,receive为函数内层提供了ActorContext, withTimers提供TimerScheduler[T]。那么我可以把HelloActor的功能再完善点，加个监管策略SupervisorStrategy：

  object HelloActor {

    sealed trait Request

    case class Greeting(fromWhom: String, replyTo: ActorRef[Greeter.Greeted]) extends Request

    def apply(): Behavior[Greeting] = {

      Behaviors.supervise(

        Behaviors.receive[Greeting] { (ctx, greeter) =>

          ctx.log.info("receive greeting from {}", greeter.fromWhom)

          greeter.replyTo ! Greeter.Greeted(s"hello ${greeter.fromWhom}!")

          Behaviors.same

        }

      ).onFailure(SupervisorStrategy.restartWithBackoff(.seconds, .minute, 0.20))

    }

  }

在akka-typed里，actor监管已经从父辈转到自身。再就是增加了BackOff-SupervisorStrategy，不需要独立的BackOffSupervisor actor了。

再看看另一个Greeter：

 object Greeter {

    sealed trait Response

    case class Greeted(hello: String) extends Response

    def apply(): Behavior[Greeted] = {

      Behaviors.setup ( ctx =>

        Behaviors.receiveMessage { message =>

          ctx.log.info(message.hello)

          Behaviors.same

        }

      )

    }

  }

这个跟HelloActor没什么不同，不过用了setup,receiveMessage套装。值得注意的是Greeter负责处理Greeted消息，这是一个不带sender ActorRef的类型，意味着处理这类消息后不需要答复消息发送者。

然后还需要一个actor来构建上面两个actor实例，启动对话：

 object GreetStarter {

    sealed trait Command

    case class SayHiTo(whom: String) extends Command

    case class RepeatedGreeting(whom: String, interval: FiniteDuration) extends Command

    def apply(): Behavior[Command] = {

      Behaviors.setup[Command] { ctx =>

        val helloActor = ctx.spawn(HelloActor(), "hello-actor")

        val greeter = ctx.spawn(Greeter(), "greeter")

        Behaviors.withTimers { timer =>

          new GreetStarter(

            helloActor,greeter,ctx,timer)

            .repeatGreeting(,)

        }

      }

    }

  }

  class GreetStarter private (

     helloActor: ActorRef[HelloActor.Greeting],

     greeter: ActorRef[Greeter.Greeted],

     ctx: ActorContext[GreetStarter.Command],

     timer: TimerScheduler[GreetStarter.Command]){

    import GreetStarter._

    private def repeatGreeting(count: Int, max: Int): Behavior[Command] =

       Behaviors.receiveMessage { msg =>

         msg match {

           case RepeatedGreeting(whom, interval) =>

             ctx.log.info2("start greeting to {} with interval {}", whom, interval)

             timer.startSingleTimer(SayHiTo(whom), interval)

             Behaviors.same

           case SayHiTo(whom) =>

             ctx.log.info2("{}th time greeting to {}",count,whom)

             if (max == count)

               Behaviors.stopped

             else {

               helloActor ! HelloActor.Greeting(whom, greeter)

               repeatGreeting(count + , max)

             }

         }

       }

  }

上面这个例子有点复杂，逻辑也有些问题，主要是为了示范一种函数式actor构建模式及actor状态转换虚构出来的。akka-typed已经不再支持become方法了。

最后，需要一个相当于main这么一个顶层的程序：

  def main(args: Array[String]) {

    val man: ActorSystem[GreetStarter.Command] = ActorSystem(GreetStarter(), "greetDemo")

    man ! GreetStarter.RepeatedGreeting("Tiger",.seconds)

    man ! GreetStarter.RepeatedGreeting("Peter",.seconds)

    man ! GreetStarter.RepeatedGreeting("Susanna",.seconds)

  }

akka-classic的顶级actor，即： /users是由系统默认创建的。akka-typed需要用户提供这个顶层actor。这个是在ActorSystem的第一个参数指定的。我们再看看akka-typed的ActorSystem的构建函数：

object ActorSystem {

  /**

   * Scala API: Create an ActorSystem

   */

  def apply[T](guardianBehavior: Behavior[T], name: String): ActorSystem[T] =

    createInternal(name, guardianBehavior, Props.empty, ActorSystemSetup.create(BootstrapSetup()))

  /**

   * Scala API: Create an ActorSystem

   */

  def apply[T](guardianBehavior: Behavior[T], name: String, config: Config): ActorSystem[T] =

    createInternal(name, guardianBehavior, Props.empty, ActorSystemSetup.create(BootstrapSetup(config)))

  /**

   * Scala API: Create an ActorSystem

   */

  def apply[T](guardianBehavior: Behavior[T], name: String, config: Config, guardianProps: Props): ActorSystem[T] =

    createInternal(name, guardianBehavior, guardianProps, ActorSystemSetup.create(BootstrapSetup(config)))

...

}

其中一个apply与akka-classic的ActorSystem构建方式很相似：

  def main(args: Array[String]) {

    val config = ConfigFactory.load("application.conf")

    val man: ActorSystem[GreetStarter.Command] = ActorSystem(GreetStarter(), "greetDemo",config)

    man ! GreetStarter.RepeatedGreeting("Tiger",.seconds)

    man ! GreetStarter.RepeatedGreeting("Peter",.seconds)

    man ! GreetStarter.RepeatedGreeting("Susanna",.seconds)

  }

下面是本次讨论的完整源代码：

build.sbt

name := "learn-akka-typed"

version := "0.1"

scalaVersion := "2.13.2"

lazy val akkaVersion = "2.6.5"

libraryDependencies ++= Seq(

  "com.typesafe.akka" %% "akka-actor-typed"            % akkaVersion,

  "ch.qos.logback"     % "logback-classic"             % "1.2.3"

)

fork in Test := true

Lesson01.scala

import akka.actor.typed._

import scaladsl._

import scala.concurrent.duration._

import com.typesafe.config._

object Lesson01 {

  object HelloActor {

    sealed trait Request

    case class Greeting(fromWhom: String, replyTo: ActorRef[Greeter.Greeted]) extends Request

    def apply(): Behavior[Greeting] = {

      Behaviors.supervise(

        Behaviors.receive[Greeting] { (ctx, greeter) =>

          ctx.log.info("receive greeting from {}", greeter.fromWhom)

          greeter.replyTo ! Greeter.Greeted(s"hello ${greeter.fromWhom}!")

          Behaviors.same

        }

      ).onFailure(SupervisorStrategy.restartWithBackoff(.seconds, .minute, 0.20))

    }

  }

  object Greeter {

    sealed trait Response

    case class Greeted(hello: String) extends Response

    def apply(): Behavior[Greeted] = {

      Behaviors.setup ( ctx =>

        Behaviors.receiveMessage { message =>

          ctx.log.info(message.hello)

          Behaviors.same

        }

      )

    }

  }

  object GreetStarter {

    sealed trait Command

    case class SayHiTo(whom: String) extends Command

    case class RepeatedGreeting(whom: String, interval: FiniteDuration) extends Command

    def apply(): Behavior[Command] = {

      Behaviors.setup[Command] { ctx =>

        val helloActor = ctx.spawn(HelloActor(), "hello-actor")

        val greeter = ctx.spawn(Greeter(), "greeter")

        Behaviors.withTimers { timer =>

          new GreetStarter(

            helloActor,greeter,ctx,timer)

            .repeatGreeting(,)

        }

      }

    }

  }

  class GreetStarter private (

     helloActor: ActorRef[HelloActor.Greeting],

     greeter: ActorRef[Greeter.Greeted],

     ctx: ActorContext[GreetStarter.Command],

     timer: TimerScheduler[GreetStarter.Command]){

    import GreetStarter._

    private def repeatGreeting(count: Int, max: Int): Behavior[Command] =

       Behaviors.receiveMessage { msg =>

         msg match {

           case RepeatedGreeting(whom, interval) =>

             ctx.log.info2("start greeting to {} with interval {}", whom, interval)

             timer.startSingleTimer(SayHiTo(whom), interval)

             Behaviors.same

           case SayHiTo(whom) =>

             ctx.log.info2("{}th time greeting to {}",count,whom)

             if (max == count)

               Behaviors.stopped

             else {

               helloActor ! HelloActor.Greeting(whom, greeter)

               repeatGreeting(count + , max)

             }

         }

       }

  }

  def main(args: Array[String]) {

    val config = ConfigFactory.load("application.conf")

    val man: ActorSystem[GreetStarter.Command] = ActorSystem(GreetStarter(), "greetDemo",config)

    man ! GreetStarter.RepeatedGreeting("Tiger",.seconds)

    man ! GreetStarter.RepeatedGreeting("Peter",.seconds)

    man ! GreetStarter.RepeatedGreeting("Susanna",.seconds)

  }

}

巴特西

akka-typed(0) - typed-actor, typed messages

最新文章

热门文章