Actor并发编程模型实现多文件的单词统计（完整源码）

给定几个文本文件&＃xff08;文本文件都是以空格分隔的&＃xff09;&＃xff0c;使用Actor并发编程来统计单词的数量

实现思路

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

实现思路

在main方法中读取指定目录(${project_root_dir}/data/)下的所有文件&＃xff0c;并打印所有的文件名

实现步骤

1. 创建用于测试的数据文件
2. 加载工程根目录&＃xff0c;获取到所有文件
3. 将每一个文件名&＃xff0c;添加目录路径
4. 打印所有文件名

实现思路

根据文件数量创建WordCountActor&＃xff0c;为了方便后续发送消息给Actor&＃xff0c;将每个Actor与文件名关联在一起

实现步骤

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

实现思路

启动所有WordCountActor&＃xff0c;并发送单词统计任务消息给每个WordCountActor

[!NOTE]

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

实现步骤

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

实现思路

读取文件文本&＃xff0c;并统计出来单词的数量。例如&＃xff1a;

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


实现步骤

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

实现思路

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

实现步骤

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

实现思路

对接收到的所有单词计数进行合并。因为该部分已经在WordCountActor已经编写过&＃xff0c;所以抽取这部分一样的代码到一个工具类中&＃xff0c;再调用合并得到最终结果

实现步骤

1. 创建一个用于单词合并的工具类
2. 抽取重复代码为一个方法
3. 在MainActor调用该合并方法&＃xff0c;计算得到最终结果&＃xff0c;并打印

1.txt

hadoop sqoop hadoop
hadoop hadoop flume
hadoop hadoop hadoop
spark


2.txt

flink hadoop hive
hadoop sqoop hadoop
hadoop hadoop hadoop
spark


MainActor.scala

package com.xuimport java.io.Fileimport scala.actors.Futureobject MainActor {def main(args: Array[String]): Unit &＃61; {// 1. 加载指定目录的文件列表// 1.1. 加载指定目录的数据文件val DIR_PATH &＃61; "./data/"// 获取到指定目录下的所有数据文件名val fileNameList &＃61; new File(DIR_PATH).list().toList// 1.2. 将数据文件添加上一个目录val fileDirNameList &＃61; fileNameList.map(DIR_PATH &＃43; _)// 1.3. 打印所有的文件名println(fileDirNameList)// 2. 创建Actor关联文件// 2.1. 创建Actorval wordCountActorList &＃61; fileNameList.map {fileName &＃61;> new WordCountActor}// 2.2. 将Actor和文件名关联在一起val actorFileNameList: List[(WordCountActor, String)] &＃61; wordCountActorList.zip(fileDirNameList)println(actorFileNameList)// 3. 启动Actor/发送/接收消息val futureList &＃61; actorFileNameList.map {actorFileName &＃61;>val actor &＃61; actorFileName._1// 启动Actoractor.start()// 发送消息到Actor中&＃xff0c;发送的是异步还有返回的消息val future: Future[Any] &＃61; actor !! WordCountTask(actorFileName._2)future}// 编写一个while循环来等待所有的Actor都已经返回数据while(futureList.filter(!_.isSet).size !&＃61; 0) {}// 获取Future中封装的数据val wordCountResultList &＃61; futureList.map(_.apply().asInstanceOf[WordCountResult])// 获取样例类中封装的单词统计结果// List[Map(hadoop->3, spark->1), Map(hadoop->2, flink->1)]val wordCountResultMap: List[Map[String, Int]] &＃61; wordCountResultList.map(_.wordCountMap)// List[hadoop->3, spark->1, hadoop->2, flink->1]val resultList &＃61; WordCountUtil.reduce(wordCountResultMap.flatten)println(resultList)}
}

MessagePackage.scala

package com.xu/*** 单词统计任务消息* &＃64;param fileName 文件名*/
case class WordCountTask(fileName:String)/*** 封装单词统计的结果* &＃64;param wordCountMap 单词-单词在某个文件中出现的数量*/
case class WordCountResult(wordCountMap: Map[String, Int])

WordCountActor.scala

package com.xuimport scala.actors.Actor
import scala.io.Sourceclass WordCountActor extends Actor{override def act(): Unit &＃61; {loop {react{case WordCountTask(fileName) &＃61;>println("接收到任务&＃xff1a;对" &＃43; fileName &＃43; "进行单词统计")// 1. 读取文件&＃xff0c;转换为列表// hadoop sqoop hadoopval lineList &＃61; Source.fromFile(fileName).getLines().toList// 2. 切割字符串&＃xff0c;转换成一个一个的单词// [hadoop, sqoop, hadoop]val wordList: List[String] &＃61; lineList.flatMap(_.split(" "))// 3. 将单词转换为一个元组// [, , ]val wordAndCountList: List[(String, Int)] &＃61; wordList.map(_ -> 1)// 4. 调用工具类方法来对单词元组列表进行分组、聚合val wordCountMap: Map[String, Int] &＃61; WordCountUtil.reduce(wordAndCountList)// 5. 打印测试println(wordCountMap)// 6. 将统计数据封装到一个样例类中&＃xff0c;发送给MainActorsender ! WordCountResult(wordCountMap)}}}
}

WordCountUtil.scala

package com.xuobject WordCountUtil {// 单词数据合并def reduce(wordAndCountList: List[(String, Int)]) &＃61; {// 4. 分组、聚合计算// {hadoop->List(,), sqoop->List()}val groupedMap: Map[String, List[(String, Int)]] &＃61; wordAndCountList.groupBy(_._1)// 聚合计算// {hadoop->2, sqoop->1}val wordCountMap: Map[String, Int] &＃61; groupedMap.map{keyVal &＃61;>keyVal._1 -> keyVal._2.map(_._2).sum}wordCountMap}
}