本站消息

站长简介/公众号

出租广告位,需要合作请联系站长

java王侯

2711

文章

1383135

访问

+关注

分类

java(0)

日期归档

2023-06(2)

Scala基础（三）模式匹配、样例类与Actor编程

发布于2021-06-14 10:01 阅读(732) 评论(0) 点赞(20) 收藏(0)

Scala语法

Scala函数式编程与面向对象编程

模式匹配

之前有写过Scala中也可以模拟出Java和C#的switch/case功能。

内容匹配

package com.aa.matchDEMO

import scala.io.StdIn

object MatchCase1 {
  def main(args: Array[String]): Unit = {
    
    //从控制台stdin（标准输入）接收用户输入的内容，根据内容进行模式匹配
    val result: String = StdIn.readLine()       // 死等，这是阻塞方法

    result match {
      case "hadoop" => println("hadoop>>>")
      case "spark" => println("spark>>>")
      case "flink" => println("flink>>>")
      case _=>println("*********hehe*********")
    }
  }
}

多运行几次：

flink
flink>>>
hadoop
hadoop>>>
哈哈
*********hehe*********
spark
spark>>>

Process finished with exit code 0

数据类型匹配

package com.aa.matchDEMO

import scala.util.Random

object MatchCase2 {
  def main(args: Array[String]): Unit = {
    val array = Array(11, 22, "hello", "bye", true, false, 3.14, 6.33)

    val result: Any = array(Random.nextInt(array.length))		//随机抽取数组元素

    result match {
      //case x: Int  => println(x)
      case x: Int if x > 15 => println(x)     				 //匹配时+守卫条件，提前过滤
      case y: String => println(y)
      case z: Double => println(z)
      case w: Boolean => println(w)
      case _ => println("-----呵呵-----")      //不满足上方的任一条件就会落入这里，相当于default
    }
  }
}

随机获取元素并进行类型匹配。使用守卫条件在随机到11时会落入最后的_中。

数组与集合匹配

package com.aa.matchDEMO

import scala.util.Random

//匹配数组、集合
object MatchCase3 {
  def main(args: Array[String]): Unit = {
    var array = Array(1, 3, 5)
    array match {
      case Array(1, x, y) => println(x + " " + y)
      case Array(1, 3, 5) => println("精准匹配")
      case Array(1, _*) => println("1...")
      case _ => println("else")
    }

    //val list = List(0)
    //val list = List(0, "a", 2)
    val list = List(0, "a")
    //val list = List(0, "a")
    //val list = List(1,"a",1)
    list match {
      case 0 :: Nil => println("只有0的列表")
      //case 0::_ =>println("0开头的列表")
      case x :: y :: Nil => println(s"只有2个元素${x},${y}的列表")
      case _ => println("else列表")
    }
  }
}

声明变量时匹配

package com.aa.matchDEMO

//变量声明时进行模式匹配
object MatchCase4 {
  def main(args: Array[String]): Unit = {
    val result: Array[Int] = (1 to 5).toArray

    val Array(_,x,y,_*) = result      //变量声明时进行模式匹配
    println(s"x=${x},y=${y},x+y=${x+y}")  
  }
}

x=2,y=3,x+y=5

Process finished with exit code 0

在声明变量时可以直接进行模式匹配，将匹配的结果作为新声明变量的初值。

样例类

样例类是Scala中的特殊类（case修饰的），用于封装数据，参与模式匹配、传递数据。显然样例类有2种：

case class：多例，需要构造器和参数列表。
case object：单例，不需要构造器（也就不需要参数列表）。

类+case变成样例类后底层发生了改变：

样例类自动实现了apply方法，不需要new。
样例类自动实现了toString方法，不再是打印内存地址，而是直接输出属性。
样例类自动实现了hashcode和equals方法，对象实例不再是比较内存地址，而是直接比较内容。

举个栗子：

package com.aa.caseDEMO

