scala 系列之 19scala 的 actor 并发编程

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

23 Actor 并发编程模型

23.1 Actor概述

为什么要学习actor?

我们现在学的Scala Actor是Scala 2.10.x版本及以前版本的Actor;

Scala在2.11.x版本中将Akka加入其中,作为默认的Actor,老版本的Actor已经废弃;

我们学习Actor的目的就是为了学习Akka做铺垫;

什么是Actor?

Actor是消息并发模型;

Scala中的Actor能够实现并行编程的强大功能,它是基于事件模型的并发机制。

Scala是运用消息(message)的发送、接收来实现多线程的。

使用Scala能够更容易地实现多线程应用的开发。

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

它每次只能处理一条消息,所以actor内部可以安全的处理状态,而不用考虑锁机制。

file

Java并发编程与Scala Actor编程的区别

Scala的Actor类似于Java中的多线程编程,但是不同的是,Scala的Actor提供的模型与多线程有所不同。

Scala的Actor尽可能地避免锁和共享状态,从而避免多线程并发时出现资源争用的情况。进而提升多线程编程的性能。

此外,Scala Actor的这种模型还可以避免死锁等一系列传统多线程编程的问题。

原因就在于Java中多数使用的是可变状态的对象资源,对这些资源进行共享来实现多线程编程的话,

控制好资源竞争与防止对象状态被意外修改是非常重要的,而对象状态的不变性也是较难以保证的。

而在Scala中,我们可以通过复制不可变状态的资源(即对象,Scala中一切都是对象,连函数、方法也是)的一个副本,

再基于Actor的消息发送、接收机制进行并行编程;

Actor方法执行顺序

1)调用 start() 方法启动 Actor;

2)执行 act() 方法;

3)向 Actor 发送消息;

发送消息的方式

! 发送异步消息,没有返回值;

!? 发送同步消息,等待返回值;

!! 发送异步消息,返回值是Future[Any];

23.2 actor 使用

scala 的 actor 在 scala 高版本被抛弃,如果想用,需要在工程中加入额外的 jar 包

file

23.2.1 actor 简单实现

package day05
import scala.actors.Actor
// 相当于继承 Thread
class ActorDemo extends Actor {
  // run方法体
  // 一个线程只执行一次
  override def act(): Unit = {
    println("一个线程只执行一次")
  }
}
object ActorDemo{
  def main(args: Array[String]): Unit = {
    val demo = new ActorDemo
    // 启动线程
    demo.start()
  }
}

23.2.2 actor通信

package day05
import scala.actors.{Actor, Future}
// 相当于继承 Thread
class ActorDemo extends Actor {
  // run方法体
  // 一个线程只执行一次
  override def act(): Unit = {
    println("一个线程只执行一次")
    while(true){
      receive{
        case "hello" => println("actor receive==> hello")
        case "id01" =>{
          println("actor receive==> id01, sleep 5s, return id02")
//          Thread.sleep(5000)
          // 回消息
          sender ! "id02"
        }
        case AsyncSendAndReturnMsg(id) => {
          println(s"actor receive==> ${id}, return id04")
          Thread.sleep(3000)
          sender ! AsyncSendAndReturnMsg("id04")
        }
      }
    }
  }
}
object ActorDemo{
  def main(args: Array[String]): Unit = {
    val demo = new ActorDemo
    // 启动线程
    demo.start()
    // 发送异步无返回消息
    demo ! "hello"
    // 发送同步有返回消息
    // 同步效率低
    val res1: Any = demo !? "id01"
    println(s"同步返回消息:${res1.asInstanceOf[String]}")
    // 异步效率高
    // 发送异步有返回消息
    val future: Future[Any] = demo !! AsyncSendAndReturnMsg("id03")
    println("继续干活,耗时4s,此时是能得到actor返回的结果")
    Thread.sleep(4000)

//    println("继续干活,耗时1s,得不到actor返回的结果")
//    Thread.sleep(1000)
    if(future.isSet){
      // future() 底层调用apply方法
      val res2: Any = future()
      val msg = res2.asInstanceOf[AsyncSendAndReturnMsg]
      println(msg)
    }else{
      println("future 没有等到结果")
    }
  }
}
// 用于封装发送和返回 异步消息
case class AsyncSendAndReturnMsg(val id:String)

23.2.3 用actor实现wordcount

file

package day05
import scala.actors.{Actor, Future}
import scala.collection.mutable.ListBuffer
import scala.io.Source
class WordCountActor extends Actor{
  override def act(): Unit = {
    receive{
      case path:String =>{
        println(s"actor receive==> path:${path}, start ...")
        val list = Source.fromFile(path).getLines().toList
        val map: Map[String, Int] = list.flatMap(_.split("\t")).map((_,1)).groupBy(_._1).mapValues(_.size)
        sender ! map
      }
    }
  }
}
object WordCountActor{
  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")
    // 用来存future的
    val futures = new ListBuffer[Future[Any]]
    // 用来装从future里提取出来的map数据
    val maps = new ListBuffer[Map[String,Int]]
    // 为每个文件创建一个actor对象处理wordcount
    for(f <- files){
      val actor = new WordCountActor
      actor.start()
      val future: Future[Any] = actor !! f
      futures += future
    }
    // 循环判断futures列表中的Futrue是否有数据返回,如果有,提取里面的数据
    while(! futures.isEmpty){
      val hasDataFutures = futures.filter(_.isSet)
      for(hasDataFutrue <- hasDataFutures){
        // 提取数据
        val data: Any = hasDataFutrue()
        val map = data.asInstanceOf[Map[String,Int]]
        println(s"局部map:${map}")
        maps += map
        // 从future列表里,把已经提取出数据的Futrue干掉
        futures -= hasDataFutrue
      }
    }
    // 把多个结果的map合并成一个
    val list1: ListBuffer[(String, Int)] = maps.flatten
    val resMap: Map[String, Int] = list1.groupBy(_._1).mapValues(_.map(_._2).sum)
    val resList: List[(String, Int)] = resMap.toList.sortBy(_._2).reverse
    println(s"resList: ${resList}")
  }
}

运行结果:

file

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