您的位置: 首页  >  文章  >  使用scala的akka实现一个RPC通信的demo

使用scala的akka实现一个RPC通信的demo

分类: 文章 • 2024-05-18 14:17:46

 

  1. 目前大多数的分布式架构底层通信都是通过RPC实现的，RPC框架非常多，比如前我们学过的Hadoop项目的RPC通信框架，但是Hadoop在设计之初就是为了运行长达数小时的批量而设计的，在某些极端的情况下，任务提交的延迟很高，所有Hadoop的RPC显得有些笨重，Spark 的RPC是通过Akka类库实现的，Akka用Scala语言开发，基于Actor并发模型实现，Akka具有高可靠、高性能、可扩展等特点，使用Akka可以轻松实现分布式RPC功能。

  2. Akka基于Actor模型，提供了一个用于构建可扩展的（Scalable）、弹性的（Resilient）、快速响应的（Responsive）应用程序的平台。

    Actor模型：在计算机科学领域，Actor模型是一个并行计算（Concurrent Computation）模型，它把actor作为并行计算的基本元素来对待：为响应一个接收到的消息，一个actor能够自己做出一些决策，如创建更多的actor，或发送更多的消息，或者确定如何去响应接收到的下一个消息。

    Actor是Akka中最核心的概念，它是一个封装了状态和行为的对象，Actor之间可以通过交换消息的方式进行通信，每个Actor都有自己的收件箱（Mailbox）。通过Actor能够简化锁及线程管理，可以非常容易地开发出正确地并发程序和并行系统，Actor具有如下特性：

    1.提供了一种高级抽象，能够简化在并发（Concurrency）/并行（Parallelism）应用场景下的编程开发

    2.提供了异步非阻塞的、高性能的事件驱动编程模型

    3.超级轻量级事件处理（每GB堆内存几百万Actor）

  3. 架构图使用scala的akka实现一个RPC通信的demo

  4. 重要类

      4.1ActorSystem

           在Akka中，ActorSystem是一个重量级的结构，他需要分配多个线程，所以在实际应用中，ActorSystem通常是一个单例              对象，我们可以使用这个ActorSystem创建很多Actor。

     4.2Actor

           在Akka中，Actor负责通信，在Actor中有一些重要的生命周期方法。

           preStart()方法：该方法在Actor对象构造方法执行后执行，整个Actor生命周期中仅执行一次。

           receive()方法：该方法在Actor的preStart方法执行完成后执行，用于接收消息，会被反复执行。

     4.3 Master

package cn.maize.akka

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.collection.mutable
import scala.concurrent.duration._

class Master(val host:String,val port:Int) extends Actor{
  //保存WorkerID 到 WorkerInfo的映射
  val idToWorkers = new mutable.HashMap[String,WorkerInfo]()
  //保存所的WorkerInfo信息
  val workers = new mutable.HashSet[WorkerInfo]()

  val CHECK_INTERVAL = 10000

  override def preStart(): Unit = {
    //导入隐式转换
    import context.dispatcher
    context.system.scheduler.schedule(0 millis,CHECK_INTERVAL millis,self,CheckTimeOutWorker)
  }

  override def receive: Receive = {
    //Worker发送个Master的注册消息
    case RegisterWorker(workerId,cores,memory) => {
      if(!idToWorkers.contains(workerId)){
        //封装worker发送的信息
        val workerInfo = new WorkerInfo(workerId,cores,memory)
        //保存workerInfo
        idToWorkers(workerId) = workerInfo
        workers += workerInfo
        //Master向Worker反馈注册成功的消息
        sender ! RegisteredWorker(s"akka.tcp://${Master.MASTER_SYSTEM}@$host:$port/user/${Master.MASTER_NAME}")
      }
    }

    //Worker发送给Master的心跳信息
    case Heartbeat(workerId) => {
      if(idToWorkers.contains(workerId)){
        val workerInfo = idToWorkers(workerId)
        //更新上一次心跳时间
        workerInfo.lastHeartbeatTime = System.currentTimeMillis()
      }
    }

    //检测超时的Worker
    case CheckTimeOutWorker => {
      val currentTime = System.currentTimeMillis()
      val deadWorkers:mutable.HashSet[WorkerInfo] = workers.filter(w => currentTime - w.lastHeartbeatTime > CHECK_INTERVAL)
      deadWorkers.foreach(w => {
        idToWorkers -= w.id
        workers -= w
      })
      println(currentTime + " alive worker size : "+workers.size)
    }
  }
}

