一、元组中的拉链操作

补充：Maven中可以创建一个大项目，大项目中可以添加Module，每一个Module就是一个完整的项目（可以称之为这个项目的一个模块）
复习：想在一个项目中编写Scala ,需要导入包：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>ScalaDay06</artifactId>
        <groupId>com.qfedu</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>day06</artifactId>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.12.8</version>
        </dependency>
    </dependencies>
</project>

还需要在项目中引入scala的支持，右键项目，选择第二个选项。

创建scala文件夹，修改为source 类型的文件夹

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 9:58
 * @Version 1.0
 */
object _01元组的拉链操作 {

    def main(args: Array[String]): Unit = {
        // 拉链  操作 两个数组中元素个数一样的情况
        var tuple1 = Array(1,2,3,4,5);
        var tuple2 = Array("a","b","c","d","e")
        val tuples: Array[(Int, String)] = tuple1.zip(tuple2)
        //println(tuples)
        for(tup <- tuples){
            println(tup._1,tup._2)
        }

        // 拉链操作，元素个数不一样
        var tuple3 = Array("hadoop","spark")
        //如果两个数组的元素个数不一致，拉链操作后生成的数组的长度为较小的那个数组的元素个数
        tuple1.zip(tuple3).foreach(tup => println(tup._1,tup._2))

        //如果就想让两个长度不一样的集合拉在一起，并且以长度最长的那个集合为准
        // zipAll 可以将两个 长度不一样的集合拉链在一起，长度短的集合可以用默认值填充
        // 0 是第一个集合的默认值填充值
        // "未知" 是第二个集合的默认值填充值
        tuple1.zipAll(tuple3,0,"未知").foreach(tup => println(tup._1,tup._2))

        // zipWithIndex  可以将某个集合跟自己的集合的小标形成一个拉链
        tuple2.zipWithIndex.foreach(tup => println(tup._1,tup._2))
        // 如果进行拉链的时候，下标不是从0开始的，可以使用如下方法
        tuple2.zip(Stream from 1).foreach(tup => println(tup._1,tup._2))

    }
}

二、Actor

介绍一下：

scala中的actor 是一个并发编程模型，用来开发比java更加高效的并发程序。是我们学习Akka的基础。

java的多线程缺点：
1、老出现线程安全问题（多个线程访问同一个资源）
2、出现死锁的问题

Scala中的Actor 是一个 并发编程模型（多线程的开发），是一个基于事件的模型（可以进行通信）

Actor1 发送消息给 Actor2 ,Actor2 可以给 Actor1 发送消息，基于事件的模型，数据不共享，不会出现java中的线程安全问题。

基调：Scala 2.11版本的时候引入了Akka,已经放弃了使用Actor ，但是思想是通用的，Actor是为了更好的理解Akka.

1、使用Actor进行并发编程

需要注意scala版本和actor版本的对应，不对应就报错

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.10.4</version>
        </dependency>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-actors</artifactId>
            <version>2.11.8</version>
        </dependency>
    </dependencies>

还需要修改这个项目所编译的scala的版本：

package com.qfedu

import scala.actors.Actor

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 10:29
 * @Version 1.0
 */
object _02Actor并发编程 {

    object Actor1 extends Actor {
        override def act(): Unit = (1 to 10).foreach(println(_))
    }

    object Actor2 extends Actor {
        override def act(): Unit = (11 to 20).foreach(println(_))
    }

    def main(args: Array[String]): Unit = {
        Actor1.start()
        Actor2.start()
    }
}

package com.qfedu

import scala.actors.Actor

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 10:29
 * @Version 1.0
 */
object _02Actor并发编程 {

    /*object Actor1 extends Actor {
        override def act(): Unit = (1 to 10).foreach(println(_))
    }

    object Actor2 extends Actor {
        override def act(): Unit = (11 to 20).foreach(println(_))
    }*/
    class Actor1 extends Actor {
        override def act(): Unit = (1 to 10).foreach(println(_))
    }

    class Actor2 extends Actor {
        override def act(): Unit = (11 to 20).foreach(println(_))
    }