object Case1 {
  def main(args: Array[String]): Unit = {
    val cat1 = new Cat("哈哈") //普通类需要先new才能创建实例对象
    val cat2 = new Cat("哈哈")

    println(cat1) //输出com.aa.caseDEMO.Cat@4cdbe50f  普通类直接输出的是内存地址
    println(cat1 == cat2) //输出false 直接比较的是内存地址

    val dog1 = new Dog("呵呵") //idea自动提示new是多余的，new关键字发灰说明可以不写
    val dog2 = Dog("呵呵")//不写new也可以创建实例对象
    println(dog1) //Dog(呵呵) 说明自动写好了toString
    println(dog1 == dog2) //true 说明自动写好了equals和hashCode方法
  }
}

class Cat(var name: String) {
  //override def toString: String = super.toString    默认的toString方法
  override def toString: String = s"Cat[$name]"

  //override def equals(obj: Any): Boolean = super.equals(obj)  默认的equals方法
  def canEqual(other: Any): Boolean = other.isInstanceOf[Cat]

  override def equals(that: Any): Boolean = that match {
    case that: Cat => (that canEqual this) && name == that.name //是同一类的实例对象才能比较对象的内容
    case _ => false
  }

  override def hashCode(): Int = { //重写hashCode方法
    //先比较hashCode，如果不同肯定内容不同，如果hashCode相同内容可能不同，再去equals比较内容，速度快
    val state = Seq(name)
    state.map(_.hashCode).foldLeft(0)((a, b) => 31 * a + b) //随便瞎写的算法
  }
}

//样例类
case class Dog(var name: String) {}

样例类的核心功能是封装数据+传输数据+模式匹配：

package com.aa.caseDEMO

import scala.util.Random

object Case2 {
  def main(args: Array[String]): Unit = {
    val array1: Array[Object] = Array(new Student("李四"), Teacher("王五"), Teacher("马六"), Zhangsan)
    val array2: Array[Object] = Array(Student("李四"), Teacher("王五"), Teacher("马六"), Zhangsan)

    val result: Object = array2(Random.nextInt(array2.length)) //Object是Java的Object，顶级父类

    result match {
      case Teacher(name) => println(name) //直接提取属性
      case Student(name) => println(name)
      case s: Student => println(s.name) //先匹配对象，在调用对象.属性
      case Zhangsan => println("这货是张三")
      case _ => println("其它情况")
    }
  }
}

class Student(var name: String)

case class Teacher(var name: String)

case object Zhangsan

object Student { //伴生对象内定义apply方法和unapply方法
  def apply(name: String): Student = new Student(name)

  def unapply(s: Student): Option[String] = {
    Some(s.name)
  }
}

补充

apply方法

package com.aa.others

//测试apply方法
object ApplyCase {
  def main(args: Array[String]): Unit = {
    val array1 = new Array[Int](10)       //没有初始值，需要new

    val array2:Array[Int] = Array(11,22,33)    //有初始值，不需要new
  }
}

之前在命令行就发现了创建集合时，有初始值就可以不用new。尝试定位：

在这里插入图片描述

在Array.class第27行看到：

  def apply(x : scala.Int, xs : scala.Int*) : scala.Array[scala.Int] = { /* compiled code */ }

Apply其实也是一种构造方法，在创建对象时，如果不+new，编译器就会尝试寻找是否有apply方法，如果有就调用apply方法实现对象的创建。

package com.aa.others

//测试apply方法创建类的实例对象
object ApplyCase1 {
  def main(args: Array[String]): Unit = {
    val student = Student //object是全局唯一的实例（静态加载，单例）

    val zhangsan = new Person("张三")//普通类创建实例对象时必须new
    val lisi:Person = Person("李四")//调用apply方法完成new
  }
}

object Student //object是静态加载的单例

class Person(var name: String)

object Person { //apply方法必须定义在类的伴生对象中
  //idea自动提示并补全apply方法
  def apply(name: String): Person = new Person(name)
}

在类的伴生对象中定义apply方法后，就可以直接调用伴生对象的apply方法，不用new。

unapply方法

从字面上也知道先得有apply方法，才能有unapply方法：

object Student{  
def unapply(s: unapplyStudent): Option[String] = {
    Some(s.name)}
}

case Student(name) => println(name)

unapply方法中可以使用Some方法，定位到Some.class：

package scala
@scala.SerialVersionUID(value = 1234815782226070388)
final case class Some[+A](val x : A) extends scala.Option[A] with scala.Product with scala.Serializable {
  def isEmpty : scala.Boolean = { /* compiled code */ }
  def get : A = { /* compiled code */ }
}

