Scala第十九章节

Scala第十九章节

scala总目录
文档资料下载

章节目标

  1. 了解Actor的相关概述
  2. 掌握Actor发送和接收消息
  3. 掌握WordCount案例

1. Actor介绍

Scala中的Actor并发编程模型可以用来开发比Java线程效率更高的并发程序。我们学习Scala Actor的目的主要是为后续学习Akka做准备。

1.1 Java并发编程的问题

在Java并发编程中，每个对象都有一个逻辑监视器（monitor），可以用来控制对象的多线程访问。我们添加sychronized关键字来标记，需要进行同步加锁访问。这样，通过加锁的机制来确保同一时间只有一个线程访问共享数据。但这种方式存在资源争夺、以及死锁问题，程序越大问题越麻烦。
在这里插入图片描述

线程死锁
在这里插入图片描述

1.2 Actor并发编程模型

Actor并发编程模型，是Scala提供给程序员的一种与Java并发编程完全不一样的并发编程模型，是一种基于事件模型的并发机制。Actor并发编程模型是一种不共享数据，依赖消息传递的一种并发编程模式，有效避免资源争夺、死锁等情况。
在这里插入图片描述

1.3 Java并发编程对比Actor并发编程
Java内置线程模型Scala Actor模型
"共享数据-锁"模型 (share data and lock)share nothing
每个object有一个monitor，监视线程对共享数据的访问不共享数据，Actor之间通过Message通讯
加锁代码使用synchronized标识
死锁问题
每个线程内部是顺序执行的每个Actor内部是顺序执行的

注意:

  1. scala在2.11.x版本中加入了Akka并发编程框架，老版本已经废弃。

  2. Actor的编程模型和Akka很像，我们这里学习Actor的目的是为学习Akka做准备。

2. 创建Actor

我们可以通过类(class)或者单例对象(object), 继承Actor特质的方式, 来创建Actor对象.

2.1 步骤
  1. 定义class或object继承Actor特质
  2. 重写act方法
  3. 调用Actor的start方法执行Actor

注意: 每个Actor是并行执行的, 互不干扰.

2.2 案例一: 通过class实现

需求

  1. 创建两个Actor，一个Actor打印1-10，另一个Actor打印11-20
  2. 使用class继承Actor实现.（如果需要在程序中创建多个相同的Actor）

参考代码

import scala.actors.Actor//案例:Actor并发编程入门, 通过class创建Actor
object ClassDemo01 {//需求: 创建两个Actor，一个Actor打印1-10，另一个Actor打印11-20//1. 创建Actor1, 用来打印1~10的数字.class Actor1 extends Actor {override def act(): Unit = for (i <- 1 to 10) println("actor1: " + i)}//2. 创建Actor2, 用来打印11~20的数字.class Actor2 extends Actor {override def act(): Unit = for (i <- 11 to 20) println("actor2: " + i)}def main(args: Array[String]): Unit = {//3. 启动两个Actor.new Actor1().start()new Actor2().start()}
}
2.3 案例二: 通过object实现

需求

  1. 创建两个Actor，一个Actor打印1-10，另一个Actor打印11-20
  2. 使用object继承Actor实现.（如果在程序中只创建一个Actor）

参考代码

import scala.actors.Actor//案例:Actor并发编程入门, 通过object创建Actor
object ClassDemo02 {//需求: 创建两个Actor，一个Actor打印1-10，另一个Actor打印11-20//1. 创建Actor1, 用来打印1~10的数字.object Actor1 extends Actor {override def act(): Unit = for (i <- 1 to 10) println("actor1: " + i)}//2. 创建Actor2, 用来打印11~20的数字.object Actor2 extends Actor {override def act(): Unit = for (i <- 11 to 20) println("actor2: " + i)}def main(args: Array[String]): Unit = {//3. 启动两个Actor.Actor1.start()Actor2.start()}
}
2.4 Actor程序运行流程
  1. 调用start()方法启动Actor
  2. 自动执行act()方法
  3. 向Actor发送消息
  4. act方法执行完成后，程序会调用**exit()**方法结束程序执行.

3. 发送消息/接收消息

我们之前介绍Actor的时候，说过Actor是基于事件（消息）的并发编程模型，那么Actor是如何发送消息和接收消息的呢？

3.1 使用方式
3.1.1 发送消息

我们可以使用三种方式来发送消息：

！发送异步消息，没有返回值
!?发送同步消息，等待返回值
!!发送异步消息，返回值是Future[Any]

例如：要给actor1发送一个异步字符串消息，使用以下代码：

actor1 ! "你好!"
3.1.2 接收消息

Actor中使用receive方法来接收消息，需要给receive方法传入一个偏函数

{case 变量名1:消息类型1 => 业务处理1case 变量名2:消息类型2 => 业务处理2...
}

