Akka之简单的自定义RPC框架（乞丐版）

关于Akka

Akka 是一个用 Scala 编写的库，用于简化编写容错的、高可伸缩性的 Java 和 Scala 的 Actor 模型应用。它已经成功运用在电信行业。系统几乎不会宕机（高可用性 99.9999999 % 一年只有 31 ms 宕机）。

自定义RPC通信框架（乞丐版）

Akka之简单的自定义RPC框架（乞丐版）

目标

woker能发送成功注册，并定时发送心跳。

master能成功接收注册，并能接收心跳及完成自检。

大体思路

1、提供一个Master，负责woker的任务分配，注册及销毁。

2、提供一个Woker，负责Master分配的任务。需要定时向Master报告状态

3、Master内部提供自检机制，为其检测过期woker并销毁。

大体思路就是这样。下面开始撸代码：

1，消息模板类

  1. package com.itunic.akka
  3. /**
  4.   * Created by itunic.com on 2016/12/12.
  5.   */
  6. trait RemoteMessage extends Serializable
  8. //注册消息
  9. //worker -> master
  10. case class Register(workerId: String, memorys: Int, cores: Int) extends RemoteMessage
  12. //返回注册成功信息 master的连接信息
  13. //master -> worker
  14. case class Registered(ip: String, port: Int) extends RemoteMessage
  16. //报活
  17. //worker -> master
  18. case class TransferHeartbeat(workerId: String) extends RemoteMessage
  20. //heartbeat ->worker
  21. case object SendHeartbeat
  23. //heartbeat->master
  24. case object CheckTimeOutWorker
  26. //master ->worker 重新注册
  27. case object RegisterAgain

2，Woker实体类

  1. package com.itunic.akka
  3. /**
  4.   * Created by itunic.com on 2016/12/13.
  5.   */
  6. class WorkerInfo(workerId: String, memory: Int, cores: Int) {
  7.   //TODO 最后更新日期
  8.   var lastTime: Long = _
  9. }

3，核心类：Master

  1. package com.itunic.akka
  3. import akka.actor.{Actor, ActorSystem, Props}
  4. import com.typesafe.config.ConfigFactory
  5. import scala.concurrent.duration._
  6. import scala.collection.mutable
  8. /**
  9.   * Created by root
  10.   */
  11. class Master(host: String, prot: Int) extends Actor {
  13.   println("constructor invoked")
  14.   var workerMap = new mutable.HashMap[String, WorkerInfo]()
  15.   //检测时间
  16.   val CHECK_TIME = 15000
  18.   override def preStart(): Unit = {
  19.     println("preStart invoked")
  20.     //启动线程
  21.     import context.dispatcher
  22.     context.system.scheduler.schedule(0 millis, CHECK_TIME millis, self, CheckTimeOutWorker)
  23.   }
  25.   // 用于接收消息
  26.   override def receive: Receive = {
  27.     case Register(workerId, memorys, cores) => {
  28.       println("a client connected")
  29.       //判断map是否包含此worker
  30.       if (!workerMap.contains(workerId)) {
  31.         val workerInfo = new WorkerInfo(workerId, memorys, cores)
  32.         workerMap += (workerId -> workerInfo)
  33.       }
  34.       sender ! Registered(host, prot)
  35.     }
  36.     case TransferHeartbeat(workerId) => {
  37.       if (workerMap.contains(workerId)) {
  38.         val info = workerMap(workerId)
  39.         info.lastTime = System.currentTimeMillis()
  40.       } else {
  41.         sender ! RegisterAgain
  42.       }
  43.     }
  44.     //检测
  45.     case CheckTimeOutWorker => {
  46.       val thisTime = System.currentTimeMillis()
  47.       val removeWorker = workerMap.filter(x => thisTime - x._2.lastTime > CHECK_TIME)
  48.       for (i <- removeWorker) {
  49.         workerMap -= i._1
  50.       }
  51.       println(workerMap.size)
  52.     }
  53.     case "hello" => {
  54.       println("hello")
  55.     }
  56.   }
  57. }
  59. object Master {
  60.   def main(args: Array[String]) {
  62.     val host = args(0)
  63.     val port = args(1).toInt
  64.     // 准备配置
  65.     val configStr =
  66.     s"""
  67.        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
  68.        |akka.remote.netty.tcp.hostname = "$host"
  69.        |akka.remote.netty.tcp.port = "$port"
  70.        """.stripMargin
  71.     val config = ConfigFactory.parseString(configStr)
  72.     //ActorSystem老大，辅助创建和监控下面的Actor，他是单例的
  73.     val actorSystem = ActorSystem("MasterSystem", config)
  74.     //创建Actor, 起个名字
  75.     val master = actorSystem.actorOf(Props(new Master(host, port)), "Master") //Master主构造器会执行
  76.     master ! "hello" //发送信息
  77.     actorSystem.awaitTermination() //让进程等待着, 先别结束
  79.   }
  80. }