可以看到Some内部有get方法，其实unapply方法就是从对象中提取属性，相当于提取器。

Option类型

量子叠加类型，表示可有可无的2种状态，分为2个类：有（Some）、无（None）。

package com.aa.others

object OptionCase {
  def main(args: Array[String]): Unit = {
    val map1 = Map("name"->"张三","age"->"13")

    //根据Key去Map中取值，有值显式，无值为None
    val result1:Option[String] = map1.get("name")
    val result2:Option[String] = map1.get("age1")

    println(result1)      //Some(张三)  有值
    println(result2)      //None
  }
}

再看看Option类型多么“薛定谔”：

package com.aa.others

object OptionCase1 {
  def main(args: Array[String]): Unit = {
    def method1(x: Int, y: Int) = x / y

    println(method1(5, 2))
    //println(method1(5, 0))  会报错
    /*
    Exception in thread "main" java.lang.ArithmeticException: / by zero
	at com.aa.others.OptionCase1$.method1$1(OptionCase1.scala:5)
	at com.aa.others.OptionCase1$.main(OptionCase1.scala:6)
	at com.aa.others.OptionCase1.main(OptionCase1.scala)

Process finished with exit code 1
     */

    def method2(x: Int, y: Int): Option[Int] = {
      if (y != 0) Some(x/y)
      else None
    }

    val result = method2(5, 3)
    println(result)
    val maybeInt = method2(5, 0)
    
    maybeInt match {
      case Some(s) =>println(s)
      case None =>println("除数=0，出错")
    }
  }
}

偏函数Partial Function

这个“偏”类似“偏导数”的“偏”。Partial Function字面意思是部分的、不完全的函数。

表面上没有match，实际上却有一组case语句：

偏函数是Partial Function[A,B]的一个实例，A代表输入参数类型，B代表返回结果类型。

举个栗子：

package com.aa.others

object PF1 {
  def main(args: Array[String]): Unit = {
    val array1: Array[Int] = Array(11, 22, 33)
    //println(array1.map(_*10))//[I@48140564  需要转化才能打印
    println(array1.map(_ * 10).toBuffer) //ArrayBuffer(110, 220, 330)

    val array2: Array[Any] = Array(11, 22, "哈哈", 33)
    //println(array2.map(_*10)) 为了保证安全，默认不允许这么做
    println(array2.filter(_.isInstanceOf[Int])//先按数据类型进行合法性过滤
      .map(_.asInstanceOf[Int])//链式编程，对合法数据进行类型转化
      .map(_*10).toBuffer)

    val pf1:PartialFunction[Any,Int]={
      case i:Int=>i*10
    }
    println(array2.collect(pf1).toBuffer)//collect方法接收偏函数作为参数
  }
}

可以看出偏函数相比链式编程的过滤→类型转换→函数运算的三步走，只需要定义输入、输出类型与函数逻辑就可以直接出结果，还是很简洁的。

package com.aa.others

object PF2 {
  def main(args: Array[String]): Unit = {
    def method1(i:Int) = i match {
      case 1 => println("一")
      case 2 => println("二")
      case 3 => println("三")
    }
    method1(2)

    val pf2:PartialFunction[Int,String] ={
      case 1 =>"一"
      case 2 =>"二"
      case 3 =>"三"
    }
    println(pf2(1))
  }
}

使用偏函数进行模式匹配和运算显然也很简洁。

正则

正则（学名Regular Expression），又名规则表达式，用于对字符串内容进行匹配。在文本处理、上位机数据提取、ETL等操作中都需要正则。Scala中的正则类就叫RegEx类，直接new即可使用，当然也可以String.r→RegEx类。

package com.aa.others

import scala.util.matching.Regex

object RegExTest {
  def main(args: Array[String]): Unit = {
    val email1 = "1009057838@QQ.com"
    val email2 = "haha@com"

    //正常情况使用三对双引号可以跨行
    //正则不行(出现了多余的标记)
//    val regex: Regex =
//    """
//      |.+@.+\..+
//      |""".r

    /*结果不正确
    List(, , , , , , , , , , , , , , , , , )
    List(, , , , , , , , )
     */

    val regex: Regex =
      """.+@.+\..+""".r
    println(regex.findAllMatchIn(email1).toList)
    println(regex.findAllMatchIn(email2).toList)
    /*
    List(1009057838@QQ.com)
    List()
     */

  }
}