object Master{

  val MASTER_SYSTEM = "MasterActorSystem"
  val MASTER_NAME = "master"

  def main(args: Array[String]): Unit = {
    val host = args(0)
    val port = args(1).toInt

    val confStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
       """.stripMargin

    val config = ConfigFactory.parseString(confStr)
    //ActorSystem是单例的，用于创建Acotor并监控actor
    val actorSystem = ActorSystem.create(MASTER_SYSTEM,config)
    //通过ActorSystem创建Actor
    val master = actorSystem.actorOf(Props(new Master(host,port)),MASTER_NAME)
    actorSystem.awaitTermination()
  }

}

4.4 Worker

package cn.maize.akka

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._

class Worker(val cores:Int,val memory:Long,val masterHost:String,val masterPort:Int) extends Actor{

  //Master的引用
  var master :ActorSelection = _
  //workerId
  val workerId = UUID.randomUUID().toString
  //masterUrl
  var masterUrl:String = _
  //心跳间隔
  var HEARTBEAT_INTERVAL = 10000

  //preStart在构造器之后receive之前执行
  override def preStart(): Unit = {
    //首先跟Master建立连接
    master = context.actorSelection(s"akka.tcp://${Master.MASTER_SYSTEM}@${masterHost}:${masterPort}/user/${Master.MASTER_NAME}")
    //通过master的引用向Master发送注册消息
    master ! RegisterWorker(workerId,cores,memory)
  }

  override def receive: Receive = {
    //Master发送给Worker注册成功的消息
    case RegisteredWorker(masterUrl) => {
      //以后Master会改造成分布式的，向master注册时由zookeeper分配，
      //所以注册成功后需要拿到masterUrl以便后面传输数据
      this.masterUrl = masterUrl
      //启动定时任务，向Master发送心跳
      //导入隐式转换
      import context.dispatcher
      context.system.scheduler.schedule(0 millis,HEARTBEAT_INTERVAL millis,self,SendHeartbeat)
    }

    case SendHeartbeat => {
      //向Master发送心跳
      master ! Heartbeat(workerId)
    }
  }
}

object Worker{
  def main(args: Array[String]): Unit = {
    //Worker的地址和端口
    val host = args(0)
    val port = args(1).toInt
    val cores = args(2).toInt
    val memory = args(3).toLong
    //Master的地址和端口
    val masterHost = args(4)
    val masterPort = args(5).toInt

    val confStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
      """.stripMargin
    val config = ConfigFactory.parseString(confStr)
    //单例的ActorSystem
    val actorSystem = ActorSystem.apply("WorkerActorSystem",config)
    //通过actorSystem来创建Actor
    val worker = actorSystem.actorOf(Props(new Worker(cores,memory,masterHost,masterPort)),"Worker")
    actorSystem.awaitTermination()
  }

}

 4.5WorkerInfo

package cn.maize.akka

class WorkerInfo(val id:String,val cores:Int,var memory:Long) {

  //最近一次的心跳时间
  var lastHeartbeatTime:Long = _

}

4.6 Message

package cn.maize.akka

//进程间信息传送，实现scala的Serializable接口
trait Message extends Serializable
//Worker  =>  Master
case class RegisterWorker(id:String,cores:Int,memery:Long) extends Message
//Master  =>  Worker
case class RegisteredWorker(masterUrl:String) extends Message
//Worker  =>  Master
case class Heartbeat(id:String) extends Message
//Worker internal message
case object SendHeartbeat
//Master internal message
case object CheckTimeOutWorker

4.7 pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>cn.maize</groupId>
    <artifactId>MyRPC</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>1.7</maven.compiler.source>
        <maven.compiler.target>1.7</maven.compiler.target>
        <encoding>UTF-8</encoding>
        <scala.version>2.10.6</scala.version>
        <scala.compat.version>2.10</scala.compat.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.11.8</version>
        </dependency>

        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor_2.11</artifactId>
            <version>2.3.16</version>
        </dependency>

        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-remote_2.11</artifactId>
            <version>2.3.16</version>
        </dependency>

        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>

        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>2.8.2</version>
        </dependency>
    </dependencies>

    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-make:transitive</arg>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>reference.conf</resource>
                                </transformer>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>cn.maize.akka.Worker</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

 

相关推荐