注意: receive方法只接收一次消息，接收完后继续执行act方法

3.2 案例一: 发送及接收一句话

需求

  1. 创建两个Actor（ActorSender、ActorReceiver）
  2. ActorSender发送一个异步字符串消息给ActorReceiver
  3. ActorReceiver接收到该消息后，打印出来

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

参考代码

//案例: 采用 异步无返回的形式, 发送消息.
object ClassDemo03 {//1. 创建发送消息的Actor, ActorSender, 发送一句话给ActorReceiverobject ActorSender extends Actor {override def act(): Unit = {//发送一句话给ActorReceiverActorReceiver ! "你好啊, 我是ActorSender!"//发送第二句话ActorReceiver ! "你叫什么名字呀? "}}//2. 创建接收消息的Actor, ActorReceiverobject ActorReceiver extends Actor {override def act(): Unit = {//接收发送过来的消息.receive {case x: String => println(x)}}}def main(args: Array[String]): Unit = {//3. 启动两个ActorActorSender.start()ActorReceiver.start()}
}
3.3 案例二: 持续发送和接收消息

如果我们想实现ActorSender一直发送消息, ActorReceiver能够一直接收消息，该怎么实现呢？

答: 我们只需要使用一个while(true)循环，不停地调用receive来接收消息就可以啦。

需求

  1. 创建两个Actor（ActorSender、ActorReceiver）
  2. ActorSender持续发送一个异步字符串消息给ActorReceiver
  3. ActorReceiver持续接收消息，并打印出来

参考代码

//案例:Actor 持续发送和接收消息.
object ClassDemo04 {//1. 创建发送消息的Actor, ActorSender, 发送一句话给ActorReceiverobject ActorSender extends Actor {override def act(): Unit = {while(true) {//发送一句话给ActorReceiverActorReceiver ! "你好啊, 我是ActorSender!"//休眠3秒.TimeUnit.SECONDS.sleep(3)       //单位是: 秒}}}//2. 创建接收消息的Actor, ActorReceiverobject ActorReceiver extends Actor {override def act(): Unit = {//接收发送过来的消息,  持续接收.while(true) {receive {case x: String => println(x)}}}}def main(args: Array[String]): Unit = {//3. 启动两个ActorActorSender.start()ActorReceiver.start()}
}
3.4 案例三: 优化持续接收消息

上述代码，是用while循环来不断接收消息的, 这样做可能会遇到如下问题:

  • 如果当前Actor没有接收到消息，线程就会处于阻塞状态
  • 如果有很多的Actor，就有可能会导致很多线程都是处于阻塞状态
  • 每次有新的消息来时，重新创建线程来处理
  • 频繁的线程创建、销毁和切换，会影响运行效率

针对上述情况, 我们可以使用loop(), 结合react()来复用线程, 这种方式比while循环 + receive()更高效.

需求

  1. 创建两个Actor（ActorSender、ActorReceiver）
  2. ActorSender持续发送一个异步字符串消息给ActorReceiver
  3. ActorReceiver持续接收消息，并打印出来

注意: 使用loop + react重写上述案例.

参考代码

//案例: 使用loop + react循环接收消息.
object ClassDemo05 {//1. 创建发送消息的Actor, ActorSender, 发送一句话给ActorReceiverobject ActorSender extends Actor {override def act(): Unit = {while(true) {//发送一句话给ActorReceiverActorReceiver ! "你好啊, 我是ActorSender!"//休眠3秒.TimeUnit.SECONDS.sleep(3)       //单位是: 秒}}}//2. 创建接收消息的Actor, ActorReceiverobject ActorReceiver extends Actor {override def act(): Unit = {//接收发送过来的消息,  持续接收.loop{react {case x: String => println(x)}}}}def main(args: Array[String]): Unit = {//3. 启动两个ActorActorSender.start()ActorReceiver.start()}
}
3.5 案例四: 发送和接收自定义消息

我们前面发送的消息都是字符串类型，Actor中也支持发送自定义消息，例如：使用样例类封装消息，然后进行发送处理。

3.5.1 示例一: 发送同步有返回消息

需求

  1. 创建一个MsgActor，并向它发送一个同步消息，该消息包含两个字段（id、message）
  2. MsgActor回复一个消息，该消息包含两个字段（message、name）
  3. 打印回复消息

注意:

  • 使用!?来发送同步消息
  • 在Actor的act方法中，可以使用sender获取发送者的Actor引用

参考代码

