简介：模拟用akka实现一个RPC分布式通信框架，实现从机向主机发送心跳，主机判断从机是否挂机，以及消息通信的简单功能。
开发平台：IntelliJ IDEA Maven架构
项目结构及依赖：
代码2个样例类：
case class Workinfo (val id:String) extends Serializable{
}
case class SendHearBeat (val id:String) extends Serializable{
}
Master类：
import akka.actor.{Actor,ActorSystem,Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
import scala.collection.mutable
class Master extends Actor{
//保存worker id
val workerMessage=new mutable.HashMap[String,Long]
//worker超时时间
val WORKEER_TIMEOUT=10 * 1000
override def preStart(): Unit ={
//导入隐式转换，用于启动定时器
import context.dispatcher
//启动定时
context.system.scheduler.schedule(0 millis,5000 millis,self,"CheckOfTimeOutWorker")
}
override def receive: Receive = {
//注册woker把workerid 和当前时间存起来
case Workinfo(id)=>{
if(!workerMessage.contains(id)) {
workerMessage.put(id,System.currentTimeMillis())
}
println("注册worker:"+id)
sender ! "MasterReply"
}
//接收worker心跳，更新接收时间
case SendHearBeat(id)=>{
workerMessage(id)=System.currentTimeMillis()
println("接收到worker:"+id+"的心跳报告。")
}
//Master自己向自己发送的定期检查超时Worker的消息
case "CheckOfTimeOutWorker" =>{
val currentTime = System.currentTimeMillis()
//过滤出超时的worker
val outTimeWorker=workerMessage.filter(m=>currentTime-m._2>WORKEER_TIMEOUT).toArray
for(worker<-outTimeWorker){
workerMessage.remove(worker._1)
}
println("woker数量："+outTimeWorker.size+"个，挂了")
}
}
}
object Master{
def main(args: Array[String]): Unit = {
//接受参数
val host="169.254.189.184"
val port="8888"
//配置信息
val configStr=
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider"
|akka.remote.netty.tcp.hostname = "$host"
|akka.remote.netty.tcp.port = "$port"
""".stripMargin
val config=ConfigFactory.parseString(configStr)
//ActorSystem,辅助创建和监控下面的Actor,单例
val actorSystem=ActorSystem("MasterSystem",config)
//执行mastor生命周期方法
val master= actorSystem.actorOf(Props(new Master),"Master")
actorSystem.awaitTermination()
}
}
Worker类：
import java.util.UUID
import akka.actor.{Actor,ActorSelection,Props}
import scala.concurrent.duration._
import com.typesafe.config.ConfigFactory
class Worker(val masterHost:String,val masterPort:String) extends Actor{
var master:ActorSelection=_
//生成uuid作为worker标识
val id=UUID.randomUUID().toString
//建立连接，链接Master
override def preStart(): Unit ={
//Master就是你前面给master起的名字
master=context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master")
//id和其他信息封装到一个类中，把这个类发送过去，注意这个类要序列化
master ! Workinfo(id)
}
override def receive: Receive = {
//接收master返回来的信息后启动定时
case "MasterReply"=>{
println("a reply from master")
//导入隐式转换，用于启动定时器
import context.dispatcher
//启动定时任务，通过case回调SendHearBeat向master发送心跳(id标识)
context.system.scheduler.schedule(0 millis,"SendHeartBeat")
}
case "SendHearBeat"=>{
println("worker send heartbeat")
master ! SendHearBeat(id)
}
}
}
object Worker{
def main(args: Array[String]): Unit = {
//接受参数
val workerHost="169.254.189.184"
val workerPort="9999"
val masterHost="169.254.189.184"
val masterPort="8888"
//配置信息
val configStr=
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider"
|akka.remote.netty.tcp.hostname = "$workerHost"
|akka.remote.netty.tcp.port = "$workerPort"
""".stripMargin
val config=ConfigFactory.parseString(configStr)
//ActorSystem,单例
val actorSystem=ActorSystem("WorkerSystem",config)
//启动Actor，woker会被实例化，生命周期方法会被调用
actorSystem.actorOf(Props(new Worker(masterHost,masterPort)),"Worker")
}
}
启动Master控制台打印：
"C:\Program Files (x86)\Java\jdk1.7.0_01\bin\java" "-javaagent:D:\Program Files (x86)\IntelliJ IDEA 2017.3.2\lib\idea_rt.jar=54133:D:\Program Files (x86)\IntelliJ IDEA 2017.3.2\bin" -Dfile.encoding=UTF-8 -classpath "C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\charsets.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\deploy.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\ext\dnsns.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\ext\localedata.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\ext\sunec.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\ext\sunjce_provider.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\ext\sunmscapi.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\ext\sunpkcs11.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\ext\zipfs.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\javaws.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\jce.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\jsse.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\management-agent.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\plugin.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\resources.jar;C:\Program Files (x86)\Java\jdk1.7.0_01\jre\lib\rt.jar;D:\Program Files (x86)\IntelliJ IDEA 2017.3.2\scalaspace\rpcScala\target\classes;C:\Users\Administrator\.m2\repository\org\scala-lang\scala-library\2.10.6\scala-library-2.10.6.jar;C:\Users\Administrator\.m2\repository\com\typesafe\akka\akka-actor_2.10\2.3.14\akka-actor_2.10-2.3.14.jar;C:\Users\Administrator\.m2\repository\com\typesafe\config\1.2.1\config-1.2.1.jar;C:\Users\Administrator\.m2\repository\com\typesafe\akka\akka-remote_2.10\2.3.14\akka-remote_2.10-2.3.14.jar;C:\Users\Administrator\.m2\repository\io\netty\netty\3.8.0.Final\netty-3.8.0.Final.jar;C:\Users\Administrator\.m2\repository\com\google\protobuf\protobuf-java\2.5.0\protobuf-java-2.5.0.jar;C:\Users\Administrator\.m2\repository\org\uncommons\maths\uncommons-maths\1.2.2a\uncommons-maths-1.2.2a.jar" Master [INFO] [01/06/2018 13:44:54.602] [main] [Remoting] Starting remoting [INFO] [01/06/2018 13:44:55.338] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://MasterSystem@169.254.189.184:8888] [INFO] [01/06/2018 13:44:55.341] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://MasterSystem@169.254.189.184:8888] woker数量：0个，挂了 woker数量：0个，挂了 woker数量：0个，挂了 woker数量：0个，挂了
启动Woker后Master控制台打印：（看见worker注册了）
woker数量：0个，挂了 woker数量：0个，挂了 注册worker:9e8ec9e7-6014-491f-816e-94d0f65eb068 woker数量：0个，挂了 woker数量：0个，挂了worker控制台打印：
a reply from master
关掉Worker后Master控制台打印：
woker数量：0个，挂了 woker数量：0个，挂了 woker数量：1个，挂了 woker数量：0个，挂了可以看到关掉worker后Master监控到有一台worker挂掉了。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。