【Scala】利用Akka的actor编程模型，实现2个进程间的通信

文章目录

• 步骤
• 一、创建maven工程，导入jar包
• 二、master进程代码开发
• 三、worker进程代码开发
• 四、控制台结果

---

步骤

一、创建maven工程，导入jar包

<properties>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <encoding>UTF-8</encoding>
    <scala.version>2.11.8</scala.version>
    <scala.compat.version>2.11</scala.compat.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
        <version>${scala.version}</version>
    </dependency>

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

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

</dependencies>

<build>
    <sourceDirectory>src/main/scala</sourceDirectory>
    <testSourceDirectory>src/test/scala</testSourceDirectory>
    <plugins>
        <!-- 限制jdk的编译版本插件 -->
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.0</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
                <encoding>UTF-8</encoding>
            </configuration>
        </plugin>


        <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>-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></mainClass>
                            </transformer>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

二、master进程代码开发

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

// todo 利用akka实现两个进程间的通信---master端
class Master extends Actor{

  println("主构造器开始执行！")

  //初始化方法，在receive方法之前执行，在构造方法之后执行，执行一次
  override def preStart(): Unit = {

  }

  //receive方法用于akka actor中消息的接收处理，可以持续不断接收消息
  override def receive: Receive = {
    case "connect" => {
      println("Receive Login information success,logining.....please wait.....")
      //返回注册成功
      sender ! "success"
    }
  }
}

object Master{
  def main(args: Array[String]): Unit = {
    //给host定义一个参数，用于接收主机
    val host = args(0)
    //给port定义一个参数，用于接收端口
    val port = args(1)

    //创建conf需要解析的String
    val confStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
         |""".stripMargin

    //创建ActorSystem所需的configuration
    val conf = ConfigFactory.parseString(confStr)

    //创建ActorSystem，由其创建监督管理其他众多的Actor
    val masterActorSystem = ActorSystem.create("masterActorSystem", conf)

    //通过ActorSystem创建masterActor
    val master: ActorRef = masterActorSystem.actorOf(Props(new Master), "masterActor")

    //master测试自连
    //    master ! "connect"
  }
}

三、worker进程代码开发

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

// todo 利用akka实现两个进程间的通信---worker端
class Worker extends Actor{

  override def preStart(): Unit = {
    //获取master actor的引用（需要通信协议、IP地址、端口号、actorSystem、masterActor、actor层级）
    val master: ActorSelection = context.actorSelection("akka.tcp://masterActorSystem@192.168.0.108:12321/user/masterActor")
    //给master发送注册信息
    master ! "connect"
  }

  override def receive: Receive = {
    case "connect" => println("WorkerClient is connecting....")
    case "success" => println("Congratulation! Login success!")
  }
}

object Worker{
  def main(args: Array[String]): Unit = {
    //worker host
    val host = args(0)
    //worker port
    val port = args(1)

    //创建conf需要的str
    val confStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
         |""".stripMargin

    //创建workerActorSystem需要的configuration
    val config: Config = ConfigFactory.parseString(confStr)

    //创建ActorSystem
    val workerActorSystem: ActorSystem = ActorSystem.create("workerActorSystem", config)

    //创建workActor
    val worker: ActorRef = workerActorSystem.actorOf(Props(new Worker), "workerActor")
  }
}

四、控制台结果

master控制台

[INFO] [04/12/2020 11:46:00.234] [main] [Remoting] Starting remoting
[INFO] [04/12/2020 11:46:00.413] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://masterActorSystem@192.168.0.108:12321]
[INFO] [04/12/2020 11:46:00.415] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://masterActorSystem@192.168.0.108:12321]
主构造器开始执行！
Receive Login information success,logining.....please wait.....

worker控制台

[INFO] [04/12/2020 11:46:07.019] [main] [Remoting] Starting remoting
[INFO] [04/12/2020 11:46:07.371] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://workerActorSystem@192.168.0.108:12322]
[INFO] [04/12/2020 11:46:07.374] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://workerActorSystem@192.168.0.108:12322]
Congratulation! Login success!