//案例: Actor发送和接收自定义消息, 采用 同步有返回的形式
object ClassDemo06 {//1. 定义两个样例类Message(表示发送数据),   ReplyMessage(表示返回数据.)case class Message(id: Int, message: String) //自定义的发送消息 样例类case class ReplyMessage(message: String, name: String) //自定义的接收消息 样例类//2. 创建一个MsgActor，用来接收MainActor发送过来的消息, 并向它回复一条消息.object MsgActor extends Actor {override def act(): Unit = {//2.1 接收 主Actor(MainActor) 发送过来的消息.loop {react {//结合偏函数使用case Message(id, message) => println(s"我是MsgActor, 我收到的消息是: ${id}, ${message}")//2.2 给MainActor回复一条消息.//sender: 获取消息发送方的Actor对象sender ! ReplyMessage("我很不好, 熏死了!...", "车磊")}}}}def main(args: Array[String]): Unit = {//3. 开启MsgActorMsgActor.start()//4. 通过MainActor, 给MsgActor发送一个 Message对象.//采用 !?  同步有返回.val reply:Any = MsgActor !? Message(1, "你好啊, 我是MainActor, 我在给你发消息!")//resutl表示最终接收到的 返回消息.val result = reply.asInstanceOf[ReplyMessage]//5. 输出结果.println(result)}
}
3.5.2 示例二: 发送异步无返回消息

需求

创建一个MsgActor，并向它发送一个异步无返回消息，该消息包含两个字段（id, message）

注意: 使用!发送异步无返回消息

参考代码

//案例: Actor发送和接收自定义消息, 采用 异步 无返回的形式
object ClassDemo07 {//1. 定义一个样例类Message(表示发送数据)case class Message(id: Int, message: String) //自定义的发送消息 样例类//2. 创建一个MsgActor，用来接收MainActor发送过来的消息, 并打印.object MsgActor extends Actor {override def act(): Unit = {//2.1 接收 主Actor(MainActor) 发送过来的消息.loop {react {//结合偏函数使用case Message(id, message) => println(s"我是MsgActor, 我收到的消息是: ${id}, ${message}")}}}}def main(args: Array[String]): Unit = {//3. 开启MsgActorMsgActor.start()//4. 通过MainActor, 给MsgActor发送一个 Message对象.//采用 !  异步无返回MsgActor ! Message(1, "我是采用 异步无返回 的形式发送消息!")}
}
3.5.3 示例三: 发送异步有返回消息

需求

  1. 创建一个MsgActor，并向它发送一个异步有返回消息，该消息包含两个字段（id、message）
  2. MsgActor回复一个消息，该消息包含两个字段（message、name）
  3. 打印回复消息

注意:

  • 使用!!发送异步有返回消息
  • 发送后，返回类型为Future[Any]的对象
  • Future表示异步返回数据的封装，虽获取到Future的返回值，但不一定有值，可能在将来某一时刻才会返回消息
  • Future的isSet()可检查是否已经收到返回消息，apply()方法可获取返回数据

图解
在这里插入图片描述

参考代码

//案例: Actor发送和接收自定义消息, 采用 异步有返回的形式
object ClassDemo08 {//1. 定义两个样例类Message(表示发送数据),   ReplyMessage(表示返回数据.)case class Message(id: Int, message: String) //自定义的发送消息 样例类case class ReplyMessage(message: String, name: String) //自定义的接收消息 样例类//2. 创建一个MsgActor，用来接收MainActor发送过来的消息, 并向它回复一条消息.object MsgActor extends Actor {override def act(): Unit = {//2.1 接收 主Actor(MainActor) 发送过来的消息.loop {react {//结合偏函数使用case Message(id, message) => println(s"我是MsgActor, 我收到的消息是: ${id}, ${message}")//2.2 给MainActor回复一条消息.//sender: 获取消息发送方的Actor对象sender ! ReplyMessage("我很不好, 熏死了!...", "糖糖")}}}}def main(args: Array[String]): Unit = {//3. 开启MsgActorMsgActor.start()//4. 通过MainActor, 给MsgActor发送一个 Message对象.//采用 !!  异步有返回.val future: Future[Any] = MsgActor !! Message(1, "你好啊, 我是MainActor, 我在给你发消息!")//5. 因为future中不一定会立马有数据, 所以我们要校验.//Future的isSet()可检查是否已经收到返回消息，apply()方法可获取返回数据//!future.isSet表示: 没有接收到具体的返回消息, 就一直死循环.while(!future.isSet){}//通过Future的apply()方法来获取返回的数据.val result = future.apply().asInstanceOf[ReplyMessage]//5. 输出结果.println(result)}
}

4. 案例: WordCount

4.1 需求

接下来，我们要使用Actor并发编程模型实现多文件的单词统计。

案例介绍

给定几个文本文件（文本文件都是以空格分隔的），使用Actor并发编程来统计单词的数量.

思路分析
在这里插入图片描述