Scala异常处理

和Java差不多，也可以使用idea的ctrl+art+t来抽取并封装到try/catch/finally中：

package com.aa.others

object ExceptionCase1 {
  def main(args: Array[String]): Unit = {
    try {//idea使用ctrl+art+t抽取并封装在try中
      val error = 5 / 0
      /*常见的分母0报错
       Exception in thread "main" java.lang.ArithmeticException: / by zero
       at com.aa.others.ExceptionCase1$.main(ExceptionCase1.scala:5)
       at com.aa.others.ExceptionCase1.main(ExceptionCase1.scala)
      */
    } catch {//捕获异常后对异常进行处理
      case e1:ArithmeticException => println("数值异常"+e1.getMessage)
      case e2:NullPointerException => println("空指针异常")
    } finally {//无论如何都会执行
      println("一定会执行")
    }

  }
}

当然偷懒的办法是throw抛异常：

package com.aa.others

object ExceptionCase2 {
  def main(args: Array[String]): Unit = {
    println("开始执行")

    throw new Exception("抛出异常")
    
    /*执行后出现
    开始执行
    Exception in thread "main" java.lang.Exception: 抛出异常
	  at com.aa.others.ExceptionCase2$.main(ExceptionCase2.scala:7)
	  at com.aa.others.ExceptionCase2.main(ExceptionCase2.scala)

    Process finished with exit code 1
     */

    println("永远不会执行")//idea中变灰，说明是无效代码，没有执行机会
  }
}

但是Scala不需要在方法上抛异常。。。

泛型

泛型是参数化类型。泛型程序设计（generic programming）。Scala中的泛型在使用时，通常用26个英文字母表示，常用T、A、B。套路一般是：

def 方法名[泛型名称](..) = {
//...
}

举个栗子：

package com.aa.genDEMO;

import java.util.ArrayList;

public class TestA {
    public static void main(String[] args) {
        Student1 student1 = new Student1();
        Person1 person1 = new Student1();//类的多态性，向上转型

        ArrayList<Student1> arrayList1 = new ArrayList<Student1>();
        //ArrayList<Person1> arrayList2 = new ArrayList<Student1>();
        //Java中类的继承关系不能延伸到集合，非变
    }
}

//定义Java类和继承类
class Person1{}
class Student1 extends Person1{}

会报错：

在这里插入图片描述

因为Java支持的是非变。

package com.aa.genDEMO

object TestGen{
  def main(args: Array[String]): Unit = {
    val array1 =new Array[String](4)    //T箭头String
    val array2 = new Array[Int](2)    //T→Int

    //需要定义多个方法。。。很麻烦
    def method1(x:Array[String]) = x(x.length/2)
    def method2(x:Array[Int]) = x(x.length/2)
    println(method1(Array("11","哈哈","呵呵","嗷嗷嗷")))
    println(method2(Array(11,22,33)))

    //使用泛型，指代一个参数，泛指多数类型，只需要定义一个方法
    def method3[A](x:Array[A])=x(x.length/2)
    println(method3(Array("11","哈哈","呵呵","嗷嗷嗷")))
    println(method3(Array(11,22,33)))
    println(method3(Array(11.11,22.22,33.33)))
  }
}

还可以定义泛型类：

package com.aa.genDEMO

object TestGen1 {
  def main(args: Array[String]): Unit = {
    val cat1:Cat1 = Cat1("哈哈")
    //val cat2 = Cat1(11)//报错Type mismatch

    val cat3 = Cat2("呵呵")
    val cat4 = Cat2(14)
    val cat5 = Cat2(true)

  }
}

case class Cat1(var name:String)
case class Cat2[T](var name:T)//定义泛型类

泛型还有上下界的问题：

[T <: 类型]    T <=类型   上界 T类型必须是后面指定的类型及其子类  

[T >: 类型]    T >=类型    下界 T类型必须是后面值的类型及其父类

如果类既有上界、又有下界。下界写在前面，上界写在后面

泛型还存在泛型延伸问题（Scala除了支持Java支持的非变外，还支持协变和逆变）：

