scala 系列之 20akka 的 actor 并发编程

文章 潘牛 ⋅ 于 2021-07-08 22:56:56 ⋅ 265 阅读

24 Akka

24.1 Akka 概述

​ Spark的RPC是通过Akka类库实现的,Akka用Scala语言开发,基于Actor并发模型实现;

​ Akka具有高可靠、高性能、可扩展等特点,使用Akka可以轻松实现分布式RPC功能。

​ Actor是Akka中最核心的概念,它是一个封装了状态和行为的对象,Actor之间可以通过交换消息的方式进行通信,每个Actor都有自己的收件箱(MailBox)。

​ 通过Actor能够简化锁及线程管理,可以非常容易地开发出正确地并发程序和并行系统。

Akka 具有如下特性:

​ 1)提供了一种高级抽象,能够简化在并发(Concurrency)/并行(Parallelism)应用场景下的编程开发;

​ 2)提供了异步非阻塞的、高性能的事件驱动编程模型;

​ 3)超级轻量级事件处理(每GB堆内存几百万Actor);

24.2 Akka 组成及架构原理

ActorSystem

在Akka中,ActorSystem是一个重量级的结构。

ActorSystem 的职责是 负责创建并管理其创建的Actor,ActorSystem的单例的,一个JVM进程中有一个即可,而Actor是多例的。

Actor

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

1)preStart()方法:该方法在Actor对象构造方法执行后执行,整 个Actor生命周期中仅执行一次, 就像 mapreduce里的 setup()

2)receive()方法:该方法在Actor的preStart方法执行完成后执行,用于接收消息,会被反复执行, 就像mapreduce里的map()

每个actor 对象有对应的外部引用xxxRef,可以通过该 actor 对象的外部引用与actor通信。

file

akka的架构原理

file

其中:

​ mailbox负责存储actor收到的消息,dispatcher负责从mailbox取消息,分配线程给actor执行具体的业务逻辑。

​ sender引用代表最近收到消息的发送actor,通常用于回消息,比如 sender() !xxxx。

24.3 Akka 的使用

使用Akka需要增加这两个的pom依赖

<!-- 添加 akka 的 actor 依赖 -->
<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-actor_2.11</artifactId>
    <version>2.4.17</version>
</dependency>
<!-- 多进程之间的 Actor 通信 -->
<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-remote_2.11</artifactId>
    <version>2.4.17</version>
</dependency>

24.3.1 发送给自己

步骤:

1)创建ActorSystem

file

2)定义处理信息的Actor实现类

class HelleAkka extends Actor{
  //接受消息
  override def receive: Receive = {
    //接受消息的处理逻辑

  }
}

file

3)创建目标Actor的ActorRef对象

file

4)往目标Actor的ActorRef对象发送信息

package day05
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
class HelloAkkaActor extends Actor{
  override def preStart() ={
    println("do preStart()")
  }
  override def receive: Receive = {
    case "start" => println("actor receive==>start ....")
    case "id01" => println("actor receive==>id01")
    case "stop this" =>{
      println("actor receive==> stop this, stop....")
      // 关闭它自己,ActorSystem不关闭
      context.stop(self)
    }
    case "stop all" =>{
      println("actor receive==> stop all, stop....")
      // 关闭ActorSystem
      context.system.terminate()
    }
  }
}
object HelloAkka{
  def main(args: Array[String]): Unit = {
    // 创建ActorSystem对象
    val sys = ActorSystem("hello_sys")
    // 创建actor,并返回ActorRef对象
    val helloRef: ActorRef = sys.actorOf(Props[HelloAkkaActor], "hello")
    helloRef ! "start"
    helloRef ! "id01"
//    helloRef ! "stop this"
    helloRef ! "stop all"
  }
}

24.3.2 发送给本机的其它线程

file

