Scala语法(六) Akka与线程通信

r囧r小猫 2022-02-09 08:23 310阅读 0赞

### 前言 ###

在初期, Scala可以通过`Akka`来实现线程通信. 当然, 现在还支持使用`Netty`方式进行通信.

本章主要介绍使用`Akka`方式进行通信的写法.

--------------------

### 正文 ###

*  Master结点

import akka.actor.Actor
    import akka.actor.ActorSystem
    import com.typesafe.config.ConfigFactory
    import akka.actor.Props
    
    class AkkaMaster extends Actor{
      // start 之前
      override def preStart() : Unit = {
        println("pre master invoke.")
      }
      // 用于接收消息
      override def receive:Receive = {
        case "connect" => {
            println("a client connected.")
            sender ! "reply"
          }
        case "hello" => {println("hello")}
      }
    }
    object AkkaMaster{
      def main(args: Array[String]): Unit = {
        // 使用创建ActorSystem来创建和监控下面的Actor对象. 单例的.
          val host = "127.0.0.1"
          val port = 8090
          // 准备配置
          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)
        // 注意名称中间不要加空格
        val actorSysetm = ActorSystem("MasterSystem",config) 
        // 创建Actor
        val master = actorSysetm.actorOf(Props(new AkkaMaster),"Master")
        master ! "hello"
        // 等待信号停止
        actorSysetm.awaitTermination()
      }
    }
    
    
    // 顺利输出
    //[INFO] [04/29/2019 16:43:20.512] [main] [Remoting] Starting remoting
    //[INFO] [04/29/2019 16:43:20.770] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://MasterSystem@127.0.0.1:8090]
    //[INFO] [04/29/2019 16:43:20.771] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://MasterSystem@127.0.0.1:8090]
    //pre master invoke.
    //hello
    
    // 1. 名称中间不要加空格
    //Exception in thread "main" java.lang.IllegalArgumentException: invalid ActorSystem name [Master System], must contain only word characters (i.e. [a-zA-Z0-9] plus non-leading '-' or '_')
    //	at akka.actor.ActorSystemImpl.<init>(ActorSystem.scala:498)
    //	at akka.actor.ActorSystem$.apply(ActorSystem.scala:142)
    //	at akka.actor.ActorSystem$.apply(ActorSystem.scala:119)
    //	at com.yanxml.quick_scala.multi.akka.AkkaMaster$.main(AkkaMaster.scala:33)
    //	at com.yanxml.quick_scala.multi.akka.AkkaMaster.main(AkkaMaster.scala)

*  Worker结点

import akka.actor.Actor
    import akka.actor.ActorSelection
    import akka.actor.ActorSystem
    import akka.actor.Props
    import com.typesafe.config.ConfigFactory
    
    class AkkaWorker extends Actor{
      // 成员变量
      val master:ActorSelection = null
      // 建立链接
      override def preStart():Unit = {
        val master = context.actorSelection("akka.tcp://MasterSystem@127.0.0.1:8090/user/Master")
        println(master)
        master ! "connect"
      }
      
      override def receive:Receive = {
        case "reply" => {
          println("a reply from master")
        }
      }
    }
    
    object AkkaWorker{
       def main(args: Array[String]): Unit = {
         // 使用创建ActorSystem来创建和监控下面的Actor对象. 单例的.
          val host = "127.0.0.1"
          val port = 8091
          // 准备配置
          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)
        // 注意名称中间不要加空格
        val actorSysetm = ActorSystem("WorkerSystem",config) 
        // 创建Actor
        val master = actorSysetm.actorOf(Props(new AkkaWorker),"Worker")
        master ! "hello"
        // 等待信号停止
        actorSysetm.awaitTermination()
       }
    }

--------------------

### 模拟RPC ###

在模拟RPC中主要有这样的流程.

其中主要包括两个结点: `Worker结点`&`Master结点`.

*  运行流程:
    
     *  `Master结点`先进行启动.
     *  `Worker结点`后进行启动.
     *  `Worker结点`向`Master结点`发送注册消息.
     *  `Matser结点`接收注册消息, 并进行记录. 并将主结点的地址返回给`Worker结点`(模拟Master是集群的情况).并记录,最后的通信时间作为心跳标志.
     *  `Worker`结点接收主结点地址, 并形成通信链接. 开始通信. 并定时发送心跳消息.

改造上方的`Demo`代码. 其基本代码如下所示:

*  RemoteMessage

