经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Scala » 查看文章
Akka-CQRS（5）- CQRS Writer Actor 部署和测试
来源：cnblogs　　作者：雪川大虫　　时间：2019/4/8 12:46:23　　对本文有异议

   上篇我们做了一个WriterActor的例子，主要目的是示范WriterActor如何作为集群分片用persistentActor特性及event-sourcing模式实现CQRS的写功能。既然是集群分片，那么我们就在这篇讲讲WriterActor的部署和测试，因为这个里面还是有些值得注意的地方。下面是一段WriteActor，即集群分片（cluster-sharding）的部署代码：

  1.     ClusterSharding(system).start(
  2.       typeName = shardName,
  3.       entityProps = writerProps,
  4.       settings = cpsSettings,
  5.       extractEntityId = getPOSId,
  6.       extractShardId = getShopId,
  7.       allocationStrategy = ClusterSharding(system).defaultShardAllocationStrategy(cpsSettings),
  8.       handOffStopMessage = PassivatePOS
  9.     )

注意带handOffStopMessage参数的start函数必须同时提供allocationStrategy。这个参数提供了passivation消息类型。

整个集群分片部署代码如下：

  1. object POSRouter extends LogSupport {
  2.   def main(args: Array[String]) {
  3.     import WriterActor._
  4.     import Commands._
  6.     val argsPat = "(.*):(.*)".r
  7.     val (host, port) = args(0) match {
  8.       case argsPat(h, p) => (h, p)
  9.       case _ => ("localhost", "2551")
  10.     }
  12.     val config = ConfigFactory.parseString("akka.remote.netty.tcp.port=\"" + port + "\"")
  13.       .withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.hostname=\"" + host + "\""))
  14.       //roles can be deployed on this node
  15.       .withFallback(ConfigFactory.parseString("akka.cluster.roles = [poswriter]"))
  16.       .withFallback(ConfigFactory.load())
  18.     log.info(s"******* hostname = $host,  port = $port *******")
  20.     val shardName = "POSShard"
  22.     case class POSMessage(id: Long, cmd: POSCommand) {
  23.       def shopId = id.toString.head.toString
  25.       def posId = id.toString
  26.     }
  28.     val getPOSId: ShardRegion.ExtractEntityId = {
  29.       case posCommand: POSMessage => (posCommand.posId, posCommand.cmd)
  30.     }
  31.     val getShopId: ShardRegion.ExtractShardId = {
  32.       case posCommand: POSMessage => posCommand.shopId
  33.     }
  36.     val system = ActorSystem("cloud-pos-server", config)
  37.     val role = "poswriter"    //role of this shard
  38.     val cpsSettings = ClusterShardingSettings(system).withRole(role) //.withPassivateIdleAfter(10 minutes)
  40.     ClusterSharding(system).start(
  41.       typeName = shardName,
  42.       entityProps = writerProps,
  43.       settings = cpsSettings,
  44.       extractEntityId = getPOSId,
  45.       extractShardId = getShopId,
  46.       allocationStrategy = ClusterSharding(system).defaultShardAllocationStrategy(cpsSettings),
  47.       handOffStopMessage = PassivatePOS
  48.     )
  50.     system.actorOf(ClusterMonitor.props, "cps-cluster-monitor")
  52.   }
  53. }

以上有几个参数需要特别注意：host和port是从main的参数解析出来的如192.168.11.162:2551，代表本节点的host和port。akka.cluster.roles代表本节点支持的角色，这里poswriter是其中之一。而ClusterShardingSettings(system).withRole("poswriter")代表这个分片shard只能在支持poswriter角色的节点上部署。如果搞错了运行时你会发现Sharding无法启动。上面这段程序代表本节点支持poswriter角色。在本节点（输入的IP地址）部署了一个名称为“POSShard”的cluster-sharding，它具备poswriter角色。