    def main(args: Array[String]): Unit = {
        /*Actor1.start()
        Actor2.start()*/
        new Actor1().start();
        new Actor2().start();
    }
}

并发编程：

1、导入对应的jar包
2、编写类继承特质（Actor）
3、创建对象，启动线程   strat 方法

2、使用Actor进行通信

package com.qfedu

import scala.actors.Actor
import scala.actors.threadpool.TimeUnit

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 11:13
 * @Version 1.0
 */
object _03Actor进行通信 {

    object ActorSender extends Actor{
        // 发送消息
        override def act(): Unit = {
            // 发送给谁，就使用谁的对象
            while(true){
                ActorReceiver ! "Hello"
                // 每次发送完，休息3秒钟
                TimeUnit.SECONDS.sleep(3)
            }

        }
    }

    object ActorReceiver extends Actor{
        // 接收消息
        override def act(): Unit = {

            //接收消息固定的写法
            while(true){
                receive {
                    case msg:String => println("接收到消息："+msg)
                }
            }

        }
    }

    def main(args: Array[String]): Unit = {
        ActorReceiver.start();
        ActorSender.start()
    }
}

以上演示的就是通信的方式
我们可以使用三种符号发送消息：

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

要给谁发送消息

actor1 ! "你好"
发送消息给actor1

接收消息：使用receive 方法来接收消息，需要传递给它一个偏函数

{
   case 变量名：类型 => 业务处理
}

receive 方法只能接收一次消息，如果想持续的接收消息，可以使用while(true)

持续的接收消息，可以使用whle(true) ，但是有一些弊端：

1、如果当前的actor没有接收到消息，线程就会阻塞
2、如果有很多个actor,也有可能会阻塞线程
3、每次有新的消息，需要重新创建线程来处理
4、频繁的创建线程，销毁，切换，影响性能

解决方案：

loop + react 来 复用线程，提升效率，效率要比 while + receive 高很多。

 object ActorReceiver extends Actor{
        // 接收消息
        override def act(): Unit = {

            //接收消息固定的写法
            loop {
                react {
                    case msg:String => println("接收到消息："+msg)
                }
            }

        }
    }

以上发送消息都是String类型的消息，比较简单，消息是否可以自定义呢？肯定可以

package com.qfedu

import scala.actors.Actor

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 11:35
 * @Version 1.0
 */
object _04Actor之自定义消息类型 {
    // 自定义消息的类
    case class Message(id:Int,msg:String)
    case class ReplyMessage(msg:String,name:String);

    object MsgActor extends Actor {
        override def act(): Unit = {
            loop{
                react {
                    case Message(id,msg) => {
                        println(s"我收到消息了：${id},${msg}")
                        // 回复消息，直接使用sender 即可，这是一个固定的对象
                        sender ! ReplyMessage("我是黑猫，消息已收到","白猫")
                    }
                }
            }
        }
    }

    def main(args: Array[String]): Unit = {
        MsgActor.start()
        val replyMessage: Any = MsgActor !? Message(1, "呼叫黑猫")
        //println("收到回复："+replyMessage)
        println("收到回复："+replyMessage.asInstanceOf[ReplyMessage])
    }


}

发送消息的其他方式：

package com.qfedu

import scala.actors.threadpool.TimeUnit
import scala.actors.{Actor, Future}

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 11:35
 * @Version 1.0
 */
object _04Actor之自定义消息类型 {
    // 自定义消息的类
    case class Message(id:Int,msg:String)
    case class ReplyMessage(msg:String,name:String);

    object MsgActor extends Actor {
        override def act(): Unit = {
            loop{
                react {
                    case Message(id,msg) => {
                        println(s"我收到消息了：${id},${msg}")
                        // 回复消息，直接使用sender 即可，这是一个固定的对象
                        TimeUnit.SECONDS.sleep(5)
                        sender ! ReplyMessage("我是黑猫，消息已收到","白猫")
                    }
                }
            }
        }
    }