trait RemoteMessage  extends Serializable{
      
    }
    
    // Worker -> Master 用来封装Worker信息 
    case class RegisterWorker(id:String,memory:Int,cores:Int)
    
    class WorkerInfo(val id:String, val memory:Int, val cores:Int){
      // 上一次心跳
      var heartbeatTime:String = _
    }

*  Master

import akka.actor.Actor
    import akka.actor.ActorSystem
    import com.typesafe.config.ConfigFactory
    import akka.actor.Props
    import scala.collection.immutable.HashMap
    
    private [simulate] class AkkaMaster extends Actor{
      val idToWorker = new scala.collection.mutable.HashMap[String,WorkerInfo]()
      // start 之前
      override def preStart() : Unit = {
        println("pre master invoke.")
      }
      // 用于接收消息
      override def receive:Receive = {
        case "connect" => {
            println("a client connected.")
            sender ! "reply"
          }
        case "hello" => {println("hello")}
        // 传输样例类
        case RegisterWorker(id,memory,cores)=>{
          if(!idToWorker.contains(id)){
            idToWorker.put(id, new WorkerInfo(id,memory,cores))
          }
          sender ! "123"
        }
      }
    }
    object AkkaMaster{
      def main(args: Array[String]): Unit = {
        // 使用创建ActorSystem来创建和监控下面的Actor对象. 单例的.
          val host = "127.0.0.1"
          val port = 8090
          // 准备配置
          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)
        // 注意名称中间不要加空格
        val actorSysetm = ActorSystem("MasterSystem",config) 
        // 创建Actor
        val master = actorSysetm.actorOf(Props(new AkkaMaster),"Master")
        master ! "hello"
        // 等待信号停止
        actorSysetm.awaitTermination()
      }
    }
    
    
    // 顺利输出
    //[INFO] [04/29/2019 16:43:20.512] [main] [Remoting] Starting remoting
    //[INFO] [04/29/2019 16:43:20.770] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://MasterSystem@127.0.0.1:8090]
    //[INFO] [04/29/2019 16:43:20.771] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://MasterSystem@127.0.0.1:8090]
    //pre master invoke.
    //hello
    
    // 1. 名称中间不要加空格
    //Exception in thread "main" java.lang.IllegalArgumentException: invalid ActorSystem name [Master System], must contain only word characters (i.e. [a-zA-Z0-9] plus non-leading '-' or '_')
    //	at akka.actor.ActorSystemImpl.<init>(ActorSystem.scala:498)
    //	at akka.actor.ActorSystem$.apply(ActorSystem.scala:142)
    //	at akka.actor.ActorSystem$.apply(ActorSystem.scala:119)
    //	at com.yanxml.quick_scala.multi.akka.AkkaMaster$.main(AkkaMaster.scala:33)
    //	at com.yanxml.quick_scala.multi.akka.AkkaMaster.main(AkkaMaster.scala)

*  Worker

import akka.actor.Actor
    import akka.actor.ActorSelection
    import akka.actor.ActorSystem
    import akka.actor.Props
    import com.typesafe.config.ConfigFactory
    import com.yanxml.quick_scala.multi.akka.simulate.RegisterWorker
    import java.util.UUID
    
    private [simulate] class  AkkaWorker(val masterHost:String, val masterPort:String, val memory:Int, val cores:Int) extends Actor{
      // 成员变量
      val master:ActorSelection = null
      // 建立链接
      override def preStart():Unit = {
        // 和Master建立链接
        val master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master")
        val workerId = UUID.randomUUID().toString()
        //println(master)
        // 向Master发送消息
        master ! RegisterWorker(workerId,memory,cores)
      }
      
      override def receive:Receive = {
        case "reply" => {
          println("a reply from master")
        }
      }
    }
    
    object AkkaWorker{
       def main(args: Array[String]): Unit = {
         // 使用创建ActorSystem来创建和监控下面的Actor对象. 单例的.
          val host = "127.0.0.1"
          val port = 8091
          // 准备配置
          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)
        // 注意名称中间不要加空格
        val actorSysetm = ActorSystem("WorkerSystem",config) 
        // 创建Actor
        val master = actorSysetm.actorOf(Props(new AkkaWorker("127.0.0.1","8090",2,2)),"Worker")
        master ! "hello"
        // 等待信号停止
        actorSysetm.awaitTermination()
       }
    }

注: 后续的通信逻辑就是丰富双方的`receive()`方法.