如果我在多部机器上运行这段代码，输入当前机器的IP+PORT就代表在这么多台机器上都部署了“POSShard”分片。上面的ClusterMonitor是个集群状态监控actor:

  1. package sdp.cluster.monitor
  3. import akka.actor._
  4. import akka.cluster.ClusterEvent._
  5. import akka.cluster._
  6. import sdp.logging.LogSupport
  8. object ClusterMonitor {
  9.   def props = Props(new ClusterMonitor)
  10. }
  12. class ClusterMonitor extends Actor with LogSupport {
  13.   val cluster = Cluster(context.system)
  14.   override def preStart(): Unit = {
  15.     cluster.subscribe(self,initialStateMode = InitialStateAsEvents
  16.       ,classOf[MemberEvent],classOf[UnreachableMember])  //订阅集群状态转换信息
  17.     super.preStart()
  18.   }
  20.   override def postStop(): Unit = {
  21.     cluster.unsubscribe(self)    //取消订阅
  22.     super.postStop()
  23.   }
  25.   override def receive: Receive = {
  26.     case MemberJoined(member) =>
  27.       log.info(s"Member is Joining: {${member.address}}")
  28.     case MemberUp(member) =>
  29.       log.info(s"Member is Up: {${member.address}}")
  30.     case MemberLeft(member) =>
  31.       log.info(s"Member is Leaving: {${member.address}}")
  32.     case MemberExited(member) =>
  33.       log.info(s"Member is Exiting: {${member.address}}")
  34.     case MemberRemoved(member, previousStatus) =>
  35.       log.info(
  36.         s"Member is Removed: {${member.address}} after {${previousStatus}")
  37.     case UnreachableMember(member) =>
  38.       log.info(s"Member detected as unreachable: {${member.address}}")
  39.       cluster.down(member.address)      //手工驱除，不用auto-down
  40.     case _: MemberEvent => // ignore
  41.   }
  43. }

有了它我们可以监视集群节点连接状态。

好了，现在假设我们在几台机器组成的集群各节点上都部署了“POSShard”分片，那么就设计个客户端来向这个“POSShard”分片发送POSMessage：

  1.     case class POSMessage(id: Long, cmd: POSCommand) {
  2.       def shopId = id.toString.head.toString
  4.       def posId = id.toString
  5.     }
  7.     val getPOSId: ShardRegion.ExtractEntityId = {
  8.       case posCommand: POSMessage => (posCommand.posId, posCommand.cmd)
  9.     }
  10.     val getShopId: ShardRegion.ExtractShardId = {
  11.       case posCommand: POSMessage => posCommand.shopId
  12.     }

这个客户端必须考虑以下几点：它必须在同一个集群，也就是它也是集群其中一个节点，否则无法和其它部署了“POSShard”分片的节点进行信息交流。但它又不能同处与部署了“POSShard”的节点，因为remote的hostname和port已经被占用。所以只能把客户端放在一个没有部署“POSShard”的节点上，然后用ClusterSharding(system).startProxy来启动一个分片中介：

  1.    //no shard deployed on this node  2558, use proxy
  2.     val posHandler = ClusterSharding(system).startProxy(
  3.       typeName = shardName,
  4.       role = Some("poswriter"),
  5.       extractEntityId = getPOSId,
  6.       extractShardId = getShopId
  7.     )
  9.     //val posHandler = ClusterSharding(system).shardRegion(shardName)
  11.     system.actorOf(POSClient.props(posHandler), "pos-client")

注意这个proxy的role必须是Some("poswriter")，只有这样才能调用其它节点上的”POSShard“，因为它们的角色都是“poswriter”。与WriterActor交互的必须是个actor，因为WriterActor会用sender()返回结果，这个sender()是个ActorRef：

  1. object POSClient {
  2.   def props(pos: ActorRef) = Props(new POSClient(pos))
  3. }
  4. class POSClient(posHandler: ActorRef)  extends Actor with LogSupport {
  6.   override def receive: Receive = {
  7.     case msg @ POSMessage(_,_) => posHandler ! msg
  8.     case resp: POSResponse  =>
  9.       log.info(s"response from server: $resp")
  10.   }
  11. }

