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://[email protected]:25251] *************** 22:14:52.810 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:51081] - Received InitJoinAck message from [Actor[akka://[email protected]:25251/system/cluster/core/daemon#313431252]] to [akka://[email protected]:51081] 22:14:52.825 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:25251] - Node [akka://[email protected]:51081] is JOINING, roles [dc-default] 22:14:52.825 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://[email protected]:51081] *************** 22:14:52.829 [ClusterSystem-akka.actor.internal-dispatcher-7] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://[email protected]:25251] - Node added [UniqueAddress(akka://[email protected]:51081,567025403336682144)] 22:14:52.858 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:51081] - Welcome from [akka://[email protected]:25251] 22:14:52.858 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://[email protected]:25251] *************** 22:14:52.858 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://[email protected]:51081] *************** 22:14:52.858 [ClusterSystem-akka.actor.internal-dispatcher-13] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://[email protected]:51081] - Node added [UniqueAddress(akka://[email protected]:25251,6076326462170320177)] 22:14:53.044 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:25251] - Leader is moving node [akka://[email protected]:51081] to [Up] 22:14:53.044 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://[email protected]:51081] *************** 22:14:53.679 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://[email protected]:51081] *************** 22:14:57.707 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:25251] - Received InitJoin message from [Actor[akka://[email protected]:25252/system/cluster/core/daemon/joinSeedNodeProcess-1#1472023843]] to [akka://[email protected]:25251] 22:14:57.707 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:25251] - Sending InitJoinAck message from node [akka://[email protected]:25251] to [Actor[akka://[email protected]: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://[email protected]:25252] - Received InitJoinAck message from [Actor[akka://[email protected]:25251/system/cluster/core/daemon#313431252]] to [akka://[email protected]:25252] 22:14:57.734 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:25251] - Node [akka://[email protected]:25252] is JOINING, roles [dc-default] 22:14:57.735 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://[email protected]:25252] *************** 22:14:57.735 [ClusterSystem-akka.actor.internal-dispatcher-26] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://[email protected]:25251] - Node added [UniqueAddress(akka://[email protected]:25252,-6913064885699273532)] 22:14:57.737 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:25252] - Welcome from [akka://[email protected]:25251] 22:14:57.737 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://[email protected]:25251] *************** 22:14:57.738 [ClusterSystem-akka.actor.internal-dispatcher-30] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://[email protected]:25252] - Node added [UniqueAddress(akka://[email protected]:25251,6076326462170320177)] 22:14:57.738 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://[email protected]:25252] *************** 22:14:57.738 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://[email protected]:51081] *************** 22:14:57.738 [ClusterSystem-akka.actor.internal-dispatcher-30] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://[email protected]:25252] - Node added [UniqueAddress(akka://[email protected]:51081,567025403336682144)] 22:14:57.740 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member joined: [akka://[email protected]:25252] *************** 22:14:57.740 [ClusterSystem-akka.actor.internal-dispatcher-16] DEBUG akka.cluster.typed.internal.receptionist.ClusterReceptionist - ClusterReceptionist [akka://[email protected]:51081] - Node added [UniqueAddress(akka://[email protected]:25252,-6913064885699273532)] 22:14:58.134 [ClusterSystem-akka.actor.default-dispatcher-3] INFO akka.cluster.Cluster - Cluster Node [akka://[email protected]:25251] - Leader is moving node [akka://[email protected]:25252] to [Up] 22:14:58.134 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://[email protected]:25252] *************** 22:14:58.755 [ClusterSystem-akka.actor.default-dispatcher-3] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://[email protected]:25252] *************** 22:14:59.146 [ClusterSystem-akka.actor.default-dispatcher-14] INFO com.learn.akka.MonitorActor$ - **************** Member is Up: [akka://[email protected]: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://[email protected]:25251", "akka://[email protected]: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) } }
轉載請註明出處,本文鏈接:https://www.uj5u.com/houduan/419.html
標籤:Scala
上一篇:akka-typed(4) - EventSourcedBehavior in action
下一篇:akka-typed(6) - cluster:group router, cluster-load-balancing
