Akka-Actor-RPC

发表于

整理业务逻辑分析思路

Akka实现简单通讯功能

情景模拟：学生和老师远程交流。并行处理双方的收发信息。

demo for RPC
赵老师编程思路：
程序执行入口 包名:com._51doit.akka.Master主服务器(scala: object or class?)要有程序执行入口，且需要创建该类的实例
从服务器 Worker 主从服务都需要单独启动，各自都要有专属的main方法

逻辑
先启动Master再启动Worker，worker给master建立连接发送消息，master返回消息给worker
给类添加相应的功能(类方法) worker发送消息的功能(自己写网络和IO过于繁琐，选择利用框架，集成或实现接口)
Master接收消息的功能通过集成重写Actor类中的Receive方法来实现
ctrl+i实现未被实现的方法 (重写抽象方法，可以去掉override)
如何启动服务 管理Actor的角色是ActorSystem(单例：一个JVM中只有一个即可 创建并管理Actor) 不要强行记忆代码(开发环境辅助编程)
两种管理系统：MasterActorSystem WorkerActorSystem　写代码不是从上至下编写的（灵活）

RPC自娱自乐模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
class Master extends Actor{
  //Actor中主要用来接收消息的方法
  // Receive[scala.any, scala.Unit]：
  //偏函数：做模式匹配 接收任何类型的消息，无返回,不响应Master
  //阻塞队列BlockingQueue(java.util.concurrent) Lock.class
  override def receive: Receive = {
    case "hello" => {
      println("receive a message")
    }
  }
}

object Master{
  def main(args: Array[String]): Unit = {
    //akka中Actor启动必须要绑定的配置信息，如：地址、端口号
    //接收main方法传入的字符串
    val host = args(0)
    val port = args(1).toInt
    //多个属性名和值的情况下如何处理到一个字符串里
    //如何取到属性值
    //指定akka通信的核心类
    val confStr =
    s"""
       |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname = $host
       |akka.remote.netty.tcp.port = $port
        """.stripMargin
    //parse:解析配置文件
    val config: Config = ConfigFactory.parseString(confStr)
    //apply(name: String,config Config)
    //ActorSystem相当于创建actor实例的工厂 actorOf()
    val masterActorSystem = ActorSystem("MasterActorSystem", config)
    //masterACtor理解为具有接收发送消息功能的master实例对象 功能可执行的行动者
    //Props[Master]:通过反射(传入要创建对象的类型)来创建Master的实例
    val masterActor = masterActorSystem.actorOf(Props[Master], name = "Master")
    masterActor ! "hello" //向masterActor发送一个异步消息(发出后不用等待)
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import akka.actor.Actor
class Worker extends Actor {
  //Actor中主要用来接收消息的方法
  override def receive: Receive = {
    case "hello" => {
      println("hello,Worker")
    }
  }

}

object Worker {
  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 conf: Conf = ConfigFactory.parseString(confStr)
    val workerActorSystem = ActorSystem("WorkerActor", conf)
    val workerActor = workerActorSystem.actorOf(Props[Worker], name = "Worker")
      workerActor ! "hello"
  }
}
0%