我们可用下面的方式来指挥WriterActor:

  1.     val posref = system.actorOf(POSClient.props(posHandler), "pos-client")
  2.     
  3.     posref ! POSMessage(1021, LogSales(SALESTYPE.plu, "", apple.code, 1, 0))
  4.     posref ! POSMessage(2021, LogSales(SALESTYPE.plu, "", pineapple.code, 2, 0))
  5.     posref ! POSMessage(3021, LogSales(SALESTYPE.plu, "", banana.code, 1, 0))
  6.     posref ! POSMessage(4021, LogSales(SALESTYPE.plu, "", grape.code, 3, 0))
  7.     posref ! POSMessage(4021,Subtotal)

 

下面是服务端分片部署源代码：

resources/application.conf

  1. akka.actor.warn-about-java-serializer-usage = off
  2. akka.log-dead-letters-during-shutdown = off
  3. akka.log-dead-letters = off
  5. akka {
  6.   loglevel = INFO
  7.   actor {
  8.     provider = "cluster"
  9.   }
  11.   remote {
  12.     log-remote-lifecycle-events = off
  13.     netty.tcp {
  14.       hostname = "127.0.0.1"
  15.       port = 0
  16.     }
  17.   }
  19.   cluster {
  20.     seed-nodes = [
  21.       "akka.tcp://cloud-pos-server@192.168.11.162:2551"]
  22.     log-info = off
  23.     sharding {
  24.       role = "poswriter"
  25.       passivate-idle-entity-after = 5 m
  26.     }
  27.   }
  29.   persistence {
  30.     journal.plugin = "cassandra-journal"
  31.     snapshot-store.plugin = "cassandra-snapshot-store"
  32.   }
  34. }
  36. cassandra-journal {
  37.   contact-points = ["192.168.11.162"]
  38. }
  40. cassandra-snapshot-store {
  41.   contact-points = ["192.168.11.162"]
  42. }

POSRouter.scala

  1. package cloud.pos.server
  3. import akka.actor._
  4. import akka.cluster.sharding._
  5. import akka.cluster.sharding.ClusterSharding
  6. import com.typesafe.config.ConfigFactory
  7. import sdp.cluster.monitor._
  8. import sdp.logging._
  10. object POSRouter extends LogSupport {
  11.   def main(args: Array[String]) {
  12.     import WriterActor._
  13.     import Commands._
  15.     val argsPat = "(.*):(.*)".r
  16.     val (host, port) = args(0) match {
  17.       case argsPat(h, p) => (h, p)
  18.       case _ => ("localhost", "2551")
  19.     }
  22.     val config = ConfigFactory.parseString("akka.remote.netty.tcp.port=\"" + port + "\"")
  23.       .withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.hostname=\"" + host + "\""))
  24.       //roles can be deployed on this node
  25.       .withFallback(ConfigFactory.parseString("akka.cluster.roles = [poswriter]"))
  26.       .withFallback(ConfigFactory.load())
  29.     log.info(s"******* hostname = $host,  port = $port *******")
  31.     val shardName = "POSShard"
  32.  
  33.     case class POSMessage(id: Long, cmd: POSCommand) {
  34.       def shopId = id.toString.head.toString
  36.       def posId = id.toString
  37.     }
  39.     val getPOSId: ShardRegion.ExtractEntityId = {
  40.       case posCommand: POSMessage => (posCommand.posId, posCommand.cmd)
  41.     }
  42.     val getShopId: ShardRegion.ExtractShardId = {
  43.       case posCommand: POSMessage => posCommand.shopId
  44.     }
  47.     val system = ActorSystem("cloud-pos-server", config)
  48.     val role = "poswriter"    //role of this shard
  49.     val cpsSettings = ClusterShardingSettings(system).withRole(role) //.withPassivateIdleAfter(10 minutes)
  51.     ClusterSharding(system).start(
  52.       typeName = shardName,
  53.       entityProps = writerProps,
  54.       settings = cpsSettings,
  55.       extractEntityId = getPOSId,
  56.       extractShardId = getShopId,
  57.       allocationStrategy = ClusterSharding(system).defaultShardAllocationStrategy(cpsSettings),
  58.       handOffStopMessage = PassivatePOS
  59.     )
  61.     system.actorOf(ClusterMonitor.props, "cps-cluster-monitor")
  63.   }
  64. }

