kka-typed(5)

 akka-cluster对每个节点的每种状态变化都会在系统消息队列里发布相关的事件。通过订阅有关节点状态变化的消息就可以获取每个节点的状态。这部分已经在之前关于akka-cluster的讨论里介绍过了。由于akka-typed里采用了新的消息交流协议，而系统消息的发布和订阅也算是消息交换，也受交流协议约束。所以想通过重写以前示范的ClusterMemberStatus来了解一下akka-typed环境下节点状态变化消息监听的一些机制。

我们需要一个actor来订阅系统发布的节点状态变化消息。这里涉及到系统、actor两端的信息交流。假设向系统订阅是一种消息的发送，那么得到的节点状态变化消息就是系统的response了。先看看actor里的消息定义：

object MonitorActor {
  sealed trait ClusterEvent
  private case class MemberStatus(event: MemberEvent) extends ClusterEvent
  private case class ReachStatus(event: ReachabilityEvent) extends ClusterEvent

  def apply(): Behavior[ClusterEvent] = Behaviors.setup[ClusterEvent] { ctx =>
    val memberEventAdapter: ActorRef[MemberEvent] = ctx.messageAdapter(MemberStatus)
    val reachEventAdapter: ActorRef[ReachabilityEvent] = ctx.messageAdapter(ReachStatus)
    Cluster(ctx.system).subscriptions ! Subscribe(memberEventAdapter,classOf[MemberEvent])
    Cluster(ctx.system).subscriptions ! Subscribe(reachEventAdapter,classOf[ReachabilityEvent])

...
}

首先，response 分为 MemberEvent, ReachabilityEvent两种。MonitorActor处理的消息类型是ClusterEvent。为了处理系统返回的response类型，即MemberEvent,ReachabilityEvent，必须提供这两种类型到ClusterEvent的转换。通过ctx.messageAdapter登记MemberEvent -> MemberStatus, ReachabilityEvent -> ReachStatus两种类型转换机制使MonitorActor可以接收到MemberStatus, ReachStatus两种消息：

object MonitorActor {
  sealed trait ClusterEvent
  private case class MemberStatus(event: MemberEvent) extends ClusterEvent
  private case class ReachStatus(event: ReachabilityEvent) extends ClusterEvent

  def apply(): Behavior[ClusterEvent] = Behaviors.setup[ClusterEvent] { ctx =>
    val memberEventAdapter: ActorRef[MemberEvent] = ctx.messageAdapter(MemberStatus)
    val reachEventAdapter: ActorRef[ReachabilityEvent] = ctx.messageAdapter(ReachStatus)
    Cluster(ctx.system).subscriptions ! Subscribe(memberEventAdapter,classOf[MemberEvent])
    Cluster(ctx.system).subscriptions ! Subscribe(reachEventAdapter,classOf[ReachabilityEvent])
    Behaviors.receiveMessage { event =>
      event match {
        case MemberStatus(status) =>
          status match {
            case MemberJoined(member) =>
              ctx.log.info("**************** Member joined: [{}] ***************", member.address)
            case MemberJoined(member) =>
              ctx.log.info("**************** Member joined: [{}] ***************", member.address)
            case MemberUp(member) =>
              ctx.log.info("**************** Member is Up: [{}] ***************", member.address)
            case MemberRemoved(member, previousStatus) =>
              ctx.log.info("**************** Member is Removed: [{}] after {} ***************",
                member.address, previousStatus)
            case MemberLeft(member) =>
              ctx.log.info("**************** Member left: [{}] ***************", member.address)
            case MemberExited(member) =>
              ctx.log.info("**************** Member exited: [{}] ***************", member.address)
            case _: MemberEvent => // ignore
          }
        case ReachStatus(status) =>
            status match {
              case UnreachableMember(member) =>
                ctx.log.info("**************** Member detected as unreachable: [{}] ***************", member)
              case ReachableMember(member) =>
                ctx.log.info("**************** Member back to reachable: [{}] ***************", member)
            }
      }
      Behaviors.same
    }
  }
}

还需要一个actor, 什么都不干。存粹构建一个MonitorActor：

object RootActor {
  def apply(): Behavior[Nothing] = Behaviors.setup[Nothing] {ctx =>
     ctx.spawn(MonitorActor(),"listner")
     Behaviors.empty
  }
}

好了，看看main是怎么实现的吧：

object ClusterMemberStatus {
  import com.typesafe.config.ConfigFactory
  def main(args: Array[String]): Unit = {
    val ports =
      if (args.isEmpty)
        Seq(25251, 25252, 0)
      else
        args.toSeq.map(_.toInt)
    ports.foreach { port =>
      startup(port)
    }

  }

