Scala学习之路（十）Scala的Actor

一、Scala中的并发编程

1、Java中的并发编程

①Java中的并发编程基本上满足了事件之间相互独立，但是事件能够同时发生的场景的需要。

②Java中的并发编程是基于共享数据和加锁的一种机制，即会有一个共享的数据，然后有若干个线程去访问这个共享的数据(主要是对这个共享的数据进行修改)，同时Java利用加锁的机制(即synchronized)来确保同一时间只有一个线程对我们的共享数据进行访问,进而保证共享数据的一致性。

③Java中的并发编程存在资源争夺和死锁等多种问题，因此程序越大问题越麻烦。

2、Scala中的并发编程

①Scala中的并发编程思想与Java中的并发编程思想完全不一样，Scala中的Actor是一种不共享数据，依赖于消息传递的一种并发编程模式，避免了死锁、资源争夺等情况。在具体实现的过程中，Scala中的Actor会不断的循环自己的邮箱，并通过receive偏函数进行消息的模式匹配并进行相应的处理。

②如果Actor A和 Actor B要相互沟通的话，首先A要给B传递一个消息，B会有一个收件箱，然后B会不断的循环自己的收件箱，若看见A发过来的消息，B就会解析A的消息并执行，处理完之后就有可能将处理的结果通过邮件的方式发送给A。

二、Scala中的Actor

1、什么是Actor

一个actor是一个容器，它包含 状态，行为，信箱，子Actor 和监管策略，所有这些包含在一个ActorReference（Actor引用）里。一个actor需要与外界隔离才能从actor模型中获益，所以actor是以actor引用的形式展现给外界的。

2、ActorSystem的层次结构

如果一个Actor中的业务逻辑非常复杂，为了降低代码的复杂度，可以将其拆分成多个子任务（在一个actor的内部可以创建一个或多个actor，actor的创建者也是该actor的监控者）

一个ActorSystem应该被正确规划，例如哪一个Actor负责监控，监控什么等等：

负责分发的actor管理接受任务的actor
拥有重要数据的actor，找出所有可能丢失数据的子actor，并且处理他们的错误。

3、ActorPath

ActorPath是通过字符串描述Actor的层级关系，并唯一标识一个Actor的方法。

ActorPath包含协议，位置和 Actor层级关系

//本地path
"akka://my-sys/user/service-a/worker1"   

//远程path　　akka.tcp://（ActorSystem的名称）@（远程地址的IP）：（远程地址的端口）/user/（Actor的名称）
"akka.tcp://my-sys@host.example.com:5678/user/service-b" 

//akka集群
"cluster://my-cluster/service-c"

远程地址不清楚是多少的话，可以在远程的服务启动的时候查看

Scala学习之路（十）Scala的Actor

4、获取Actor Reference

获取Actor引用的方式有两种：创建和查找。

要创建Actor，可以调用ActorSystem.actorOf(..)，它创建的actor在guardian actor之下，接着可以调用ActorContext的actorOf(…) 在刚才创建的Actor内生成一个actor树。这些方法会返回新创建的actor的引用，每一个actor都可以通过访问ActorContext来获得自己（self），子Actor（children，child）和父actor（parent）。

要查找Actor Reference，可以调用ActorSystem或ActorContext的actorSelection(“path”)，在查找ActorRef时，可以使用相对路径或绝对路径，如果是相对路径，可以用 .. 来表示parent actor。

actorOf / actorSelection / actorFor的区别

actorOf 创建一个新的actor，创建的actor为调用该方法所属的context的直接子actor。

actorSelection 查找现有actor，并不会创建新的actor。

actorFor 查找现有actor，不创建新的actor，已过时。

5、Actor和ActorSystem

Actor：
就是用来做消息传递的
用来接收和发送消息的，一个actor就相当于是一个老师或者是学生。
如果我们想要多个老师，或者学生，就需要创建多个actor实例。
ActorSystem:
用来创建和管理actor，并且还需要监控Actor。ActorSystem是单例的（object）
在同一个进程里面，只需要一个ActorSystem就可以了

三、Actor的示例

1、示例说明

Scala学习之路（十）Scala的Actor

2、代码实现

MyResourceManager.scala（服务端）

package com.rpc

import akka.actor._
import com.typesafe.config.{Config, ConfigFactory}

import scala.collection.mutable

class MyResourceManager(var resourceManagerHostName:String, var resourceManagerPort:Int) extends Actor {
  /**
    * 定义一个Map,接受MyNodeManager的注册信息，key是主机名，
    * value是NodeManagerInfo对象，里面存储主机名、CPU和内存信息
    * */
  var registerMap = new mutable.HashMap[String,NodeManagerInfo]()
  /**
    * 定义一个Set,接受MyNodeManager的注册信息，key是主机名，
    * value是NodeManagerInfo对象，里面存储主机名、CPU和内存信息
    * 实际上和上面的Map里面存档内容一样，容易变历，可以不用写，主要是模仿后面Spark里面的内容
    * 方便到时理解Spark源码
    * */
  var registerSet = new mutable.HashSet[NodeManagerInfo]()


  override def preStart(): Unit = {
    import scala.concurrent.duration._
    import context.dispatcher
    context.system.scheduler.schedule(0 millis, 5000 millis, self,CheckTimeOut)
  }

  //对MyNodeManager传过来的信息进行匹配
  override def receive: Receive = {
    //匹配到NodeManager的注册信息进行对应处理
    case NodeManagerRegisterMsg(nodeManagerID,cpu,memory) => {
      //将注册信息实例化为一个NodeManagerInfo对象
      val registerMsg = new NodeManagerInfo(nodeManagerID,cpu,memory)
      //将注册信息存储到registerMap和registerSet里面，key是主机名，value是NodeManagerInfo对象
      registerMap.put(nodeManagerID,registerMsg)
      registerSet += registerMsg
      //注册成功之后，反馈个MyNodeManager一个成功的信息
      sender() ! new RegisterFeedbackMsg("注册成功!" + resourceManagerHostName+":"+resourceManagerPort)
    }
    //匹配到心跳信息做相应处理
    case HeartBeat(nodeManagerID) => {
      //获取当前时间
      val time:Long = System.currentTimeMillis()
      //根据nodeManagerID获取NodeManagerInfo对象
      val info = registerMap(nodeManagerID)
      info.lastHeartBeatTime = time
      //更新registerMap和registerSet里面nodeManagerID对应的NodeManagerInfo对象信息(最后一次心跳时间)
      registerMap(nodeManagerID) = info
      registerSet += info
    }
    //检测超时，对超时的数据从集合中删除
    case CheckTimeOut => {
      var time = System.currentTimeMillis()
      registerSet
        .filter( nm => time - nm.lastHeartBeatTime > 10000)
        .foreach(deadnm => {
          registerSet -= deadnm
          registerMap.remove(deadnm.nodeManagerID)
        })
      println("当前注册成功的节点数：" + registerMap.size)
    }
  }
}

object MyResourceManager {
  def main(args: Array[String]): Unit = {
    /**
      * 传参：
      *   ResourceManager的主机地址、端口号
      * */
    val RM_HOSTNAME = args(0)
    val RM_PORT = args(1).toInt

    val str:String =
      """
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname =localhost
        |akka.remote.netty.tcp.port=19888
      """.stripMargin
    val conf: Config = ConfigFactory.parseString(str)
    val actorSystem = ActorSystem(Conf.RMAS,conf)
    actorSystem.actorOf(Props(new MyResourceManager(RM_HOSTNAME,RM_PORT)),Conf.RMA)
  }
}

View Code