下面是这个测试项目的源代码：

build.sbt

  1. name := "cloud-pos-client"
  3. version := "0.1"
  5. scalaVersion := "2.12.8"
  7. libraryDependencies := Seq(
  8.   "com.typesafe.akka" %% "akka-cluster-sharding" % "2.5.19",
  9.   "com.typesafe.akka" %% "akka-persistence" % "2.5.19",
  10.   "com.typesafe.akka" %% "akka-persistence-cassandra" % "0.93",
  11.   "com.typesafe.akka" %% "akka-persistence-cassandra-launcher" % "0.93" % Test,
  12.   "ch.qos.logback"  %  "logback-classic"   % "1.2.3"
  13. )

resources/application.conf

  1. akka.actor.warn-about-java-serializer-usage = off
  2. akka.log-dead-letters-during-shutdown = off
  3. akka.log-dead-letters = off
  5. akka {
  6.   loglevel = INFO
  7.   actor {
  8.     provider = "cluster"
  9.   }
  11.   remote {
  12.     log-remote-lifecycle-events = off
  13.     netty.tcp {
  14.       hostname = "192.168.11.162"
  15.       port = 2558
  16.     }
  17.   }
  19.   cluster {
  20.     seed-nodes = [
  21.       "akka.tcp://cloud-pos-server@192.168.11.162:2551"]
  22.     log-info = off
  23.   }
  25. }

resources/logback.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <configuration>
  3.     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
  4.         <encoder>
  5.             <Pattern>
  6.                 %d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n
  7.             </Pattern>
  8.         </encoder>
  9.     </appender>
  10.  
  11.     <root level="debug">
  12.         <appender-ref ref="STDOUT" />
  13.     </root>
  14. </configuration>

ClientDemo.scala

  1. package cloud.pos.client
  2. import akka.actor._
  3. import akka.cluster.sharding.ClusterSharding
  4. import sdp.cluster.monitor._
  5. import sdp.logging._
  6. import Commands._
  7. import States._
  8. import Items._
  9. import akka.cluster.sharding._
  12. object POSClientDemo extends LogSupport {
  13.   def main(args: Array[String]) {
  15.     val system = ActorSystem("cloud-pos-server")
  17.     val shardName = "POSShard"
  19.     val getPOSId: ShardRegion.ExtractEntityId = {
  20.       case posCommand: POSMessage => (posCommand.posId, posCommand.cmd)
  21.     }
  22.     val getShopId: ShardRegion.ExtractShardId = {
  23.       case posCommand: POSMessage => posCommand.shopId
  24.     }
  26.     //no shard deployed on this node  2558, use proxy
  27.     val posHandler = ClusterSharding(system).startProxy(
  28.       typeName = shardName,
  29.       role = Some("poswriter"),
  30.       extractEntityId = getPOSId,
  31.       extractShardId = getShopId
  32.     )
  34.     //val posHandler = ClusterSharding(system).shardRegion(shardName)
  36.     system.actorOf(ClusterMonitor.props, "cps-cluster-monitor")
  38.     val posref = system.actorOf(POSClient.props(posHandler), "pos-client")
  40.     posref ! POSMessage(1021, LogSales(SALESTYPE.plu, "", apple.code, 1, 0))
  41.     posref ! POSMessage(2021, LogSales(SALESTYPE.plu, "", pineapple.code, 2, 0))
  42.     posref ! POSMessage(3021, LogSales(SALESTYPE.plu, "", banana.code, 1, 0))
  43.     posref ! POSMessage(4021, LogSales(SALESTYPE.plu, "", grape.code, 3, 0))
  44.     posref ! POSMessage(4021,Subtotal)
  48.     scala.io.StdIn.readLine()
  50.     system.terminate()
  52.   }
  53. }