  def startup(port: Int): Unit = {
    val config = ConfigFactory.parseString(s"""
      akka.remote.artery.canonical.port=$port
      """).withFallback(ConfigFactory.load("cluster.conf"))
    ActorSystem[Nothing](RootActor(),"ClusterSystem",config)
  }

}

下面是测试结果显示：

22:14:52.755 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:25251] ***************
22:14:52.810 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:51081] - Received InitJoinAck message from [Actor[akka://ClusterSystem@127.0.0.1:25251/system/cluster/core/daemon#313431252]] to [akka://ClusterSystem@127.0.0.1:51081]
22:14:52.825 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:25251] - Node [akka://ClusterSystem@127.0.0.1:51081] is JOINING, roles [dc-default]
22:14:52.825 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://ClusterSystem@127.0.0.1:51081] ***************
22:14:52.829 [ClusterSystem-akka.actor.internal-dispatcher-7] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://ClusterSystem@127.0.0.1:25251] - Node added [UniqueAddress(akka://ClusterSystem@127.0.0.1:51081,567025403336682144)]
22:14:52.858 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:51081] - Welcome from [akka://ClusterSystem@127.0.0.1:25251]
22:14:52.858 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:25251] ***************
22:14:52.858 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://ClusterSystem@127.0.0.1:51081] ***************
22:14:52.858 [ClusterSystem-akka.actor.internal-dispatcher-13] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://ClusterSystem@127.0.0.1:51081] - Node added [UniqueAddress(akka://ClusterSystem@127.0.0.1:25251,6076326462170320177)]
22:14:53.044 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:25251] - Leader is moving node [akka://ClusterSystem@127.0.0.1:51081] to [Up]
22:14:53.044 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:51081] ***************
22:14:53.679 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:51081] ***************
22:14:57.707 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:25251] - Received InitJoin message from [Actor[akka://ClusterSystem@127.0.0.1:25252/system/cluster/core/daemon/joinSeedNodeProcess-1#1472023843]] to [akka://ClusterSystem@127.0.0.1:25251]
22:14:57.707 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:25251] - Sending InitJoinAck message from node [akka://ClusterSystem@127.0.0.1:25251] to [Actor[akka://ClusterSystem@127.0.0.1:25252/system/cluster/core/daemon/joinSeedNodeProcess-1#1472023843]] (version [2.6.5])
22:14:57.732 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:25252] - Received InitJoinAck message from [Actor[akka://ClusterSystem@127.0.0.1:25251/system/cluster/core/daemon#313431252]] to [akka://ClusterSystem@127.0.0.1:25252]
22:14:57.734 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:25251] - Node [akka://ClusterSystem@127.0.0.1:25252] is JOINING, roles [dc-default]
22:14:57.735 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://ClusterSystem@127.0.0.1:25252] ***************
22:14:57.735 [ClusterSystem-akka.actor.internal-dispatcher-26] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://ClusterSystem@127.0.0.1:25251] - Node added [UniqueAddress(akka://ClusterSystem@127.0.0.1:25252,-6913064885699273532)]
22:14:57.737 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:25252] - Welcome from [akka://ClusterSystem@127.0.0.1:25251]
22:14:57.737 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:25251] ***************
22:14:57.738 [ClusterSystem-akka.actor.internal-dispatcher-30] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://ClusterSystem@127.0.0.1:25252] - Node added [UniqueAddress(akka://ClusterSystem@127.0.0.1:25251,6076326462170320177)]
22:14:57.738 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://ClusterSystem@127.0.0.1:25252] ***************
22:14:57.738 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:51081] ***************
22:14:57.738 [ClusterSystem-akka.actor.internal-dispatcher-30] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://ClusterSystem@127.0.0.1:25252] - Node added [UniqueAddress(akka://ClusterSystem@127.0.0.1:51081,567025403336682144)]
22:14:57.740 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://ClusterSystem@127.0.0.1:25252] ***************
22:14:57.740 [ClusterSystem-akka.actor.internal-dispatcher-16] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://ClusterSystem@127.0.0.1:51081] - Node added [UniqueAddress(akka://ClusterSystem@127.0.0.1:25252,-6913064885699273532)]
22:14:58.134 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://ClusterSystem@127.0.0.1:25251] - Leader is moving node [akka://ClusterSystem@127.0.0.1:25252] to [Up]
22:14:58.134 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:25252] ***************
22:14:58.755 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:25252] ***************
22:14:59.146 [ClusterSystem-akka.actor.default-dispatcher-14] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://ClusterSystem@127.0.0.1:25252] ***************

下面是本次示范的全部源代码：

build.sbt

