【Scala】Actor并发编程实现单机版wordCount

文章目录

• 对单个文本文件进行单词计数
• 对多个文本文件进行单词计数

对单个文本文件进行单词计数

import scala.actors.Actor
import scala.io.Source//读取文件名称样例类
case class SubmitTask(fileName:String)class Actor2WordCount extends Actor{
  override def act(): Unit = {
    loop{
      react{
        case SubmitTask(fileName) =>
          //根据文件名称读取文件内容 Source.formFile
          val fileContent: String = Source.fromFile(fileName).mkString
          //拿到文件内容后进行分割
          //首先文件有两行，所以按照换行符先进行分割   window系统:/r/n   linux系统:/n   mac OS系统:/r
          val lines: Array[String] = fileContent.split("\r")
          //再根据单词间的分隔符进行切割压平
          val words: Array[String] = lines.flatMap(x => x.split(" "))
          //还有一种写法
          // val words = lines.flatMap(_.split(" "))
          //每个单词都标记成一次
          val countOne: Array[(String, Int)] = words.map(x => (x, 1))
          //进行分组，把元祖第一个元素相同分配到一组
          val groupedWords: Map[String, Array[(String, Int)]] = countOne.groupBy(_._1)
          //mapValues方法可以作用于map中的v，操作之后和和之前的k返回构成一个新的map
          val wordsCount: Map[String, Int] = groupedWords.mapValues(_.length)
          println(wordsCount.toBuffer)
      }
    }
  }
}object WordCount{
  def main(args: Array[String]): Unit = {
    //创建Actor2WordCount对象
    val actor = new Actor2WordCount
    //启动Actor2WordCount
    actor.start()
    //发送待处理文件名称给Actor2WordCount
    actor ! SubmitTask("/Users/zhaozhuang/Desktop/3、Scala/2、Scala第二天/wordCount/1.txt")
  }
}

对多个文本文件进行单词计数

import scala.actors.{Actor,Future}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.io.{BufferedSource,Source}//读取文件名称样例类
case class SubmitTask(fileName: String)case class Reply(wordsCount: Map[String, Int])class Actor2WordCount extends Actor {
  override def act(): Unit = {
    loop {
      react {
        case SubmitTask(fileName) =>
          //根据文件名称读取文件内容 Source.formFile
          val fileContent: String = Source.fromFile(fileName).mkString
          //拿到文件内容后进行分割
          //首先文件有两行，所以按照换行符先进行分割   window系统:/r/n   linux系统:/n   mac OS系统:/r
          val lines: Array[String] = fileContent.split("\r")
          //再根据单词间的分隔符进行切割压平
          val words: Array[String] = lines.flatMap(x => x.split(" "))
          //还有一种写法
          // val words = lines.flatMap(_.split(" "))
          //每个单词都标记成一次
          val countOne: Array[(String, Int)] = words.map(x => (x, 1))
          //进行分组，把元祖第一个元素相同分配到一组
          val groupedWords: Map[String, Array[(String, Int)]] = countOne.groupBy(_._1)
          //mapValues方法可以作用于map中的v，操作之后和和之前的k返回构成一个新的map
          val wordsCount: Map[String, Int] = groupedWords.mapValues(_.length)          sender ! Reply(wordsCount)
          //          println(wordsCount.toBuffer)
      }
    }
  }
}object WordCount {
  def main(args: Array[String]): Unit = {
    //将要读取的文件路径封装到一个数组中
    val files = Array("/Users/zhaozhuang/Desktop/3、Scala/2、Scala第二天/wordCount/1.txt",
      "/Users/zhaozhuang/Desktop/3、Scala/2、Scala第二天/wordCount/2.txt",
      "/Users/zhaozhuang/Desktop/3、Scala/2、Scala第二天/wordCount/3.txt")    //定义一个集合，用于保存每个actor处理完返回的状态
    val futureSet: mutable.HashSet[Future[Any]] = new mutable.HashSet[Future[Any]]()
    //定义一个集合，用于保存买个actor返回的结果
    val resultList = new ListBuffer[Reply]
    //循环遍历每一个文件
    for (f <- files) {
      //创建Actor2WordCount对象
      val actor = new Actor2WordCount
      //启动Actor2WordCount
      actor.start()
      //发送待处理文件名称给Actor2WordCount   异步消息且有返回值
      val future = actor !! SubmitTask(f)
      //把返回状态添加到future集合中
      futureSet.+=(future)
    }
    //遍历futureSet,找出真正处理完的actor，提取其结果future.apply()
    while (futureSet.size > 0){
      val completeFuture: mutable.HashSet[Future[Any]] = futureSet.filter(x => x.isSet)
      //提取处理的结果
      for(f <- completeFuture){
        val finalResult: Any = f.apply()
        val reply: Reply = finalResult.asInstanceOf[Reply]
        resultList += reply
        //提取完结果后，把future从futureSet移除
        futureSet.remove(f)
      }
    }
    println(resultList.map(_.wordsCount).flatten.groupBy(_._1).mapValues(x => x.foldLeft(0)(_+_._2)))
  }
}