实现思路

  1. MainActor获取要进行单词统计的文件
  2. 根据文件数量创建对应的WordCountActor
  3. 将文件名封装为消息发送给WordCountActor
  4. WordCountActor接收消息，并统计单个文件的单词计数
  5. 将单词计数结果发送给MainActor
  6. MainActor等待所有的WordCountActor都已经成功返回消息，然后进行结果合并
4.2 步骤一: 获取文件列表

实现思路

  1. 在当前项目下的data文件夹下有: 1.txt, 2.txt两个文本文件, 具体存储内容如下:

    1.txt文本文件存储内容如下:

    hadoop sqoop hadoop
hadoop hadoop flume
hadoop hadoop hadoop
spark

    2.txt文本文件存储内容如下:

    flink hadoop hive
hadoop sqoop hadoop
hadoop hadoop hadoop
spark

  2. 获取上述两个文本文件的路径, 并将结果打印到控制台上.

参考代码

object MainActor {def main(args: Array[String]): Unit = {//1. 获取所有要统计的文件的路径.//1.1 定义变量dir, 记录保存所有文件的: 文件夹路径.  ./data/var dir = "./data/"//1.2 获取该文件夹下, 所有的文件名.var fileNameList = new File(dir).list().toList //List("1.txt", "2.txt")//1.3 对获取到的文件名进行封装, 获取其全路径.     ./data/1.txt      ./data/2.txtvar fileDirList = fileNameList.map(dir + _)//println(fileDirList)}
}
4.3 步骤二: 创建WordCountActor

实现思路

  1. 根据文件数量创建对应个数的WordCountActor对象.
  2. 为了方便后续发送消息给Actor，将每个Actor与文件名关联在一起

实现步骤

  1. 创建WordCountActor
  2. 将文件列表转换为WordCountActor
  3. 为了后续方便发送消息给Actor，将Actor列表和文件列表拉链到一起
  4. 打印测试

参考代码

  • WordCountActor.scala文件中的代码

    //2.1 先创建WordCountActor类, 用来获取WordCountActor对象.
//创建WordCountActor类, 每一个WordCountActor对象, 统计一个文件.
class WordCountActor extends Actor {override def act(): Unit = { }
}

  • MainActor.scala文件中的代码

    object MainActor {def main(args: Array[String]): Unit = {//1. 获取所有要统计的文件的路径.//1.1 定义变量dir, 记录保存所有文件的: 文件夹路径.  ./data/var dir = "./data/"//1.2 获取该文件夹下, 所有的文件名.var fileNameList = new File(dir).list().toList //List("1.txt", "2.txt")//1.3 对获取到的文件名进行封装, 获取其全路径.     ./data/1.txt      ./data/2.txtvar fileDirList = fileNameList.map(dir + _)//println(fileDirList)//2. 根据文件数量, 创建对应的WordCountActor对象.//2.1 先创建WordCountActor类, 用来获取WordCountActor对象.//2.2 根据文件数量, 创建对应的WordCountActor对象.val wordCountList = fileNameList.map(_ => new WordCountActor) //根据两个txt文件, 创建了两个wordCount对象.//println(wordCountList)//2.3 将WordCountActor和文件全路径关联起来val actorWithFile = wordCountList.zip(fileDirList) //WordCountActor -> ./data/1.txt ,  WordCountActor -> ./data/2.txtprintln(actorWithFile)}
}
4.4 步骤三: 启动Actor/发送/接收任务消息

实现思路

启动所有WordCountActor对象，并发送单词统计任务消息给每个WordCountActor对象.

注意: 此处应发送异步有返回消息

实现步骤

  1. 创建一个WordCountTask样例类消息，封装要进行单词计数的文件名
  2. 启动所有WordCountActor，并发送异步有返回消息
  3. 获取到所有的WordCountActor中返回的消息（封装到一个Future列表中）
  4. 在WordCountActor中接收并打印消息

参考代码

  • MessagePackage.scala文件中的代码

    /*** 表示: MainActor 给每一个WordCountActor发送任务的 格式.* @param fileName 具体的要统计的 文件路径.*/
case class WordCountTask(fileName:String)

  • MainActor.scala文件中的代码

