Akka实战、使用Scala的Akka编写wordCount实例，wordCount的目标文件是多个文件 22

本是古典何须时尚 2022-01-27 05:43 184阅读 0赞

# 前言 #

需要计算的目标文件是多个txt文档，代码如下

import java.io.File
    
    import scala.actors.{ Actor, Future}
    import scala.collection.immutable._
    import scala.collection.mutable
    import scala.collection.mutable.ListBuffer
    import scala.io.{ BufferedSource, Source}
    
    /** * */
    //0. 样例类,
    //发送消息
    case class SendMesN(fileName: String)
    
    case class ReturnMesN(map: Map[String, Int])
    
    class WordCountN extends Actor { 
      override def act(): Unit = { 
        loop { 
          react { 
            case SendMesN(fileName) => { 
              //1. 读取源文件
              val file: BufferedSource = Source.fromFile(new File(fileName))
              val contents: String = file.mkString
    
              //切割为数组
              val lines: Array[String] = contents.split("\r\n")
              //对每一行按照空格切割, 然后flat到同一层级
              val array1: Array[String] = lines.flatMap(_.split(" "))
    
              //2. 自定义map逻辑, k2 v2
              val array2: Array[(String, Int)] = array1.map((_, 1))
    
              //3. 分组,
              val map1: Map[String, Array[(String, Int)]] = array2.groupBy(x => x._1)
    
              //4. 自定义reduce逻辑
              val mapNew: Map[String, Int] = map1.mapValues(x => x.size)
    // println(mapNew)
    
              sender ! ReturnMesN(mapNew)
    
    
            }
            case _ => { 
              println("没有收到消息")
            }
    
          }
        }
    
      }
    
    }
    
    object WordCountN { 
      def main(args: Array[String]): Unit = { 
        //封装url
        val array: Array[String] = Array("E:\\01HM\\00pro\\scala-day02\\aaa\\wordCount.txt", "E:\\01HM\\00pro\\scala-day02\\aaa\\wordCount1.txt", "E:\\01HM\\00pro\\scala-day02\\aaa\\wordCount2.txt")
        //调用start()方法
        val wo: WordCountN = new WordCountN
        wo.start()
    
    
        //定义一个set集合, 用于接收未来返回值
        val resultSet = new mutable.HashSet[Future[Any]]
        val resultList = new mutable.ListBuffer[ReturnMesN]
    
        //循环发送异步有返回值的消息
        for (fileName <- array) { 
    
          val future: Future[Any] = wo !! SendMesN(fileName)
          resultSet += future
        }
    
        //判断resultSet中是否有值
        while (resultSet.size > 0) { 
          //如果有值, 需要过滤掉isSet为false的为来值
          val futureResult: mutable.HashSet[Future[Any]] = resultSet.filter(_.isSet)
    
          //遍历set中的结果,
          for (item <- futureResult) { 
            //通过apply()和强转方法, 得到每一个文件的ReturnMesN(map)
            resultList += item.apply().asInstanceOf[ReturnMesN]
    
            //需要从set中移除,要不然,循环停不了
            resultSet -= item
    
          }
        }
        
          //resultList
          //[{result:{hadoop -> 2, love -> 2, i -> 2, too -> 1}},{result:{hadoop -> 2, love -> 2, i -> 2, too -> 1}},{result:{hadoop -> 2, love -> 2, i -> 2, too -> 1}}]
          //1. 得到内存
          // [{hadoop -> 2, love -> 2, i -> 2, too -> 1},{hadoop -> 2, love -> 2, i -> 2, too -> 1},{hadoop -> 2, love -> 2, i -> 2, too -> 1}]
          val list1: ListBuffer[Map[String, Int]] = resultList.map(_.map)
          //2. 内存再抽离出来
          // [(hadoop -> 2), (love -> 2), (i -> 2), (too -> 1)...]
          val list2: ListBuffer[(String, Int)] = list1.flatten
    
          //3. 分组
          val map1: Map[String, ListBuffer[(String, Int)]] = list2.groupBy(_._1)
    
          //4. 自定义reduce逻辑
          val mapNew: Map[String, Int] = map1.mapValues(_.foldLeft(0)(_ + _._2))
    // val mapNew: Map[String, Int] = map1.mapValues(_.size)
          println(mapNew)
        }
    }