    def main(args: Array[String]): Unit = {
        MsgActor.start()
        // 目前发送的是一个同步的消息，并且有返回值
        // val replyMessage: Any = MsgActor !? Message(1, "呼叫黑猫")
        //println("收到回复："+replyMessage)
         //println("收到回复："+replyMessage.asInstanceOf[ReplyMessage])
        // 如何发送异步消息，并且也有返回值呢？
        val future: Future[Any] = MsgActor !! Message(1, "呼叫黑猫")
        // 什么时候能接收到消息呢
        println("验证异步消息，立即会打印")

        // 判断是否消息已经回来了，如果回来了，这个语句就执行结束
        while(!future.isSet){}

        // apply() 方法是用来接收消息的
        println(future.apply())

    }
}

三、高阶函数

在scala中，函数属于一等公民，它跟Int,String,Class 等都是出于同等的位置。就可以像其他变量一样可以进行传递和操作。
高阶函数是指使用其他函数作为参数、或者返回一个函数作为结果的函数。在Scala中函数是“一等公民”，所以允许定义高阶函数。这里的术语可能有点让人困惑，我们约定，使用函数值作为参数，或者返回值为函数值的“函数”和“方法”，均称之为“高阶函数”。
函数包含：
1、作为值进行传递
2、匿名函数
3、闭包
4、柯里化

作为值进行传递：

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 14:12
 * @Version 1.0
 */
object _05作为值传递 {



    def main(args: Array[String]): Unit = {
        // List(1,2,3...)  =>   *,**,***
        val list: List[Int] = (1 to 10).toList
        val func:Int => String = (num: Int) => "*" * num
        // 函数当做值进行传递了
        val newList: List[String] = list.map(func)
        println(newList)

        // 匿名函数   num => "*" * num
        val newList2: List[String] = list.map(num => "*" * num)
        //简化
        list.map("*" * _)
    }
}

柯里化：

在spark以及flink大量的使用了柯里化。

def func(x:Int,y:Int) =>  x + y  -->  def func(x:Int)(y:Int)

def func(x:Int)(y:Int)
以上这个怎么执行呢？
def func(x:Int) {
    (y:Int) =>  {x + y}
}

柯里化：将多个参数的方法或者函数转换为 多个一个一个的参数的方法，这个过程就叫做柯里化。

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 14:27
 * @Version 1.0
 */
object _06函数的柯里化 {

    // 计算两个数  这两个数可以相加 ，可以相减.....
    /*def calc_liangge(x:Double,y:Double) = {
        x + y;
    }*/
    def calc_liangge(x:Double,y:Double)(func_calc:(Double,Double) => Double) = {
        func_calc(x,y)
    }

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

        val d: Double = calc_liangge(10, 20) {
            (x, y) => x + y
        }
        println(d)
        println(calc_liangge(10,20)(_ * _ ))
        println(calc_liangge(10,20)(_ - _ ))
        println(calc_liangge(10,20)(_ + _ ))
    }
}

闭包：

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 14:36
 * @Version 1.0
 */
object _07闭包 {

    // 柯里化其实就是闭包
    def add2(x:Int)(y:Int) = {
        x + y
    }
    // 相当于
    def add3(x:Int) = {
        (y:Int) => x + y // 不就是闭包的写法吗？
    }

    def main(args: Array[String]): Unit = {
        val y = 10
        // 闭包其实是一个函数，只是这个函数它的返回值，依赖于函数之外的变量
        // 可以访问不在当前作用域范围内的函数
        val add = (x:Int) => { x + y }
        println(add(5))


    }
}

四、隐式转换以及隐式参数

隐式转换以及隐式参数是scala中独有的特点，其他语言没有。可以使用隐式转换丰富目前现有的类（对类的功能进行拓展）

隐式转换：

是使用implicit关键字声明的带有单个参数的方法，它可以自动被调用。将某种数据类型转换为另一种。

package com.qfedu

import java.io.File
import scala.io.Source

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 14:48
 * @Version 1.0
 */

class RichFile(val file:File){
    def read() = {
        // 获取文件，将文件变为string类型
        Source.fromFile(file).mkString
    }
}
object RichFile {
    // 定义隐式转换的方法
    implicit def file2RichFile(file:File) = new RichFile(file)
}

object _08隐式转换 {

