使用akka框架和scala语言编写简单的RPC通信案例

前言

1）akka框架是一个并发的、分布式的、可伸缩性的、高性能的RPC通信框架，大数据开发框架Spark、flink底层原理中或多或少都用到了

2）scala语言真的很强大、好用、方便，结合了面向对象语言和函数式语言的特点
akka的原理图

大多数分布式框架或工具 都遵循着主从节点的架构设计，在这里我们暂不考虑高可用的模式（高可用可参考文章Zookeeper之HDFS-HA高可用模式）
每个机器上的一个进程中只存在着1个通信角色对象  ActorSystem  ，也就是说 ActorSystem 对象的示例只有一个，但由它创建的Master和Worker可以有多个，是多例
1）启动master   内部定时器定期检查有无超时连接（就是在一定时间内没有向我发送心跳的worker），并将失效的进行移除
2）启动worker，跟master建立网络连接，将自己的信息（workerid，内存，内核数cpu等信息）发给master进行注册
3）master收到注册信息，将注册的信息进行保存到内存（高效），也可以持久化到磁盘或zookeeper当中（数据安全），之后向worker发送注册成功的信息
4）worker收到master发来的注册成功的信息，很高兴，并启动定时器，定期发送心跳，向master报活

代码实现

Worker类代码：

1. import java.util.UUID
2. import java.util.concurrent.TimeUnit
4. import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
5. import com.typesafe.config.ConfigFactory
7. import scala.concurrent.duration._
9. /**
10. * @author:tom
11. * @Date:Created in 16:49 2020/12/18
12. */
13. class Worker extends Actor {
16.   var masterRef: ActorSelection = _
18.   var workerId = UUID.randomUUID().toString
20.   //在执行构造函数（实例化对象）之后、receive方法执行之前一定会执行一次
21.   override def preStart(): Unit = {
23.     //向master 进行注册信息
24.     //可以与master建立连接
25.     masterRef = context.actorSelection("akka.tcp://MasterActorSystem@localhost:8888/user/MasterActor")
26.     //发送消息
27.     masterRef ! RegisterWorker(workerId, "2048", 4)
28.   }
30.   override def receive: Receive = {
31.     //自己给自己发送的周期消息
32.     case SendHeartbeat => {
33.       //      if () {
34.       //
35.       //      } 向Master发送心跳
36.       masterRef ! HeartBeat(workerId)
37.     }
39.     case RegisteredWorker => {
40.       //      println("a response from master")
42.       //启动一个定时器
43.       import context.dispatcher
44.       context.system.scheduler.schedule(Duration(0, TimeUnit.MILLISECONDS), 10000.millisecond, self, SendHeartbeat)
46.     }
47.   }
48. }
50. object Worker {
52.   def main(args: Array[String]): Unit = {
53.     val host = "localhost"
54.     val port = 9999
55.     val configStr =
56.       s"""
57.          |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
58.          |akka.remote.netty.tcp.hostname = $host
59.          |akka.remote.netty.tcp.port = $port
60.          |""".stripMargin
61.     val config = ConfigFactory.parseString(configStr)
62.     //创建workerActorSystem
63.     val workerActorSystem = ActorSystem.apply("workerActorSystem", config)
64.     //创建workerActor
65.     val workerActor = workerActorSystem.actorOf(Props(new Worker), "WorkerActor")
66.   }
69. }

复制代码

Master代码：

1. import akka.actor.{Actor, ActorSystem, Props}
2. import com.typesafe.config.ConfigFactory
4. import scala.collection.mutable
5. import scala.concurrent.duration._
7. /**
8. * @author:tom
9. * @Date:Created in 16:08 2020/12/18
10. */
11. class Master extends Actor {
13.   //定义一个可变的HashMap集合用来存储worker的信息
14.   val idToWorker = new mutable.HashMap[String, WorkerInfo]()
17.   //master定期检查自己  是否有新的节点（worker出现）
18.   override def preStart(): Unit = {
19.     import context.dispatcher
20.     context.system.scheduler.schedule(0 millisecond, 15000.millisecond, self, CheckTimeOutWorker)
21.   }
23.   //用来接收消息
24.   override def receive: Receive = {
26.     //模式匹配
27.     case "hello" => {
28.       println("hello~")
29.     }
30.     case "hi" => {
31.       println("hi~")
32.     }
34.     //定时检查
35.     case CheckTimeOutWorker => {
36.       val deadWorkers = idToWorker.values.filter(w => System.currentTimeMillis() - w.lastHeartbeatTime > 30000)
37.       deadWorkers.foreach(dw => {
38.         idToWorker -= dw.workerId
39.       })
40.       println(s"current alive worker size:${idToWorker.size}")
41.     }
43.     //有worker来进行注册信息需要执行的逻辑
44.     case RegisterWorker(workerId, memory, cores) => {
45.       //      println(s"workerId:$workerId,memory:$memory,cores:$cores")
47.       //worker 注册成功应该执行的逻辑
49.       //将信息存入到内存集合当中
50.       val workerInfo: WorkerInfo = new WorkerInfo(workerId, memory, cores)
51.       idToWorker.put(workerId, workerInfo)
52.       //返回一个注册成功的信息
53.       sender() ! RegisteredWorker
54.     }
56.     //worker端发送过来的心跳信息
57.     case HeartBeat(workerId) => {
58.       //根据workerId到Map中查找对应的WorkerInfo
59.       if (idToWorker.contains(workerId)) {
60.         //如果存在 则取出信息
61.         val workerInfo = idToWorker(workerId)
62.         //更新上一次的心跳时间
63.         workerInfo.lastHeartbeatTime = System.currentTimeMillis()
64.       }
65.     }
67.   }
68. }
70. object Master {
72.   def main(args: Array[String]): Unit = {
73.     val host = "localhost"
74.     val port = 8888
75.     val configStr =
76.       s"""
77.          |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
78.          |akka.remote.netty.tcp.hostname = $host
79.          |akka.remote.netty.tcp.port = $port
80.          |""".stripMargin
81.     val config = ConfigFactory.parseString(configStr)
82.     //创建一个ActorSystem实例（单例）
83.     val masterActorSystem = ActorSystem("MasterActorSystem", config)
84.     //创建一个Actor
85.     val actor = masterActorSystem.actorOf(Props[Master], "MasterActor")
86.     //自己给自己发消息
87.     actor ! "hello"
88.   }
91. }

复制代码

更多学习、面试资料尽在微信公众号：Hadoop大数据开发

已经typed了，否则消息类别多了不好调试