    object MainActor {def main(args: Array[String]): Unit = {//1. 获取所有要统计的文件的路径.//1.1 定义变量dir, 记录保存所有文件的: 文件夹路径.  ./data/var dir = "./data/"//1.2 获取该文件夹下, 所有的文件名.var fileNameList = new File(dir).list().toList //List("1.txt", "2.txt")//1.3 对获取到的文件名进行封装, 获取其全路径.     ./data/1.txt      ./data/2.txtvar fileDirList = fileNameList.map(dir + _)//println(fileDirList)//2. 根据文件数量, 创建对应的WordCountActor对象.//2.1 先创建WordCountActor类, 用来获取WordCountActor对象.//2.2 根据文件数量, 创建对应的WordCountActor对象.val wordCountList = fileNameList.map(_ => new WordCountActor) //根据两个txt文件, 创建了两个wordCount对象.//println(wordCountList)//2.3 将WordCountActor和文件全路径关联起来val actorWithFile = wordCountList.zip(fileDirList) //WordCountActor -> ./data/1.txt ,  WordCountActor -> ./data/2.txtprintln(actorWithFile)//3. 启动WordCountActor, 并给每一个WordCountActor发送任务./*Map(spark -> 1, hadoop -> 7, sqoop -> 1, flume -> 1)Map(sqoop -> 1, flink -> 1, hadoop -> 6, spark -> 1, hive -> 1)*/val futureList: List[Future[Any]] = actorWithFile.map {       //futureList: 记录的是所有WordCountActor统计的结果.keyVal => //keyVal的格式: WordCountActor -> ./data/1.txt//3.1 获取具体的要启动的WordCountActor对象.val actor = keyVal._1 //actor: WordCountActor//3.2 启动具体的WordCountActor.actor.start()//3.3 给每个WordCountActor发送具体的任务(文件路径)  异步有返回.val future: Future[Any] = actor !! WordCountTask(keyVal._2)future      //记录的是某一个WordCountActor返回的统计结果.}}
}

  • WordCountActor.scala文件中的代码

    //2.1 先创建WordCountActor类, 用来获取WordCountActor对象.
//创建WordCountActor类, 每一个WordCountActor对象, 统计一个文件.
class WordCountActor extends Actor {override def act(): Unit = { loop {react {//3.4 接收具体的任务case WordCountTask(fileName) =>//3.5 打印具体的任务println(s"接收到的具体任务是: ${fileName}")}}}
}
4.5 步骤四: 统计文件单词计数

实现思路

读取文件文本，并统计出来单词的数量。例如：

(hadoop, 3), (spark, 1)...

实现步骤

  1. 读取文件内容，并转换为列表
  2. 按照空格切割文本，并转换为一个一个的单词
  3. 为了方便进行计数，将单词转换为元组
  4. 按照单词进行分组，然后再进行聚合统计
  5. 打印聚合统计结果

参考代码

  • WordCountActor.scala文件中的代码

    class WordCountActor extends Actor {override def act(): Unit = {//采用loop + react 方式接收数据.loop {react {//3.4 接收具体的任务case WordCountTask(fileName) =>//3.5 打印具体的任务println(s"接收到的具体任务是: ${fileName}")//4. 统计接收到的文件中的每个单词的数量.//4.1 获取指定文件中的所有的文件. List("hadoop sqoop hadoop","hadoop hadoop flume")val lineList = Source.fromFile(fileName).getLines().toList//4.2 将上述获取到的数据, 转换成一个一个的字符串.  //List("hadoop", "sqoop", "hadoop","hadoop", "hadoop", "flume")val strList = lineList.flatMap(_.split(" "))//4.3 给每一个字符串后边都加上次数, 默认为1.             //List("hadoop"->1, "sqoop"->1, "hadoop"->1, "hadoop"->1, "flume"->1)val wordAndCount = strList.map(_ -> 1)//4.4 按照 字符串内容分组.                              //"hadoop" -> List("hadoop"->1, "hadoop"->1),   "sqoop" -> List("sqoop"->1)val groupMap = wordAndCount.groupBy(_._1)//4.5 对分组后的内容进行统计, 统计每个单词的总次数.    "hadoop" -> 2,   "sqoop" -> 1val wordCountMap = groupMap.map(keyVal => keyVal._1 -> keyVal._2.map(_._2).sum)//4.6 打印统计后的结果. println(wordCountMap)}}}
}
4.6 步骤五: 返回结果给MainActor

实现思路

  • 将单词计数的结果封装为一个样例类消息，并发送给MainActor
  • MainActor等待所有WordCountActor均已返回后,获取到每个WordCountActor单词计算后的结果

实现步骤

  1. 定义一个样例类封装单词计数结果
  2. 将单词计数结果发送给MainActor
  3. MainActor中检测所有WordCountActor是否均已返回，如果均已返回，则获取并转换结果
  4. 打印结果

参考代码

  • MessagePackage.scala文件中的代码

    /*** 表示: MainActor 给每一个WordCountActor发送任务的 格式.* @param fileName 具体的要统计的 文件路径.*/
case class WordCountTask(fileName:String)/*** 每个WordCountActor统计完的返回结果的: 格式* @param wordCountMap  具体的返回结果, 例如:  Map("hadoop"->6, "sqoop"->1)*/
case class WordCountResult(wordCountMap:Map[String, Int])

  • WordCountActor.scala文件中的代码

