Scala系列——Akka并发编程

素颜马尾好姑娘i 2022-04-13 12:14 214阅读 0赞

# 一、Akka简介 #

## 1、基本介绍 ##

Akka基于Actor模型，提供了一个用于构建可扩展的（Scalable）、弹性的（Resilient）、快速响应的（Responsive）应用程序的平台。

Actor模型：在计算机科学领域，Actor模型是一个并行计算（Concurrent Computation）模型，它把actor作为并行计算的基本元素来对待：为响应一个接收到的消息，一个actor能够自己做出一些决策，如创建更多的actor，或发送更多的消息，或者确定如何去响应接收到的下一个消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIyMTcyMTMz_size_16_color_FFFFFF_t_70][]

Actor是Akka中最核心的概念，它是一个封装了状态和行为的对象，Actor之间可以通过交换消息的方式进行通信，每个Actor都有自己的收件箱（Mailbox）。通过Actor能够简化锁及线程管理，可以非常容易地开发出正确地并发程序和并行系统，Actor具有如下特性：

*  提供了一种高级抽象，能够简化在并发（Concurrency）/并行（Parallelism）应用场景下的编程开发
 *  提供了异步非阻塞的、高性能的事件驱动编程模型
 *  超级轻量级事件处理（每GB堆内存几百万Actor）

## 2、重要类介绍 ##

### 1.ActorSystem ###

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

## 2.Actor ##

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

*  preStart()方法：该方法在Actor对象构造方法执行后执行，整个Actor生命周期中仅执行一次。
 *  receive()方法：该方法在Actor的preStart方法执行完成后执行，用于接收消息，会被反复执行。

# 二、RPC项目搭建 #

## 1、需求 ##

目前大多数的分布式架构底层通信都是通过RPC实现的，RPC框架非常多，比如前我们学过的Hadoop项目的RPC通信框架，但是Hadoop在设计之初就是为了运行长达数小时的批量而设计的，在某些极端的情况下，任务提交的延迟很高，所有Hadoop的RPC显得有些笨重。

Spark 的RPC是通过Akka类库实现的，Akka用Scala语言开发，基于Actor并发模型实现，Akka具有高可靠、高性能、可扩展等特点，使用Akka可以轻松实现分布式RPC功能。

## 2、架构图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIyMTcyMTMz_size_16_color_FFFFFF_t_70 1][]

## 3、代码实现 ##

首先maven构建如下：

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.scala</groupId>
        <artifactId>my_rpc</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            <maven.compiler.source>1.7</maven.compiler.source>
            <maven.compiler.target>1.7</maven.compiler.target>
            <encoding>UTF-8</encoding>
            <scala.version>2.10.6</scala.version>
            <scala.compat.version>2.10</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.10</artifactId>
                <version>2.3.14</version>
            </dependency>
    
            <dependency>
                <groupId>com.typesafe.akka</groupId>
                <artifactId>akka-remote_2.10</artifactId>
                <version>2.3.14</version>
            </dependency>
    
        </dependencies>
    
        <build>
            
            <sourceDirectory>src/main/scala</sourceDirectory>
            <testSourceDirectory>src/test/scala</testSourceDirectory>
            <plugins>
                <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>-make:transitive</arg>
                                    <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>com.scala.rpc.Worker</mainClass>
                                    </transformer>
                                </transformers>
                            </configuration>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>
    
    </project>

然后Master如下：

class Master extends Actor{
      println("constructor invoked")
    
      override def preStart(): Unit = {
        println("preStart invoked")
      }
    
      // 用于接收消息
      override def receive: Receive = {
        case "connect" => {
          println("a client connected")
          sender ! "reply"
        }
    
        case "hello" => {
          println("hello")
        }
      }
    }
    
    object Master {
      def main(args: Array[String]) {
    
        val host = args(0)
        val port = args(1).toInt
        // 准备配置
        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)
        //ActorSystem老大，辅助创建和监控下面的Actor，他是单例的
        val actorSystem = ActorSystem("MasterSystem", config)
        //创建Actor, 起个名字
        val master = actorSystem.actorOf(Props[Master], "Master")//Master主构造器会执行
        master ! "hello"  //发送信息
        actorSystem.awaitTermination()  //让进程等待着, 先别结束
      }
    }

Worker如下：

class Worker(val masterHost: String, val masterPort: Int) extends Actor{
    
      var master : ActorSelection = _
      //建立连接
      override def preStart(): Unit = {
        //在master启动时会打印下面的那个协议, 可以先用这个做一个标志, 连接哪个master
        //继承actor后会有一个context, 可以通过它来连接
        //需要有/user, Master要和master那边创建的名字保持一致(val master = actorSystem.actorOf(Props[Master], "Master"))
        master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master")
        master ! "connect"
      }
    
      override def receive: Receive = {
        case "reply" => {
          println("a reply form master")
        }
      }
    }
    
    object Worker {
      def main(args: Array[String]) {
        val host = args(0)
        val port = args(1).toInt
        val masterHost = args(2)
        val masterPort = args(3).toInt
        // 准备配置
        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)
        //ActorSystem老大，辅助创建和监控下面的Actor，他是单例的
        val actorSystem = ActorSystem("WorkerSystem", config)
        actorSystem.actorOf(Props(new Worker(masterHost, masterPort)), "Worker")
        actorSystem.awaitTermination()
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIyMTcyMTMz_size_16_color_FFFFFF_t_70]: /images/20220413/c46fcb93683a473394dc7800ee0417c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIyMTcyMTMz_size_16_color_FFFFFF_t_70 1]: /images/20220413/860064dcda4f43408c324f199bdab331.png