package day05
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
class GirlActor extends Actor{
  override def receive: Receive = {
    case "踩我脚了" =>{
      println("GirlActor receive=> 踩我脚了, 思索片刻,回复 ‘谁啊’")
      Thread.sleep(1000)
      sender() ! "谁啊"
    }
    case "美女是你啊" =>{
      println("GirlActor receive=> 美女是你啊, 回复 ‘咋滴啦’")
      sender() ! "咋滴啦"
    }
  }
}
class BoyActor(girlRef:ActorRef) extends Actor{
  override def receive: Receive = {
    case "action" =>{
      println("BoyActor receive==> action, dialog start ....")
      girlRef ! "踩我脚了"
    }
    case "谁啊" =>{
      println("BoyActor receive==> 谁啊, send '美女是你啊'")
      girlRef ! "美女是你啊"
    }
    case "咋滴啦" =>{
      println("BoyActor receive==> 咋滴啦, send ‘踩我脚了’")
      girlRef ! "踩我脚了"
    }
  }
}
object DialogAkka {
  def main(args: Array[String]): Unit = {
    val sys = ActorSystem("dialog_sys")
    // 创建 GirlActor,并返回girl的ActorRef对象
    val girlRef: ActorRef = sys.actorOf(Props[GirlActor], "girl")
    // 创建 BoyActor , 并返回Boy的ActorRef对象
    val boyRef: ActorRef = sys.actorOf(Props[BoyActor](new BoyActor(girlRef)), "boy")
    boyRef ! "action"
  }
}

24.3.3 发送给不同的进程

file

**

远端actorRef设置参数:

akka.actor.provider = "akka.remote.RemoteActorRefProvider"

akka.remote.netty.tcp.hostname = $host

akka.remote.netty.tcp.port = $port

1) 创建一个 Server 端用于回复消息

server

package day05
import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
class ServerActor extends Actor{
  override def receive: Receive = {
    case "start" => println("ServerActor receive==> start, start....")
    case "马上开始发送计算任务" =>{
      println("ServerActor receive==> 马上开始发送计算任务")
    }
    case Client2ServerMsg(num1, symbol, num2) =>{
      println(s"ServerActor receive==> ${num1} ${symbol} ${num2}")
      var errCode = "000000"
      var errMsg = "成功"
      var result:Int = 0
      symbol match {
        case "+" => result = num1 + num2
        case "-" => result = num1 - num2
        case "*" => result = num1 * num2
        case _ =>{
          errCode = "err001"
          errMsg = "当前版本只支持+、-、*运算,待版本升级后尝试其他运算"
        }
      }
      val msg = Server2ClientMsg(errCode, errMsg, result)
      println(s"ServerActor send==>${msg}")
      // 返回数据(errCode, errMsg, result)
      sender() ! msg
    }
  }
}
object ServerActor{
  def main(args: Array[String]): Unit = {
    val host:String = "127.0.0.1"
    val port:Int = 8888
    // 解析配置参数
    val config:Config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = $host
         |akka.remote.netty.tcp.port = $port
      """.stripMargin
    )
    val sys = ActorSystem("server_sys", config)
    // akka地址:akka.tcp://server_sys@127.0.0.1:8888/user/server
    val serverRef = sys.actorOf(Props[ServerActor], "server")
    serverRef ! "start"
  }
}

启动后结果:

file

等待客户端发送消息,实现交互

2) 创建一个 Client 端发送消息

client

package day05
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
import scala.util.Random
class ClientActor(val serverHost:String, val serverPort:Int) extends Actor{
  var serverRef: ActorSelection = _
  override def preStart(): Unit = {
    // 通过akka地址获取到ServerActor的Ref对象,如果这个Actor对象不存在,该方法是不创建对象的
    serverRef = context.actorSelection(s"akka.tcp://server_sys@${serverHost}:${serverPort}/user/server")
  }
  override def receive: Receive = {
    case "start" => {
      println("ClientActor receive==> start, start....")
      println("ClientActor send==> 马上开始发送计算任务")
      serverRef ! "马上开始发送计算任务"
    }
    case SendClientMsg(data) =>{
      println(s"ClientActor receive==>${data}")
      val arr = data.split(" ")
      if(arr.size != 3){
        println(s"ClientActor receive data error, info: ${data}")
      }else{
        val num1 = arr(0).toInt
        val num2 = arr(2).toInt
        val symbol = arr(1)
        val msg = Client2ServerMsg(num1, symbol, num2)
        println(s"ClientActor send msg to Server==>${msg}")
        serverRef ! msg
      }
    }
    case msg:Server2ClientMsg =>{
      println(s"ClientActor from Server receive==>${msg}")
    }
  }
}
object ClientActor{
  def main(args: Array[String]): Unit = {
    val serverHost:String = "127.0.0.1"
    val serverPort:Int = 8888
    val host:String = "127.0.0.1"
    val port:Int = 8889
    // 解析配置参数
    val config:Config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = $host
         |akka.remote.netty.tcp.port = $port
      """.stripMargin
    )
    val sys = ActorSystem("client_sys", config)
    // akka地址:akka.tcp://client_sys@127.0.0.1:8889/user/client
    val clientRef = sys.actorOf(Props[ClientActor](new ClientActor(serverHost, serverPort)), "client")
    clientRef ! "start"
    // 随机生成 100 以内的计算
      val arr = Array[String]("+", "-", "*", "/")
      val random = new Random()
      while(true){
        val num1 = random.nextInt(100)
        val num2 = random.nextInt(100)
        val sysbol = arr(random.nextInt(arr.size))
        clientRef ! SendClientMsg(s"$num1 ${sysbol} $num2")
        Thread.sleep(1000)
    }
  }
}