class Super
class Sub extends Super

class Temp1[T]  //非变 
class Temp2[+T] //保持一致 继续延伸
class Temp3[-T] //逆转过来 延伸

def main(args: Array[String]): Unit = {
	val a:Temp1[Sub] = new Temp1[Sub]
	// 编译报错
	// 非变
	//val b:Temp1[Super] = a
    
	// 协变
	val c: Temp2[Sub] = new Temp2[Sub]
	val d: Temp2[Super] = c
	
    // 逆变
	val e: Temp3[Super] = new Temp3[Super]
	val f: Temp3[Sub] = e
}

Scala Actor并发编程

准备工作

如果导包时没有需要的类库：

在这里插入图片描述

这是∵类库路径有误（安装Scala时默认有该类），需要在idea的左上角点file→Project Structure→Global Libraries中设置Scala SDK的路径：

在这里插入图片描述

这里点+号，选择安装路径的lib包，确认后即可导包。

并发编程对比

Java的并发编程

模型是基于多线程+共享数据：线程越多，并发能力越强。当然不易维护，容易产生死锁。

package com.aa.lock;


//jps -l 查看java进程 jstack + 进程号 查看死锁的情况

public class DeadLock {
    private  Object lock1 = new Object();
    private  Object lock2 = new Object();

    public  void method1() throws InterruptedException {
        synchronized(lock1){
            System.out.println(Thread.currentThread().getName() + "获取到lock1,请求获取lock2....");
            Thread.sleep(1000);
            synchronized (lock2){
                System.out.println("获取到lock2....");
            }
        }
    }
    public  void method2() throws InterruptedException {
        synchronized(lock2){
            System.out.println(Thread.currentThread().getName() + "获取到lock2,请求获取lock1....");
            Thread.sleep(1000);
            synchronized (lock1){
                System.out.println("获取到lock1....");
            }
        }
    }