    class WordCountActor extends Actor {override def act(): Unit = {//采用loop + react 方式接收数据.loop {react {//3.4 接收具体的任务case WordCountTask(fileName) =>//3.5 打印具体的任务println(s"接收到的具体任务是: ${fileName}")//4. 统计接收到的文件中的每个单词的数量.//4.1 获取指定文件中的所有的文件.                       List("hadoop sqoop hadoop","hadoop hadoop flume")val lineList = Source.fromFile(fileName).getLines().toList//4.2 将上述获取到的数据, 转换成一个一个的字符串.        List("hadoop", "sqoop", "hadoop","hadoop", "hadoop", "flume")val strList = lineList.flatMap(_.split(" "))//4.3 给每一个字符串后边都加上次数, 默认为1.             List("hadoop"->1, "sqoop"->1, "hadoop"->1,"hadoop"->1, "hadoop"->1, "flume"->1)val wordAndCount = strList.map(_ -> 1)//4.4 按照 字符串内容分组.                              "hadoop" -> List("hadoop"->1, "hadoop"->1),   "sqoop" -> List("sqoop"->1)val groupMap = wordAndCount.groupBy(_._1)//4.5 对分组后的内容进行统计, 统计每个单词的总次数.      "hadoop" -> 2,   "sqoop" -> 1val wordCountMap = groupMap.map(keyVal => keyVal._1 -> keyVal._2.map(_._2).sum)//4.6 把统计后的结果返回给: MainActor.sender ! WordCountResult(wordCountMap)}}}
}
4.7 步骤六: 结果合并

实现思路

对接收到的所有单词计数进行合并。

参考代码

  • MainActor.scala文件中的代码