    def main(args: Array[String]): Unit = {
        val file = new File("data/1.txt")
        // file 对象是没有读取文件内容的功能，怎么让它拥有这个功能？
        // 编写完成后，需要导入
        import RichFile.file2RichFile
        println(file.read());
    }
}

1、在object中，定义隐式转换的方法
2、在需要的隐式转换的地方，引入隐式转换（import）
3、自动调用隐式转换的方法
     // 调用的时机：对象调用类中不存在的方法或者成员的时候，编译器自动回将对象进行隐式转换。

这个里面，我们需要手动的导入隐式转换，能够自动的导入呢？可以，将隐式转换的方法，写入main所在的object中即可实现自动导入。

package com.qfedu

import java.io.File
import scala.io.Source

class RichFile(val file:File){
    def read() = {
        // 获取文件，将文件变为string类型
        Source.fromFile(file).mkString
    }
}
/*object RichFile {
    // 定义隐式转换的方法
    implicit def file2RichFile(file:File) = new RichFile(file)
}*/

object _08隐式转换 {

    implicit def file2RichFile(file:File) = new RichFile(file)
    def main(args: Array[String]): Unit = {
        val file = new File("data/1.txt")
        // file 对象是没有读取文件内容的功能，怎么让它拥有这个功能？
        println(file.read());
    }
}

隐式参数：

在方法参数列表中，出现了 implicit 关键字。

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 15:07
 * @Version 1.0
 */
object _09隐式参数 {

    // 传递一个书名，在书名的左右两边 《》
    def quote(book:String)(implicit delimiter:(String,String)) = {

        delimiter._1 + book + delimiter._2
    }
    /*object  ImplicitParam {
        implicit val DEFAULT_BOOK = ("<<",">>")
    }*/

    def main(args: Array[String]): Unit = {
        //import ImplicitParam.DEFAULT_BOOK
        implicit val DEFAULT_BOOK = ("<<",">>")
        println(quote("金花的秘密"))
    }
}

五、昨日的作业

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 15:34
 * @Version 1.0
 */
object _10HomeWork01 {

    // 编写枚举，描述RGB 8 个角

    /**
     *
     *   编程中：普通的颜色 是 6 位 16进制的数字表示的
     *     #FFFFFF  白色
     *     #000000  黑色
     *   如果你发现颜色是 8 位 16进制的数字表示的：
     *     是因为加入了透明度
     */
    object RGB extends Enumeration {
        val RED =Value(0xff0000,"red")
        val BLACK =Value(0x000000,"black")
        val GREEN =Value(0x00ff00,"green")
        val YELLOW =Value(0xffff00,"yellow")
    }
    def main(args: Array[String]): Unit = {

        println(RGB.RED);
    }
}

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 15:42
 * @Version 1.0
 */
object _11HomeWork02 {

    def swap(x:(Int,Int)) = {
        x match {
            case (a,b) => (b,a)
            case _ => (0,0)
        }
    }

    def main(args: Array[String]): Unit = {
        println(swap((8,9)))
    }
}

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 15:44
 * @Version 1.0
 */
object _12HomeWork03 {

    // 交换数组中的前两个值
    def swap(arr:Array[String]) = {
        arr match {
            case Array(x,y,items @ _*) => Array(y,x) ++ items
        }
    }

    def main(args: Array[String]): Unit = {
        swap(Array("a","b","c","d")).foreach(print(_))
    }
}

package com.qfedu

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 15:51
 * @Version 1.0
 */
object _13HomeWork04 {

    // 编写一个通用的函数，计算所有叶子节点的元素之和
    abstract class BinaryTree
    case class Leaf(num:Int) extends BinaryTree
    case class Node(left:BinaryTree,right:BinaryTree)  extends BinaryTree

    // 此处使用 了递归思想
    def getNode(x:BinaryTree):Int = {
        x match {
            case Leaf(x) => x
            case Node(left,right) => getNode(left) + getNode(right)
        }
    }
    def main(args: Array[String]): Unit = {

        val node: Node = Node(Leaf(10),Leaf(20));
        println(getNode(node))

    }
}

六、Akka

1、介绍

Akka是一个用于高并发，分布式，可扩展的基于事件驱动的工具包（Actor）.Akka是使用的scala的开发库。
Akka是可以进行多线程，可以发送消息的技术。