name := "learn-akka-typed"

version := "0.1"

scalaVersion := "2.13.1"
scalacOptions in Compile ++= Seq("-deprecation", "-feature", "-unchecked", "-Xlog-reflective-calls", "-Xlint")
javacOptions in Compile ++= Seq("-Xlint:unchecked", "-Xlint:deprecation")

val AkkaVersion = "2.6.5"
val AkkaPersistenceCassandraVersion = "1.0.0"


libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-cluster-sharding-typed" % AkkaVersion,
  "com.typesafe.akka" %% "akka-persistence-typed" % AkkaVersion,
  "com.typesafe.akka" %% "akka-persistence-query" % AkkaVersion,
  "com.typesafe.akka" %% "akka-serialization-jackson" % AkkaVersion,
  "com.typesafe.akka" %% "akka-persistence-cassandra" % AkkaPersistenceCassandraVersion,
  "com.typesafe.akka" %% "akka-slf4j" % AkkaVersion,
  "ch.qos.logback"     % "logback-classic"             % "1.2.3"
)

cluster.conf

akka {
  actor {
    provider = cluster

    serialization-bindings {
      "com.learn.akka.CborSerializable" = jackson-cbor
    }
  }
  remote {
    artery {
      canonical.hostname = "127.0.0.1"
      canonical.port = 0
    }
  }
  cluster {
    seed-nodes = [
      "akka://ClusterSystem@127.0.0.1:25251",
      "akka://ClusterSystem@127.0.0.1:25252"]
  }
}

ClusterMemberStatus.scala

package com.learn.akka
import akka.actor.typed._
import akka.actor.typed.scaladsl.Behaviors
import akka.cluster.ClusterEvent._
import akka.cluster.typed.Subscribe
import akka.cluster.typed.Cluster
import akka.actor.typed.ActorSystem

object MonitorActor {
  sealed trait ClusterEvent
  private case class MemberStatus(event: MemberEvent) extends ClusterEvent
  private case class ReachStatus(event: ReachabilityEvent) extends ClusterEvent

  def apply(): Behavior[ClusterEvent] = Behaviors.setup[ClusterEvent] { ctx =>
    val memberEventAdapter: ActorRef[MemberEvent] = ctx.messageAdapter(MemberStatus)
    val reachEventAdapter: ActorRef[ReachabilityEvent] = ctx.messageAdapter(ReachStatus)
    Cluster(ctx.system).subscriptions ! Subscribe(memberEventAdapter,classOf[MemberEvent])
    Cluster(ctx.system).subscriptions ! Subscribe(reachEventAdapter,classOf[ReachabilityEvent])
    Behaviors.receiveMessage { event =>
      event match {
        case MemberStatus(status) =>
          status match {
            case MemberJoined(member) =>
              ctx.log.info("**************** Member joined: [{}] ***************", member.address)
            case MemberJoined(member) =>
              ctx.log.info("**************** Member joined: [{}] ***************", member.address)
            case MemberUp(member) =>
              ctx.log.info("**************** Member is Up: [{}] ***************", member.address)
            case MemberRemoved(member, previousStatus) =>
              ctx.log.info("**************** Member is Removed: [{}] after {} ***************",
                member.address, previousStatus)
            case MemberLeft(member) =>
              ctx.log.info("**************** Member left: [{}] ***************", member.address)
            case MemberExited(member) =>
              ctx.log.info("**************** Member exited: [{}] ***************", member.address)
            case _: MemberEvent => // ignore
          }
        case ReachStatus(status) =>
            status match {
              case UnreachableMember(member) =>
                ctx.log.info("**************** Member detected as unreachable: [{}] ***************", member)
              case ReachableMember(member) =>
                ctx.log.info("**************** Member back to reachable: [{}] ***************", member)
            }
      }
      Behaviors.same
    }
  }
}
object RootActor {
  def apply(): Behavior[Nothing] = Behaviors.setup[Nothing] {ctx =>
     ctx.spawn(MonitorActor(),"listner")
     Behaviors.empty
  }
}
object ClusterMemberStatus {
  import com.typesafe.config.ConfigFactory
  def main(args: Array[String]): Unit = {
    val ports =
      if (args.isEmpty)
        Seq(25251, 25252, 0)
      else
        args.toSeq.map(_.toInt)
    ports.foreach { port =>
      startup(port)
    }

  }

  def startup(port: Int): Unit = {
    val config = ConfigFactory.parseString(s"""
      akka.remote.artery.canonical.port=$port
      """).withFallback(ConfigFactory.load("cluster.conf"))
    ActorSystem[Nothing](RootActor(),"ClusterSystem",config)
  }

}