client/Commands.scala

  1. package cloud.pos.client
  3. object Commands {
  5.   sealed trait POSCommand {}
  7.   case class LogOn(opr: String, passwd: String) extends POSCommand
  8.   case object LogOff extends POSCommand
  9.   case class SuperOn(su: String, passwd: String) extends POSCommand
  10.   case object SuperOff extends POSCommand
  11.   case class MemberOn(cardnum: String, passwd: String) extends POSCommand
  12.   case object MemberOff extends POSCommand   //remove member status for the voucher
  13.   case object RefundOn extends POSCommand
  14.   case object RefundOff extends POSCommand
  15.   case object VoidOn extends POSCommand
  16.   case object VoidOff extends POSCommand
  17.   case object VoidAll extends POSCommand
  18.   case object Suspend extends POSCommand
  20.   case class VoucherNum(vnum: Int) extends POSCommand
  23.   case class LogSales(salesType: Int, dpt: String, code: String, qty: Int, price: Int) extends POSCommand
  24.   case object Subtotal extends POSCommand
  25.   case class Discount(code: String, percent: Int) extends POSCommand
  27.   case class OfflinePay(acct: String, num: String, amount: Int) extends POSCommand          //settlement   结算支付
  28.   //read only command, no event process
  29.   case class VCBalance(acct: String, num: String, passwd: String) extends POSCommand
  30.   case class VCPay(acct: String, num: String, passwd: String, amount: Int) extends POSCommand
  31.   case class AliPay(acct: String, num: String, amount: Int) extends POSCommand
  32.   case class WxPay(acct: String, num: String, amount: Int) extends POSCommand
  35.   // read only command, no update event
  36.   case class Plu(itemCode: String) extends POSCommand  //read only
  37.  
  38.   case class POSMessage(id: Long, cmd: POSCommand) {
  39.     def shopId = id.toString.head.toString
  40.     def posId = id.toString
  41.   }
  44. }

client/States.scala

  1. package cloud.pos.client
  3. object States {
  5.   object TXNTYPE {
  6.     val sales: Int = 0
  7.     val refund: Int = 1
  8.     val void: Int = 2
  9.     val voided: Int = 3
  10.     val voidall: Int = 4
  11.     val subtotal: Int = 5
  12.     val logon: Int = 6
  13.     val supon: Int = 7       // super user on/off
  14.     val suspend: Int = 8
  16.   }
  18.   object SALESTYPE {
  19.     val plu: Int = 0
  20.     val dpt: Int = 1
  21.     val cat: Int = 2
  22.     val brd: Int = 3
  23.     val ra:  Int = 4
  24.     val sub: Int = 5
  25.     val ttl: Int = 6
  26.     val dsc: Int = 7
  27.     val crd: Int = 8
  28.   }
  31.   case class TxnItem(
  32.                       txndate: String = ""
  33.                       ,txntime: String = ""
  34.                       ,opr: String = ""//工号
  35.                       ,num: Int = 0 //销售单号
  36.                       ,seq: Int = 1 //交易序号
  37.                       ,txntype: Int = TXNTYPE.sales//交易类型
  38.                       ,salestype: Int = SALESTYPE.plu //销售类型
  39.                       ,qty: Int =  1 //交易数量
  40.                       ,price: Int = 0 //单价（分）
  41.                       ,amount: Int = 0 //码洋（分）
  42.                       ,dscamt: Int = 0 //折扣：负值  net实洋 = amount + dscamt
  43.                       ,member: String = "" //会员卡号
  44.                       ,code: String = "" //编号（商品、账号...）
  45.                       ,desc: String = "" //项目名称
  46.                       ,dpt: String = ""
  47.                       ,department: String = ""
  48.                       ,cat: String = ""
  49.                       ,category: String = ""
  50.                       ,brd: String = ""
  51.                       ,brand: String = ""
  52.                     )
  55.   case class VchStatus( //操作状态锁留给前端维护
  56.                         qty: Int = 1,
  57.                         refund: Boolean = false,
  58.                         void: Boolean = false)
  60.   case class VchStates(
  61.                         opr: String = "",      //收款员
  62.                         jseq: BigInt = 0,      //begin journal sequence for read-side replay
  63.                         num: Int = 0,          //当前单号
  64.                         seq: Int = 1,          //当前序号
  65.                         void: Boolean = false, //取消模式
  66.                         refd: Boolean = false, //退款模式
  67.                         due: Boolean = true,   //当前余额
  68.                         su: String = "",
  69.                         mbr: String = ""
  70.                       )
  73. }