    public static void main(String[] args) {
        DeadLock deadLock = new DeadLock();

        new Thread(()-> {
            try {
                deadLock.method1();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        new Thread(()-> {
            try {
                deadLock.method2();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

    }
}

Scala的并发编程

模型是基于actor事件+消息发送和接收：actor越多并发能力越强。由于不共享数据，当然不存在锁的问题。

简单使用

package com.aa.actorDEMO

import scala.actors.Actor

class Actor1 extends Actor{
  override def act(): Unit = {
    for(i <- 1 to 10){
      println("Actor1:"+i)
      Thread.sleep(2000)
    }
  }
}
class Actor2 extends Actor{
  override def act(): Unit = {
    for(i <- 1 to 10){
      println("Actor2:"+i)
      Thread.sleep(2000)
    }
  }
}

object Test1{
  def main(args: Array[String]): Unit = {
    //创建actor实例
    val actor = new Actor1
    val actor2 = new Actor2
    //启动actor
    actor.start()
    actor2.start()
  }
}

执行后看起来还是很像多线程的：

Actor2:1
Actor1:1
Actor1:2
Actor2:2
Actor1:3
Actor2:3

Process finished with exit code -1

死循环收发消息

Scala容易报错，为了防止不必要的麻烦，尽量不同名（不使用class Actor2）：

package com.aa.actorDEMO

import scala.actors.Actor
class Actor3 extends Actor{
  override def act(): Unit = {
    while (true){
      receive{
        case "hello"  => println("\\(@^0^@)/")
        case "bye" => println("ヾ(•ω•`)o")
        case "stop" => System.exit(0)
      }
    }
  }
}

object Actor3{
  def main(args: Array[String]): Unit = {
    val actor = new Actor3
    actor.start()
    //自己给自己发送消息
    actor!"hello"
    actor!"hello"
    actor!"hello"
    actor!"bye"
    actor!"stop"
    actor!"hello"
  }
}

\(@^0^@)/
\(@^0^@)/
\(@^0^@)/
ヾ(•ω•`)o

Process finished with exit code 0

其中：

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

while(true)是阻塞式，效率低下，也不能复用线程。

在这里插入图片描述

由于构造了伴生关系，idea左侧的图表都变了。。。很智能的玩意儿。。。

React收发消息

react方法配合loop可以复用线程，更高效：

package com.aa.actorDEMO

import scala.actors.Actor
class Actor4 extends Actor{
  override def act(): Unit = {
    loop{
      react{
        case "hello"  => println("\\(@^0^@)/")
        case "bye" => println("ヾ(•ω•`)o")
        case "stop" => System.exit(0)
      }
    }
  }
}



object Actor4{
  def main(args: Array[String]): Unit = {
    val actor = new Actor4
    actor.start()
    //自己给自己发送消息
    actor!"hello"
    actor!"hello"
    actor!"hello"
    actor!"bye"
    actor!"stop"
    actor!"hello"
  }
}

Actor的3种消息发送方式

package com.aa.actorDEMO

import scala.actors.Actor
import scala.actors.Future

case class AsyncMessage(id: Int, msg: String)

case class Reply(msg: String)

case class SyncMessage(id: Int, msg: String)

class Actor5 extends Actor {
  override def act(): Unit = {
    loop {
      react {
        case AsyncMessage(id, msg) => { //异步
          println(s"id:${id},msg:${msg}")

          sender ! Reply("返回消息")
        }
        case SyncMessage(id, msg) => { //同步
          println(s"id:${id},msg:${msg}")
        }
      }
    }
  }
}

object Actor5 {
  def main(args: Array[String]): Unit = {
    val actor = new Actor5
    actor.start()

    //Actor有3种发送消息的方式
    actor ! AsyncMessage(1, "异步发送，不需要返回值")

    actor !? SyncMessage(2, "同步发送，阻塞等待返回值")

    val future: Future[Any] = actor !! AsyncMessage(3, "异步发送，需要返回值")
    val result: Reply = future.apply().asInstanceOf[Reply]
    println(result)

  }
}

同步发送一定会阻塞和死等，由于是单线程运行，直接卡在这一步：

id:1,msg:异步发送，不需要返回值
id:2,msg:同步发送，阻塞等待返回值

暂时屏蔽同步发送：

id:1,msg:异步发送，不需要返回值
id:3,msg:异步发送，需要返回值
Reply(返回消息)

显然异步可以介绍到消息。。。

Actor并发WordCount

比起之前在MapReduce中写的WordCount还是简单不少的：

原理是将分散在多个文件的数据，利用actor异步处理，将返回值先置于futureSet中，再运算到resultList中，最后整合到FinalResult中。

package com.aa.actorDEMO

import scala.actors.{Actor, Future}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.io.Source


//定义个样例类 用于封装提交的任务
case class SubmitTask(filePath: String)

//定义样例类 用于封装task处理的结果
case class ReplyResult(result: Map[String, Int])


class Task extends Actor {
  override def act(): Unit = {
    loop {
      react {
        //接收的task任务 处理数据 进行wc
        case SubmitTask(filePath) => {
          //从文件中读取数据 并且读取的数据转换为String。底层是全文读取
          val content: String = Source.fromFile(filePath).mkString
          //将读取的内容根据换行符切割成为一行行内容，\r\n就是分隔符
          val lines: Array[String] = content.split("\r\n")
          //根据分隔符进行单词切割，并扁平化操作（将零散的单词压入一个集合中）
          val words: Array[String] = lines.flatMap(line => line.split(" "))
          //将单词标记为键值对，单词→(单词，1) 的对偶元组（二元组）
          val wordAndOne: Array[(String, Int)] = words.map(word => (word, 1))
          //将单词相同的分为一组
          val wordGrouped: Map[String, Array[(String, Int)]] = wordAndOne.groupBy(tuple2 => tuple2._1)
          //scala集合中有个方法mapValues，将value位置的map处理为新的value，并将处理的结果和之前的key组成一个新的map
          val result: Map[String, Int] = wordGrouped.mapValues(v => v.length)
          //for( (k,v)<- result) println(s"${k}---->${v}")

          //task处理完的结果进行返回
          sender ! ReplyResult(result)
        }
      }
    }
  }
}

object WordCount {
  def main(args: Array[String]): Unit = {

    //创建一个集合 用于保存每个任务Future
    val futureSet = new mutable.HashSet[Future[Any]]()
    //创建一个集合 用于保存最终的每个task处理结果
    val resultList = new ListBuffer[ReplyResult]

    //待处理数据路径
    val files = Array("D:\\datasets\\scala\\1.txt", "D:\\datasets\\scala\\2.txt", "D:\\datasets\\scala\\3.txt")
    //val filePath ="D:\\datasets\\scala\\1.txt"
    //遍历待处理文件文件，分别启动task处理各个文件
    for (f <- files) {
      //创建启动TaskActor
      val task = new Task
      task.start()
      //把待处理的数据路径（任务）发送给task处理
      //因为需要对最终的结果进行合并，使用异步有返回值的消息
      val future: Future[Any] = task !! SubmitTask(f)
      //把future保存至集合中
      futureSet += future
    }

    //判断future是否已经完成有结果，如果有结果就提取结果，future.isSet用于判断结果是否有结果
    while (futureSet.nonEmpty) {
      //过滤出已经有结果的future
      val completed: mutable.HashSet[Future[Any]] = futureSet.filter(f => f.isSet)
      //遍历已经完成的 提取结果
      for (c <- completed) {
        //提取结果
        val result: ReplyResult = c.apply().asInstanceOf[ReplyResult]
        //将结果添加至resultList
        resultList += result
        //把提取完结果的future从futureSet给剔除
        futureSet.remove(c)
      }
    }

    println(resultList.flatMap(r => r.result)
      .groupBy(_._1)
      .mapValues(values => values.map(x => x._2).sum))
  }
}

按照函数式编程的做法，可以精简大量代码：

package com.aa.actorDEMO

import scala.actors.Actor
import scala.io.Source

//优化后的简化版本

//定义个样例类 用于封装提交的任务
case class SubmitTask(filePath:String)


class Task extends Actor {
  override def act(): Unit ={
    loop{
      react{
        //接收的task任务 处理数据 进行wc
        case SubmitTask(filePath) =>{
          //从文件中读取数据，并且读取的数据转换为String，底层是全文读
          val content: String = Source.fromFile(filePath).mkString
          val result: Map[String, Int] = content.split("\r\n")
            .flatMap(_.split(" "))
            .map((_, 1))
            .groupBy(_._1)
            .mapValues(_.length)

          println(result)
        }
      }
    }
  }
}

object WordCount {
  def main(args: Array[String]): Unit = {

    //待处理数据路径
    val filePath ="D:\\datasets\\scala\\1.txt"

    //创建启动TaskActor
    val task = new Task
    task.start()

    //把待处理的数据路径（任务）异步提交给task处理
    task ! SubmitTask(filePath)
  }
}

简化后可读性变差，但代码很简洁。

这里有个小细节：

在这里插入图片描述

传的参数必须是小写字母开头，不能大写字母开头，否则会报错：Cannot resolve symbol。Scala这点做的很奇怪。。。

在这里插入图片描述

对命令行的输入进行WordCount事实证明也是可行的：

package com.aa

import scala.collection.mutable.ListBuffer
import scala.io.StdIn
import scala.io.Source

object WordCount_Simple {
  def main(args: Array[String]): Unit = {
    var input = ListBuffer("哈 哈 哈 哈", "哈 哈 呵 呵 呵呵 哈哈")
    val inputPath = "E:\\study\\TopN\\origin.txt"
    var inputStr:String = "哈 哈 哈 哈"
    println("输入路径："+inputPath)
    val context:String = Source.fromFile(inputPath).mkString
    var wordcount_In = context.split("\r\n")
      .flatMap(_.split(" "))
      .map((_, 1))
      .groupBy(_._1)
      .map(x => (x._1, x._2.size))
      .toList.sortBy(_._2)
    println(wordcount_In)             //路径文件的WC

    val output = input.flatMap(_.split(" "))
      .map((_, 1))
      .groupBy(_._1)
      .map(x => (x._1, x._2.size))
      .toList.sortBy(_._2)
    println("测试数据："+output)
    while (true) {
      inputStr =StdIn.readLine()
      println("输入的内容："+inputStr)

      val input_WC: List[(String, Int)] = inputStr.split("\r\n")
        .flatMap(_.split(" "))
        .map((_, 1))
        .groupBy(_._1)
        .map(x => (x._1, x._2.size))
        .toList.sortBy(_._2)
      println(input_WC)

    }
  }
}