    object MainActor {def main(args: Array[String]): Unit = {//1. 获取所有要统计的文件的路径.//1.1 定义变量dir, 记录保存所有文件的: 文件夹路径.  ./data/var dir = "./data/"//1.2 获取该文件夹下, 所有的文件名.var fileNameList = new File(dir).list().toList //List("1.txt", "2.txt")//1.3 对获取到的文件名进行封装, 获取其全路径.     ./data/1.txt      ./data/2.txtvar fileDirList = fileNameList.map(dir + _)//println(fileDirList)//2. 根据文件数量, 创建对应的WordCountActor对象.//2.1 先创建WordCountActor类, 用来获取WordCountActor对象.//2.2 根据文件数量, 创建对应的WordCountActor对象.val wordCountList = fileNameList.map(_ => new WordCountActor) //根据两个txt文件, 创建了两个wordCount对象.//println(wordCountList)//2.3 将WordCountActor和文件全路径关联起来val actorWithFile = wordCountList.zip(fileDirList) //WordCountActor -> ./data/1.txt ,  WordCountActor -> ./data/2.txtprintln(actorWithFile)//3. 启动WordCountActor, 并给每一个WordCountActor发送任务./*Map(spark -> 1, hadoop -> 7, sqoop -> 1, flume -> 1)Map(sqoop -> 1, flink -> 1, hadoop -> 6, spark -> 1, hive -> 1)*/val futureList: List[Future[Any]] = actorWithFile.map {       //futureList: 记录的是所有WordCountActor统计的结果.keyVal => //keyVal的格式: WordCountActor -> ./data/1.txt//3.1 获取具体的要启动的WordCountActor对象.val actor = keyVal._1 //actor: WordCountActor//3.2 启动具体的WordCountActor.actor.start()//3.3 给每个WordCountActor发送具体的任务(文件路径)  异步有返回.val future: Future[Any] = actor !! WordCountTask(keyVal._2)future      //记录的是某一个WordCountActor返回的统计结果.}//5. MainActor对接收到的数据进行合并.//5.1 判断所有的future都有返回值后, 再往下执行.//       过滤没有返回值的future         不为0说明还有future没有收到值while(futureList.filter(!_.isSet).size != 0) {} //futureList:  future1, future2//5.2 从每一个future中获取数据.//wordCountMap:  List(Map(spark -> 1, hadoop -> 7, sqoop -> 1, flume -> 1), Map(sqoop -> 1, flink -> 1, hadoop -> 6, spark -> 1, hive -> 1))val wordCountMap = futureList.map(_.apply().asInstanceOf[WordCountResult].wordCountMap)//5.3 对获取的数据进行flatten, groupBy, map, 然后统计.val result = wordCountMap.flatten.groupBy(_._1).map(keyVal => keyVal._1 -> keyVal._2.map(_._2).sum)//5.4 打印结果println(result)}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/151339.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

LabVIEW开发教学实验室自动化INL和DNL测试系统

LabVIEW开发教学实验室自动化INL和DNL测试系统

LabVIEW开发教学实验室自动化INL和DNL测试系统 如今&#xff0c;几乎所有的测量仪器都是基于微处理器的设备。模拟输入量在进行数字处理之前被转换为数字量。对于参加电气和电子测量课程的学生来说&#xff0c;了解ADC以及如何欣赏其性能至关重要。ADC的不确定性可以根据其传输…
阅读更多...
Unity Golang教程-Shader编写一个流动的云效果

Unity Golang教程-Shader编写一个流动的云效果

创建目录 一个友好的项目&#xff0c;项目目录结构是很重要的。我们先导入一个登录界面模型资源。 我们先创建Art表示是美术类的资源&#xff0c;资源是模型创建Model文件夹&#xff0c;由于是在登录界面所以创建Login文件夹&#xff0c;下面依次是模型对应的资源&#xff0c…
阅读更多...
世界前沿技术发展报告2023《世界信息技术发展报告》（六）网络与通信技术

世界前沿技术发展报告2023《世界信息技术发展报告》（六）网络与通信技术

&#xff08;六&#xff09;网络与通信技术 1. 概述2. 5G与光通讯2.1 美国研究人员利用电磁拓扑绝缘体使5G频谱带宽翻倍2.2 日本东京工业大学推出可接入5G网络的高频收发器2.3 美国得克萨斯农工大学通过波束管理改进5G毫米波通信2.4 联发科完成全球首次5G NTN卫星手机连线测试2…
阅读更多...
自动定时删除磁盘文件的脚本（从文件日期最早的开始删）

自动定时删除磁盘文件的脚本（从文件日期最早的开始删）

#!/bin/bash# 指定的挂载点 MOUNTPOINT"/media/vm/MyDisk512GB"# 设置磁盘大小的限制 (例如&#xff1a;800G) LIMIT$((800 * 1024 * 1024)) # 单位是KB# 获取挂载点的已使用空间 USED_SPACE$(df -kP "$MOUNTPOINT" | tail -1 | awk {print $3})echo &quo…
阅读更多...
【Oracle】Oracle系列十九--Oracle的体系结构

【Oracle】Oracle系列十九--Oracle的体系结构

文章目录 往期回顾前言1. 物理结构2. 内存结构2.1 SGA2.2 后台进程 3. 逻辑结构 往期回顾 【Oracle】Oracle系列之一–Oracle数据类型 【Oracle】Oracle系列之二–Oracle数据字典 【Oracle】Oracle系列之三–Oracle字符集 【Oracle】Oracle系列之四–用户管理 【Oracle】Or…
阅读更多...
应用案例 | dataFEED OPC Suite为化工行业中的质量控制和成本节约提供数据集成方案

应用案例 | dataFEED OPC Suite为化工行业中的质量控制和成本节约提供数据集成方案

一 背景 在当今化工行业中&#xff0c;质量控制对于特种塑料供应商至关重要。一家国际性的特种塑料供应商在全球拥有五个生产基地&#xff0c;每个基地都运行着2-6台塑料挤出机。为了确保塑料质量&#xff0c;他们需要每两小时分析一次挤出样品——导致这项工作占用了较大的生…
阅读更多...
Bigemap是如何在生态林业科技行业去应用的

Bigemap是如何在生态林业科技行业去应用的

选择Bigemap的原因&#xff1a; ①之前一直是使用的谷歌地球&#xff0c;现在谷歌不能使用了就在网上搜索找一款可以替代的软件&#xff0c;工作使用需求还是挺大的&#xff0c;谷歌不能用对工作进展也非常影响&#xff0c;在网上搜索到软件大部分功能都可以满足需求 ②软件卫…
阅读更多...
Tauri | 新版2.0路线图：更强大的插件以及支持 iOS、Android 应用构建

Tauri | 新版2.0路线图：更强大的插件以及支持 iOS、Android 应用构建

Tauri官方在9月7号发布了新版2.0的路线图&#xff0c;该版本主要是对移动端进行升级&#xff0c;主要特性如下&#xff1a; 强大的插件系统&#xff0c;官方把常用的功能进行了插件化&#xff08;见下图&#xff09;支持使用 Swift、Kotlin 编程语言开发插件&#xff0c;对 iO…
阅读更多...
Nginx配置文件的通用语法介绍

Nginx配置文件的通用语法介绍

要是参考《Ubuntu 20.04使用源码安装nginx 1.14.0》安装nginx的话&#xff0c;nginx配置文件在/nginx/conf目录里边&#xff0c;/nginx/conf里边的配置文件结构如下图所示&#xff1a; nginx.conf是主配置文件&#xff0c;它是一个ascii文本文件。配置文件由指令&#xff08;…
阅读更多...
【Vue面试题九】、Vue中给对象添加新属性界面不刷新？

【Vue面试题九】、Vue中给对象添加新属性界面不刷新？

文章底部有个人公众号&#xff1a;热爱技术的小郑。主要分享开发知识、学习资料、毕业设计指导等。有兴趣的可以关注一下。为何分享&#xff1f; 踩过的坑没必要让别人在再踩&#xff0c;自己复盘也能加深记忆。利己利人、所谓双赢。 面试官&#xff1a;动态给vue的data添加一个…
阅读更多...
C# OpenCvSharp Yolov8 Pose 姿态识别

C# OpenCvSharp Yolov8 Pose 姿态识别

效果 项目 代码 using OpenCvSharp; using OpenCvSharp.Dnn; using System; using System.Collections.Generic; using System.ComponentModel; using System.Data; using System.Drawing; using System.Linq; using System.Text; using System.Windows.Forms;namespace OpenC…
阅读更多...
【数据结构--八大排序】之希尔排序

【数据结构--八大排序】之希尔排序

&#x1f490; &#x1f338; &#x1f337; &#x1f340; &#x1f339; &#x1f33b; &#x1f33a; &#x1f341; &#x1f343; &#x1f342; &#x1f33f; &#x1f344;&#x1f35d; &#x1f35b; &#x1f364; &#x1f4c3;个人主页 &#xff1a;阿然成长日记 …
阅读更多...
C++简单上手helloworld 以及 vscode找不到文件的可能性原因

C++简单上手helloworld 以及 vscode找不到文件的可能性原因

helloworld #include <iostream>int main() {std::cout << "hello world!" << std::endl;return 0; }输入输出小功能 #include <iostream> using namespace std; /* *主函数 *输出一条语句 */int main() {// 输出一条语句cout << &q…
阅读更多...
javaWeb宠物领养系统

javaWeb宠物领养系统

一、引言 1.1 系统背景 计算机网络的发展&#xff0c;促进了社会各行业的进步&#xff0c;带来了经济快速增长。管理员通过流浪宠物的信息&#xff0c;在平台上和领养人进行实时的交流&#xff0c;达成领养协议。用户登录后&#xff0c;把想要领养的宠物向本平台发起申请&…
阅读更多...
Python3操作Redis最新版|CRUD基本操作(保姆级)

Python3操作Redis最新版|CRUD基本操作(保姆级)

Python3中类的高级语法及实战 Python3(基础|高级)语法实战(|多线程|多进程|线程池|进程池技术)|多线程安全问题解决方案 Python3数据科学包系列(一):数据分析实战 Python3数据科学包系列(二):数据分析实战 Python3数据科学包系列(三):数据分析实战 Win11查看安装的Python路…
阅读更多...
【SpringBoot】| Thymeleaf 模板引擎

【SpringBoot】| Thymeleaf 模板引擎

目录 Thymeleaf 模板引擎 1. 第一个例子 2. 表达式 ①标准变量表达式 ②选择变量表达式&#xff08;星号变量表达式&#xff09; ③链接表达式&#xff08;URL表达式&#xff09; 3. Thymeleaf的属性 ①th:action ②th:method ③th:href ④th:src ⑤th:text ⑥th:…
阅读更多...
Swift SwiftUI CoreData 过滤数据 2

Swift SwiftUI CoreData 过滤数据 2

预览 Code import SwiftUI import CoreDatastruct HomeSearchView: View {Environment(\.dismiss) var dismissState private var search_value ""FetchRequest(entity: Bill.entity(),sortDescriptors: [NSSortDescriptor(keyPath: \Bill.c_at, ascending: false)…
阅读更多...
Golang 程序漏洞检测利器 govulncheck（三）：github 集成方法

Golang 程序漏洞检测利器 govulncheck（三）：github 集成方法

上一篇文章详细介绍了 Golang 程序漏洞扫描工具 govulncheck 使用的漏洞数据库&#xff08;Go vulnerability database&#xff09;&#xff0c;本文详细讲解下 Github 项目如何使用 govulncheck。 govulncheck 为 Golang 开发者提供了一种准确可靠的方式来了解程序中可能存在…
阅读更多...
Kafka 高可用

Kafka 高可用

正文 一、高可用的由来 1.1　为何需要Replication 在Kafka在0.8以前的版本中&#xff0c;是没有Replication的&#xff0c;一旦某一个Broker宕机&#xff0c;则其上所有的Partition数据都不可被消费&#xff0c;这与Kafka数据持久性及Delivery Guarantee的设计目标相悖。同时Pr…
阅读更多...
MySQL增删查改（进阶1）

MySQL增删查改（进阶1）

一、数据库约束 约束&#xff1a;按照一定条件进行规范的做事&#xff1b; 表定义的时候&#xff0c;某些字段保存的数据需要按照一定的约束条件&#xff1b; 1.null约束 字段null&#xff1a;该字段可以为空&#xff1b;not null&#xff1a;该字段不能为空不指定的话就是…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023