client/POSClient.scala

  1. package cloud.pos.client
  3. import akka.actor._
  4. import sdp.logging._
  5. import Responses._
  6. import Commands._
  8. object POSClient {
  9.   def props(pos: ActorRef) = Props(new POSClient(pos))
  10. }
  11. class POSClient(posHandler: ActorRef)  extends Actor with LogSupport {
  13.   override def receive: Receive = {
  14.     case msg @ POSMessage(_,_) => posHandler ! msg
  15.     case resp: POSResponse  =>
  16.       log.info(s"response from server: $resp")
  17.   }
  18. }

client/Responses.scala

  1. package cloud.pos.client
  3. import States._
  4. object Responses {
  6.   object STATUS {
  7.     val OK: Int = 0
  8.     val FAIL: Int = -1
  9.   }
  11.   case class POSResponse (sts: Int, msg: String, voucher: VchStates, txnItems: List[TxnItem])
  12. }

client/DataAccess.scala

  1. package cloud.pos.client
  3. import java.time.LocalDate
  4. import java.time.format.DateTimeFormatter
  7. case class Item(
  8.                  brd: String
  9.                  ,dpt: String
  10.                  ,cat: String
  11.                  ,code: String
  12.                  ,name: String
  13.                  ,price: Int
  15.                )
  16. object Items {
  17.   val apple = Item("01","02","01","001", "green apple", 820)
  18.   val grape = Item("01","02","01","002", "red grape", 1050)
  19.   val orage = Item("01","02","01","003", "sunkist orage", 350)
  20.   val banana = Item("01","02","01","004", "demon banana", 300)
  21.   val pineapple = Item("01","02","01","005", "hainan pineapple", 1300)
  22.   val peach = Item("01","02","01","006", "xinjiang peach", 2390)
  24.   val tblItems = List(apple, grape, orage, banana, pineapple, peach)
  26.   sealed trait QueryItemsResult {}
  28.   case class QueryItemsOK(items: List[Item]) extends QueryItemsResult
  30.   case class QueryItemsFail(msg: String) extends QueryItemsResult
  32. }
  35. object Codes {
  36.   case class User(code: String, name: String, passwd: String)
  37.   case class Department(code: String, name: String)
  38.   case class Category(code: String, name: String)
  39.   case class Brand(code: String, name: String)
  40.   case class Ra(code: String, name: String)
  41.   case class Account(code: String, name: String)
  42.   case class Disc(code: String, best: Boolean, aggr: Boolean, group: Boolean)
  44.   val ras = List(Ra("01","Delivery"),Ra("02","Cooking"))
  45.   val dpts = List(Department("01","Fruit"),Department("02","Grocery"))
  46.   val cats = List(Category("0101","Fresh Fruit"),Category("0201","Dry Grocery"))
  47.   val brds = List(Brand("01","Sunkist"),Brand("02","Demon"))
  48.   val accts = List(Account("001","Cash"),Account("002","Value Card"), Account("003", "Visa")
  49.     ,Account("004","Alipay"),Account("005","WXPay"))
  51.   val users = List(User("1001","Tiger", "123"),User("1002","John", "123"),User("1003","Maria", "123"))
  53.   def getDpt(code: String) = dpts.find(d => d.code == code)
  54.   def getCat(code: String) = cats.find(d => d.code == code)
  55.   def getBrd(code: String) = brds.find(b => b.code == code)
  56.   def getAcct(code: String) = accts.find(a => a.code == code)
  57.   def getRa(code: String) = ras.find(a => a.code == code)
  58. }
  60. object DAO {
  61.   import Items._
  62.   import Codes._
  64.   def getItem(code: String): QueryItemsResult = {
  65.     val optItem = tblItems.find(it => it.code == code)
  66.     optItem match {
  67.       case Some(item) => QueryItemsOK(List(item))
  68.       case None => QueryItemsFail("Invalid item code!")
  69.     }
  70.   }
  72.   def validateDpt(code: String) = dpts.find(d => d.code == code)
  73.   def validateCat(code: String) = cats.find(d => d.code == code)
  74.   def validateBrd(code: String) = brds.find(b => b.code == code)
  75.   def validateRa(code: String) = ras.find(ac => ac.code == code)
  76.   def validateAcct(code: String) = accts.find(ac => ac.code == code)
  78.   def validateUser(userid: String, passwd: String) = users.find(u => (u.code == userid && u.passwd == passwd))
  80.   def lastSecOfDateStr(ldate: LocalDate): String = {
  81.     ldate.format(DateTimeFormatter.ofPattern( "yyyy-MM-dd"))+" 23:59:59"
  82.   }
  85. }