2、本地通信案例：

<dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.11.8</version>
        </dependency>
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor_2.11</artifactId>
            <version>2.5.21</version>
        </dependency>
    </dependencies>

代码：

package com.qfedu

import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

/**
 * @Author laoyan 
 * @Description TODO
 * @Date 2022/5/6 16:19
 * @Version 1.0
 */
case class SubmitMessage(msg:String)
case class SuccessSubmitMessage(msg:String)

object SenderActor extends Actor {
    override def receive: Receive = {
        case "start" => {
            // 此处需要给ReceiverActor 对象发送消息
            val receiverActor: ActorSelection = this.context.actorSelection("akka://localAkkaDemo/user/receiverActor")
            receiverActor ! SubmitMessage("请务必炸掉001号高地！")
        }
        case SuccessSubmitMessage(msg) => {
            println("收到对方的回复："+msg)
        }
    }
}
object ReceiverActor extends Actor {
    override def receive: Receive = {
        //case "start" => println("开始")
        case SubmitMessage(msg) => println("收到领导的指示："+msg)
            sender ! SuccessSubmitMessage("消息收到，务必完成任务")
    }
}
object LocalAkka {

    def main(args: Array[String]): Unit = {
        // 给 sendActor  发送一个消息：start
        // 在 senderActor 里面发送消息 给 ReceiverActor
        // ReceiverActor 接收消息，打印，返回消息给 senderActor
        // senderActor 接收到以后，打印接收的消息
        // 这个是核心组价
        val actorSystem: ActorSystem = ActorSystem("localAkkaDemo", ConfigFactory.load())
        // 创建一个发送的对象
        val senderActor: ActorRef = actorSystem.actorOf(Props(SenderActor), "senderActor")
        // 创建一个接收的对象
        val receiverActor: ActorRef = actorSystem.actorOf(Props(ReceiverActor), "receiverActor")

        senderActor ! "start"
    }
}

Day06-Scala第五天 - 图4

1、ActorSystem:在Akka中，ActorSystem 是一个非常重要的组件，是一个单例的对象，通过这个对象可以创建多个Actor,并且还可以监控这些Actor,是一个王者。
2、Actor 通过 ActorSystem 创建的，里面可以直接使用 receive 方法接收消息，直接处理。
3、path路径：akka://localAkkaDemo/user/receiverActor
分为两种：
本地路径： akka://actorSystem的名字/user/Actor的名字 
远程的路径： akka.tcp://IP地址@端口号/user/Actor的名字

Scala中遇到的问题及解决方案：

Exception in thread "main" java.lang.NoClassDefFoundError: scala/Product$class
    at org.apache.spark.SparkConf$DeprecatedConfig.<init>(SparkConf.scala:723)
    at org.apache.spark.SparkConf$.<init>(SparkConf.scala:571)
    at org.apache.spark.SparkConf$.<clinit>(SparkConf.scala)
    at org.apache.spark.SparkConf.set(SparkConf.scala:92)
    at org.apache.spark.SparkConf.set(SparkConf.scala:81)
    at org.apache.spark.SparkConf.setMaster(SparkConf.scala:113)
    at cn.spark.WordCount$.main(WordCount.scala:23)
    at cn.spark.WordCount.main(WordCount.scala)
Caused by: java.lang.ClassNotFoundException: scala.Product$class
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    ... 8 more

Process finished with exit code 1

这种错误是因为 Scala 版本不对，要使用2.11.x 的版本

如果遇到了这个问题：

Exception in thread "main" java.lang.NoSuchMethodError: scala.Predef$.refArrayOps([Ljava/lang/Object;)Lscala/collection/mutable/ArrayOps;
    at org.apache.spark.util.Utils$.getCallSite(Utils.scala:1434)
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:76)
    at cn.spark.WordCount$.main(WordCount.scala:24)
    at cn.spark.WordCount.main(WordCount.scala)

Process finished with exit code 1

idea配置的版本和本机版本不一致。
如果出现这个异常：

scala/Predef$any2stringadd$

提升版本至：2.18.x