case class 用于传输消息

// 本地发送给ClientActor消息
case class SendClientMsg(val data:String)
// ClientActor 发送给 ServerActor的消息
case class Client2ServerMsg(val num1:Int, val symbol:String, val num2:Int)
// ServerActor 把计算结果返回给ClientActor
case class Server2ClientMsg(val errCode:String, val errMsg:String, val result:Int)

运行效果:

clientActor

file

serverActor

file

24.4 打包上集群

1)修改代码参数

file

2)添加打包pom,确保资源文件目录配置与实际一致

    <build>
        <resources>
            <resource>
                <directory>src/main/resources</directory>
            </resource>
        </resources>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptors>
                        <descriptor>src/main/resources/assembly.xml</descriptor>
                    </descriptors>
                    <!--<archive>-->
                        <!--<manifest>-->
                            <!--<mainClass>${package.mainClass}</mainClass>-->
                        <!--</manifest>-->
                    <!--</archive>-->
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.12</version>
                <configuration>
                    <skip>true</skip>
                    <forkMode>once</forkMode>
                    <excludes>
                        <exclude>**/**</exclude>
                    </excludes>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

3)添加 assembly.xml

<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"    xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">  
  <id>hainiu</id>
  <formats>  
    <format>jar</format>
  </formats>  
  <includeBaseDirectory>false</includeBaseDirectory>
  <fileSets>
    <fileSet>
         <directory>${project.build.directory}/classes</directory>
         <outputDirectory>/</outputDirectory>
         <excludes>
            <exclude>*.xml</exclude>
            <exclude>*.properties</exclude>
         </excludes>
     </fileSet>
  </fileSets>  
  <dependencySets>
    <dependencySet>  
      <outputDirectory>/</outputDirectory>  
      <useProjectArtifact>false</useProjectArtifact>  
      <unpack>true</unpack>  
      <scope>runtime</scope>  
    </dependencySet>    
  </dependencySets> 
</assembly>

注意:

​ 如果网址有红线,Alt+Enter,选择 Ignored Schemas and DTDs(忽略)即可。到时候,运行打包时Maven会在Maven中央仓库中寻找最新版的assembly插件。

4)执行clean,然后执行 工程的rebuild

5)执行 assembly:assembly 打包

file

5)将冲突的配置进行合并,然后导入到导出的jar包中

file

手动合并成一个reference.conf,并添加到导出的jar 包里。

6)集群执行启动server Actor,启动本地client Actor,与集群的ServerActor 通信

file

服务端信息:

file

客户端信息:

file

24.5 用akka实现Wordcount(利用分布式思想)

定义MapperActor

  内部实现输出 单词,1

  要考虑到多个reduce分配数据的问题

定义ReducerActor

​ 内部实现接收多个MapperActor 发过来的数据,实现按照 单词进行聚合

package day05
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.io.Source
// 定义MapperActor
class MapperActor(val rRefs:ListBuffer[ActorRef]) extends Actor{
  override def receive: Receive = {
    case path:String =>{
      println(s"MapperActor ${self.path.name} receive==>${path}")
      val list = Source.fromFile(path).getLines().toList
      // 不带有combiner的
//      val tuples = list.flatMap(_.split("\t")).map((_,1))
      // 带有combiner的
      val tuples = list.flatMap(_.split("\t")).map((_,1)).groupBy(_._1).mapValues(_.size).toList
      // 一个一个发给reduce
      // 根据key的hash % reduce 个数来发数据到指定reduce
      for(t <- tuples){
        val key = t._1
        val value = t._2
        val rIndex = (key.hashCode & Integer.MAX_VALUE) % rRefs.size
        // 发送消息
        rRefs(rIndex) ! MapperSendReducerMsg(key, value)
      }
      // mapper数据处理完成, 通知所有ReducerActor发完了
      for(rRef <- rRefs){
        rRef ! MapperSendReduerEndMsg(self.path.name)
      }
    }
  }
}
// 定义ReducerActor
class ReducerActor(val mapMaxNum:Int) extends Actor{
  // 用来统计wordcount的结果
  val map = new mutable.HashMap[String,Int]
  // 用来装mapper name的set
  val set = new mutable.HashSet[String]
  override def receive: Receive = {
    case msg:MapperSendReducerMsg =>{
      println(s"ReducerActor ${self.path.name} receive==>${msg}")
      val option: Option[Int] = map.get(msg.key)
      if(option == None){
        // key还不存在,直接put
        map.put(msg.key, msg.value)
      }else{
        // key存在,把原来的value和本次的value加一起
        val lastValue = option.get
        map.put(msg.key, lastValue + msg.value)
      }
//      println(s"ReducerActor ${self.path.name} count==>${map}")
    }
    case MapperSendReduerEndMsg(mName) =>{
      set.add(mName)
      if(set.size == mapMaxNum){
        // 代表所有map的数据都已经发到reduce了, 输出最终结果
        println(s"ReducerActor ${self.path.name} count==>${map}")
      }
    }
  }
}
case class MapperSendReducerMsg(val key:String,val value:Int)
case class MapperSendReduerEndMsg(val mName:String)
object WordCountAkka {
  def main(args: Array[String]): Unit = {
    val files = Array("/tmp/scala/input/word1.txt",
      "/tmp/scala/input/word2.txt",
      "/tmp/scala/input/word3.txt",
      "/tmp/scala/input/word4.txt")
    val sys = ActorSystem("wordcount_sys")
    // 装reduceRef对象
    val rRefs = new ListBuffer[ActorRef]
    // 设定reducer个数
    val reducerNum = 5
    val mapMaxNum = files.size
    for(i <- 0 until reducerNum){
      val ref = sys.actorOf(Props[ReducerActor](new ReducerActor(mapMaxNum)), s"r${i}")
      rRefs += ref
    }
    // 根据有几个文件创建几个MapperActor对象
    for(i <- 0 until files.size){
      val ref = sys.actorOf(Props[MapperActor](new MapperActor(rRefs)), s"m${i}")
      ref ! files(i)
    }
  }
}

file

版权声明:原创作品,允许转载,转载时务必以超链接的形式表明出处和作者信息。否则将追究法律责任。来自海牛部落-潘牛,http://hainiubl.com/topics/75755
成为第一个点赞的人吧 :bowtie:
回复数量: 0
    暂无评论~~
    • 请注意单词拼写,以及中英文排版,参考此页
    • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`, 更多语法请见这里 Markdown 语法
    • 支持表情,可用Emoji的自动补全, 在输入的时候只需要 ":" 就可以自动提示了 :metal: :point_right: 表情列表 :star: :sparkles:
    • 上传图片, 支持拖拽和剪切板黏贴上传, 格式限制 - jpg, png, gif,教程
    • 发布框支持本地存储功能,会在内容变更时保存,「提交」按钮点击时清空
    Ctrl+Enter