• 项目实战：利用Akka的Actor编程模型，实现2个进程间的通信

• 架构图:

  • Master、Master Actor <======通过RPC相互通信======> Worker、Worker Actor

• 重要类介绍

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

    • 1、ActorSystem是一个进程中的老大，它负责创建和监督actor
    • 2、ActorSystem是一个单例对象
    • 3、actor负责通信

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

    • 1、preStart()方法：该方法在Actor对象构造方法执行后执行，整个Actor生命周期中仅执行一次。
    • 2、receive()方法：该方法在Actor的preStart方法执行完成后执行，用于接收消息，会被反复执行。
• 具体代码：
  Master类：

package cn.akka.rpc

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

//todo:利用akka中的actor模型,实现两个进程之间的通信------Master端
class Master extends Actor {

  println("====== Master constructor invoked ======")

  //todo:它会在构造代码块之后被调用，并且只会被执行一次
  override def preStart(): Unit = {
    println("====== preStart method invoked =======")
  }

  //todo:它会在preStart方法执行后被调用，并且是一直循环被执行
  override def receive: Receive = {
    case "connect" => {
      println("====== a client connected ======")

      sender ! "success"
    }
  }

}

object Master {

  def main(args: Array[String]): Unit = {
    //todo:定义master的IP地址
    val host = args(0)

    //todo:定义master的port端口
    val port = args(1)

    //todo:准备配置文件信息
    val configStr =
      s"""
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname = "$host"
        |akka.remote.netty.tcp.port = "$port"
      """.stripMargin

    //todo:利用ConfigFactory解析字符串，获取对应的配置
    val config = ConfigFactory.parseString(configStr)

    //todo:1.创建ActorSystem,它是整个进程的老大，负责创建和监督actor，它是单例对象
    val masterActorSystem: ActorSystem = ActorSystem("masterActorSystem", config)

    //todo:2.ActorSystem创建actor
    val masterActor: ActorRef = masterActorSystem.actorOf(Props(new Master), "masterActor")

    //todo:3.向master actor发送消息
    //masterActor ! "connect"

  }
}

Worker类：

package cn.akka.rpc

import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

//todo:利用akka中的actor模型,实现两个进程之间的通信------Worker端
class Worker extends Actor {

  println("====== Worker constructor invoked ======")

  override def preStart(): Unit = {
    println("====== preStart method invoked ======")
    //todo:1.拿到 master actor 的引用
    //todo:2.AcotorContext上下文对象调用actorSelection
    //todo:3.需要：通信协议、master的ip地址、master的端口、master对应的ActorSystem,actor层级
    val master: ActorSelection = context.actorSelection("akka.tcp://masterActorSystem@192.168.106.26:8888/user/masterActor")

    //todo:4.向master发送消息
    master ! "connect"

  }

  override def receive: Receive = {
    case "connect" => {
      println("====== a client connected ======")
    }

    case "success" => {
      println("====== 我已经注册成功！！！ ======")
    }

  }

}

object Worker {

  def main(args: Array[String]): Unit = {
    //todo:worker的IP地址
    val host = args(0)

    //todo:worker的端口
    val port = args(1)

    //todo:准备配置信息
    val configStr =
      s"""
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname = "$host"
        |akka.remote.netty.tcp.port = "$port"
      """.stripMargin
    //todo:利用ConfigActor解析字符串配置信息
    val config = ConfigFactory.parseString(configStr)

    //todo:1.创建ActorSystem对象，它是整个进程的老大，负责创建和监督actor，它是单例对象
    val workerActorSystem: ActorSystem = ActorSystem("workerActorSystem", config)

    //todo:2.利用workerActorSystem创建Worker的actor
    val workerActor: ActorRef = workerActorSystem.actorOf(Props(new Worker), "workerActor")

    //todo:3.向workerActor发送消息
    workerActor ! "connect"

  }

}

• 运行结果：

Master：

====== Master constructor invoked ======
====== preStart method invoked =======
====== a client connected ======

Worker:

====== Worker constructor invoked ======
====== preStart method invoked ======
====== a client connected ======
====== 我已经注册成功！！！ ======

每一个成功的背后都有无数个无人知晓的黑夜。

因为

夜晚，是超越对手的最佳时机。

===================== 码农1024 =====================#蔺光岭#