logging/Log.scala

  1. package sdp.logging
  3. import org.slf4j.Logger
  5. /**
  6.   * Logger which just wraps org.slf4j.Logger internally.
  7.   *
  8.   * @param logger logger
  9.   */
  10. class Log(logger: Logger) {
  12.   // use var consciously to enable squeezing later
  13.   var isDebugEnabled: Boolean = logger.isDebugEnabled
  14.   var isInfoEnabled: Boolean = logger.isInfoEnabled
  15.   var isWarnEnabled: Boolean = logger.isWarnEnabled
  16.   var isErrorEnabled: Boolean = logger.isErrorEnabled
  18.   def withLevel(level: Symbol)(msg: => String, e: Throwable = null): Unit = {
  19.     level match {
  20.       case 'debug | 'DEBUG => debug(msg)
  21.       case 'info | 'INFO => info(msg)
  22.       case 'warn | 'WARN => warn(msg)
  23.       case 'error | 'ERROR => error(msg)
  24.       case _ => // nothing to do
  25.     }
  26.   }
  28.   def debug(msg: => String): Unit = {
  29.     if (isDebugEnabled && logger.isDebugEnabled) {
  30.       logger.debug(msg)
  31.     }
  32.   }
  34.   def debug(msg: => String, e: Throwable): Unit = {
  35.     if (isDebugEnabled && logger.isDebugEnabled) {
  36.       logger.debug(msg, e)
  37.     }
  38.   }
  40.   def info(msg: => String): Unit = {
  41.     if (isInfoEnabled && logger.isInfoEnabled) {
  42.       logger.info(msg)
  43.     }
  44.   }
  46.   def info(msg: => String, e: Throwable): Unit = {
  47.     if (isInfoEnabled && logger.isInfoEnabled) {
  48.       logger.info(msg, e)
  49.     }
  50.   }
  52.   def warn(msg: => String): Unit = {
  53.     if (isWarnEnabled && logger.isWarnEnabled) {
  54.       logger.warn(msg)
  55.     }
  56.   }
  58.   def warn(msg: => String, e: Throwable): Unit = {
  59.     if (isWarnEnabled && logger.isWarnEnabled) {
  60.       logger.warn(msg, e)
  61.     }
  62.   }
  64.   def error(msg: => String): Unit = {
  65.     if (isErrorEnabled && logger.isErrorEnabled) {
  66.       logger.error(msg)
  67.     }
  68.   }
  70.   def error(msg: => String, e: Throwable): Unit = {
  71.     if (isErrorEnabled && logger.isErrorEnabled) {
  72.       logger.error(msg, e)
  73.     }
  74.   }
  76. }

