scala-akka实现简单的分布式RPC通信框架（主从监听，消息发送）

我会带着你远行 2022-06-02 06:27 194阅读 0赞

简介：模拟用akka实现一个RPC分布式通信框架，实现从机向主机发送心跳，主机判断从机是否挂机，以及消息通信的简单功能。

开发平台：IntelliJ IDEA     Maven架构

项目结构及依赖：

![Center][]

代码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, ActorSystem, 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,5000 millis,self,"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,辅助创建和监控下面的Actor,单例
          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挂掉了。

[Center]: /images/20220602/485b9e0d19c3485ea8a6789a9e33e65d.png