4，核心类：Woker

  1. package com.itunic.akka
  3. import java.util.UUID
  4. import scala.concurrent.duration._
  6. import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
  7. import com.typesafe.config.ConfigFactory
  9. import scala.language.postfixOps
  11. /**
  12.   * Created by itunic.com on 2016/12/13.
  13.   */
  14. class Worker(val masterHost: String, val masterPort: Int) extends Actor {
  16.   var master: ActorSelection = _
  17.   //worker唯一标识
  18.   val workerId = UUID.randomUUID().toString
  19.   //内存
  20.   val memory = 40000
  21.   //cpu核心数
  22.   val cores = 4
  23.   //心跳时间
  24.   val HEARTBEAT_TIME = 10000
  26.   //建立连接
  27.   override def preStart(): Unit = {
  28.     //在master启动时会打印下面的那个协议, 可以先用这个做一个标志, 连接哪个master
  29.     //继承actor后会有一个context, 可以通过它来连接
  30.     master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master") //需要有/user, Master要和master那边创建的名字保持一致
  31.     master ! Register(workerId, memory, cores)
  32.   }
  34.   override def receive: Receive = {
  35.     //向Master注册。
  36.     case Registered(host, port) => {
  37.       println("a reply form master" + host + ":" + port)
  38.       //启动线程，进行发送心跳
  39.       import context.dispatcher
  40.       context.system.scheduler.schedule(0 millis, HEARTBEAT_TIME millis, self, SendHeartbeat)
  41.     }
  42.     //发送心跳给Master
  43.     case SendHeartbeat => {
  44.       println("from SendHeartbeat by worker")
  46.       master ! TransferHeartbeat(workerId)
  47.     }
  48.     //重新注册
  49.     case RegisterAgain => {
  50.       master ! Register(workerId, memory, cores)
  51.     }
  52.   }
  53. }
  55. object Worker {
  56.   def main(args: Array[String]) {
  57.     val host = args(0)
  58.     val port = args(1).toInt
  59.     val masterHost = args(2)
  60.     val masterPort = args(3).toInt
  62.     // 准备配置
  63.     val configStr =
  64.     s"""
  65.        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
  66.        |akka.remote.netty.tcp.hostname = "$host"
  67.        |akka.remote.netty.tcp.port = "$port"
  68.        """.stripMargin
  69.     val config = ConfigFactory.parseString(configStr)
  70.     //ActorSystem老大，辅助创建和监控下面的Actor，他是单例的
  71.     val actorSystem = ActorSystem("WorkerSystem", config)
  72.     actorSystem.actorOf(Props(new Worker(masterHost, masterPort)), "Worker")
  73.     actorSystem.awaitTermination()
  74.   }
  75. }

 

5，Maven pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>com.itunic.akka</groupId>
  8.     <artifactId>my-rpc</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  10.     <properties>
  11.         <maven.compiler.source>1.7</maven.compiler.source>
  12.         <maven.compiler.target>1.7</maven.compiler.target>
  13.         <encoding>UTF-8</encoding>
  14.         <scala.version>2.11.8</scala.version>
  15.         <scala.compat.version>2.11</scala.compat.version>
  16.     </properties>
  18.     <dependencies>
  19.         <dependency>
  20.             <groupId>org.scala-lang</groupId>
  21.             <artifactId>scala-library</artifactId>
  22.             <version>${scala.version}</version>
  23.         </dependency>
  25.         <dependency>
  26.             <groupId>com.typesafe.akka</groupId>
  27.             <artifactId>akka-actor_2.11</artifactId>
  28.             <version>2.3.15</version>
  29.         </dependency>
  32.         <dependency>
  33.             <groupId>com.typesafe.akka</groupId>
  34.             <artifactId>akka-remote_2.11</artifactId>
  35.             <version>2.3.15</version>
  36.         </dependency>
  38.     </dependencies>
  40.     <build>
  41.         <sourceDirectory>src/main/scala</sourceDirectory>
  42.         <testSourceDirectory>src/test/scala</testSourceDirectory>
  43.         <plugins>
  44.             <plugin>
  45.                 <groupId>net.alchim31.maven</groupId>
  46.                 <artifactId>scala-maven-plugin</artifactId>
  47.                 <version>3.2.2</version>
  48.                 <executions>
  49.                     <execution>
  50.                         <goals>
  51.                             <goal>compile</goal>
  52.                             <goal>testCompile</goal>
  53.                         </goals>
  54.                         <configuration>
  55.                             <args>
  56.                                 <!--<arg>-make:transitive</arg>-->
  57.                                 <arg>-dependencyfile</arg>
  58.                                 <arg>${project.build.directory}/.scala_dependencies</arg>
  59.                             </args>
  60.                         </configuration>
  61.                     </execution>
  62.                 </executions>
  63.             </plugin>
  65.             <plugin>
  66.                 <groupId>org.apache.maven.plugins</groupId>
  67.                 <artifactId>maven-shade-plugin</artifactId>
  68.                 <version>2.4.3</version>
  69.                 <executions>
  70.                     <execution>
  71.                         <phase>package</phase>
  72.                         <goals>
  73.                             <goal>shade</goal>
  74.                         </goals>
  75.                         <configuration>
  76.                             <filters>
  77.                                 <filter>
  78.                                     <artifact>*:*</artifact>
  79.                                     <excludes>
  80.                                         <exclude>META-INF/*.SF</exclude>
  81.                                         <exclude>META-INF/*.DSA</exclude>
  82.                                         <exclude>META-INF/*.RSA</exclude>
  83.                                     </excludes>
  84.                                 </filter>
  85.                             </filters>
  86.                             <transformers>
  87.                                 <transformer
  88.                                         implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
  89.                                     <resource>reference.conf</resource>
  90.                                 </transformer>
  91.                                 <transformer
  92.                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  93.                                     <mainClass>com.itunic.akka.Master</mainClass>
  94.                                 </transformer>
  95.                             </transformers>
  96.                         </configuration>
  97.                     </execution>
  98.                 </executions>
  99.             </plugin>
  100.         </plugins>
  101.     </build>
  103. </project>

 

 

  • Akka之简单的自定义RPC框架（乞丐版）已关闭评论
  • 97 views
  • A+
所属分类：未分类
avatar