logging/LogSupport.scala

  1. package sdp.logging
  3. import org.slf4j.LoggerFactory
  5. trait LogSupport {
  7.   /**
  8.     * Logger
  9.     */
  10.   protected val log = new Log(LoggerFactory.getLogger(this.getClass))
  12. }

logging/ClusterMonitor.scala

  1. package sdp.cluster.monitor
  3. import akka.actor._
  4. import akka.cluster.ClusterEvent._
  5. import akka.cluster._
  6. import sdp.logging.LogSupport
  8. object ClusterMonitor {
  9.   def props = Props(new ClusterMonitor())
  10. }
  12. class ClusterMonitor extends Actor with LogSupport {
  13.   val cluster = Cluster(context.system)
  14.   override def preStart(): Unit = {
  15.     cluster.subscribe(self,initialStateMode = InitialStateAsEvents
  16.       ,classOf[MemberEvent],classOf[UnreachableMember])  //订阅集群状态转换信息
  17.     super.preStart()
  18.   }
  20.   override def postStop(): Unit = {
  21.     cluster.unsubscribe(self)    //取消订阅
  22.     super.postStop()
  23.   }
  25.   override def receive: Receive = {
  26.     case MemberJoined(member) =>
  27.       log.info(s"Member is Joining: {${member.address}}")
  28.     case MemberUp(member) =>
  29.       log.info(s"Member is Up: {${member.address}}")
  30.     case MemberLeft(member) =>
  31.       log.info(s"Member is Leaving: {${member.address}}")
  32.     case MemberExited(member) =>
  33.       log.info(s"Member is Exiting: {${member.address}}")
  34.     case MemberRemoved(member, previousStatus) =>
  35.       log.info(
  36.         s"Member is Removed: {${member.address}} after {${previousStatus}")
  37.     case UnreachableMember(member) =>
  38.       log.info(s"Member detected as unreachable: {${member.address}}")
  39.       cluster.down(member.address)      //手工驱除，不用auto-down
  40.     case _: MemberEvent => // ignore
  41.   }
  43. }

 

原文链接：http://www.cnblogs.com/tiger-xc/p/10662854.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Scala热门文章

  • scala (3) Function 和 Method
  • flink 两个datastream实现left_join的两种方法
  • scala case class + BeanProperty+fastjson 快速实现类的序列化与反序列化
  • Scala解析MDF文件
  • flink 如何实现对watermark 的checkpoint,防止数据复写
  • Spark笔记(一):错误总结
  • 踩坑 scala 打包上传 找不到类
  • akka-typed(2) - typed-actor交流方式和交流协议
  • Scala(二)：元组、数组、映射
  • scala 命名规范

    • Scala推荐文章

  • Scala中Array和List的区别说明
  • alpakka-kafka(5)-kafka集群配置与分布式应用部署
  • alpakka-kafka(2)-consumer
  • alpakka-kafka(1)-producer
  • akka-typed(1) - actor生命周期管理
  • akka-typed(2) - typed-actor交流方式和交流协议
  • akka-typed(3) - PersistentActor has EventSourcedBehavior
  • akka-typed(4) - EventSourcedBehavior in action
  • kka-typed(5) - cluster：集群节点状态监视
  • akka-typed(6) - cluster：group router, cluster-load-balancing

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号