V2AS问路
大数据学习day16------第三阶段-----scala04--------1. 模式匹配和样例类 2 Akka通信框架
阅读原文时间：2023年07月10日阅读：2

1. 模式匹配和样例类

　　Scala有一个十分强大的模式匹配机制，可以应用到很多场合：如switch语句、类型检查等。并且Scala还提供了样例类，对模式匹配进行了优化，可以快速进行匹配

1.1 模式匹配

1.1.1 匹配字符串

object CaseDemo1 extends App { // 继承App特质（接口）后就不需要写main方法了
val arr = Array("A","B","C")
val name = arr(Random.nextInt(arr.length))
name match{
case "A" => println("我是A")
case "B" => println("我是B")
case _ => println("我可以匹配一切")
}
}

1.1.2 匹配类型

object CaseDemo02 extends App {
val arr = Array("hello", 1, 2.0, CaseDemo02)
val v = arr(Random.nextInt(arr.length))
v match{
case v:String => println("String:"+v)
case v:Int => println("Int:"+v)
case v:Double => println("Double:"+v)
case _ => throw new Exception("not match exception")
}
}

注意：case后面接收的一定是一个值，不能是参数类型，自己犯的错误：

case String => println("String:"+v)

报错如下：

1.1.3 匹配数组、集合、元组

数组

object CaseDemo03 extends App {
// 不可变数组
val arr = Array(1, 3, 6)
arr match {
case Array(1, x, y) => println(x + " " + y)
case Array(0) => println("only 0")
case Array(0, _*) => println("0 …")
case _ => println("something else")
}
}

集合

// 不可变集合
val lst = List(3, -1)
lst match {
case 0 :: Nil => println("only 0")
case x :: y :: Nil => println(s"x: $x y: $y") case 0 :: tail => println("0 …")
case _ => println("something else")

注意：s 底层是一个方法，其是一个简单的值插入器，它可以通过${}的形式从任意表达式中取出其字符串形式的值，上面因为就一个参数，所以可以省略{}，若是上式改成如下

其打印的结果为x: 3+y  y: -1 所以若想取多个参数的值必须使用${}

知识点补充：

  • “+:”和"::" 都是用来拼接集合的
  • 一个集合分为头部和尾部（第一个元素为头，其余元素组成的集合为尾）
  • Nil代表空

注意：  在Scala中列表要么为空（Nil表示空列表）要么是一个head元素加上一个tail列表。9 :: List(5, 2)  ":: "操作符是将给定的头和尾创建一个新的列表

　　　 " :: "操作符是右结合的，如9 :: 5 :: 2 :: Nil相当于 9 :: (5 :: (2 :: Nil))

1.2 样例类

　　在Scala中样例是一种特殊的类，可用于模式匹配。case class是多例的，后面要跟构造参数，case object是单例的。其内部直接实现了序列化并且创建对象不需要new，直接写类名即可

// case class主要是为了封装数据和模式匹配
case class SubmitTask(id: String, name: String)
case class HeartBeat(time: Long)
//case class Heart() //case class一定要有参数，此处是空参数
case object CheckTimeOutTask // 不能有参数

object CaseDemo04 extends App{
val arr = Array(CheckTimeOutTask, HeartBeat(12333), SubmitTask("0001", "task-0001"))

arr(Random.nextInt(arr.length)) match {
case SubmitTask(id, name) => {
println(s"$id, $name")
}
case HeartBeat(time) => {
println(time)
}
case CheckTimeOutTask => {
println("check")
}
}
}

补充：模式匹配也可以匹配自定义的类型，那么自定义的类型和case class/object有哪些区别呢？

　　case class创建对象不用new，实现了序列化接口、比自定义的类型匹配更加高效

1.3 Option类型

　　在Scala中Option类型样例类用来表示可能存在或也可能不存在的值(Option的子类有Some和None)。Some包装了某个值，None表示没有值

object OptionDemo {
def main(args: Array[String]) {
val map = Map("a" -> 1, "b" -> 2)
val v = map.get("b") match {
case Some(i) => i
case None => 0
}
println(v)
//更好的方式
val v1 = map.getOrElse("c", 0)
println(v1)
}
}

1.4 偏函数

　　被包在花括号内没有match的一组case语句是一个偏函数（类型是PartialFunction，有case，无match），它是PartialFunction[A, B]的一个实例，A代表参数类型，B代表返回类型，常用作输入模式匹配

object PartialFuncDemo {
　// 使用偏函数
def func1: PartialFunction[String, Int] = {
case "one" => 1
case "two" => 2
case _ => -1
}
　// 不使用偏函数，也能实现这个功能，只是使用偏函数简洁些
def func2(num: String) : Int = num match {
case "one" => 1
case "two" => 2
case _ => -1
}

def main(args: Array[String]) {
println(func1("one"))
println(func2("one"))
}
}

2. Akka通信框架

2.0 背景 ：

　在JVM领域，比较流行的、高效的、通信框架有哪些

  • Netty: 使用用Java编写，基于Java Nio、事件驱动的并发编程框架，可以实现多种网络编程协议（Http、TCP、UDP）【Hadoop、Spark】

  • Akka：用scala编程的一个异步通信框架、基于Actor编程模型，通过发送消息实现并发，支持Scala和Java的API【Spark、Flink】

  • Mina：用Java编程的NIO通信框架

2.1 Akka简介：　

　Akka是J在AVA虚拟机JVM平台上构建高并发、分布式和容错应用的工具包。Akka用Scala语言写成，同时提供了Scala和JAVA的开发接口。

Akka处理并发的方法基于Actor模型。在Akka里，Actor之间通信的唯一机制就是消息传递。

2.2 Akka的特点 ：

  • Simple Concurrency & Distribution单并发性&分布式

  • Resilient by Design （可恢复的、弹性的）

  • High Performance高效

  • Elastic & Decentralized 弹性、去中心的（分散）

  • Extensible可扩展

    **2.3 Actors：（来源：https://www.jianshu.com/p/449850aa8e82）**

    　　Actor模型是一个概念模型，用于处理并发计算。它定义了一系列系统组件应该如何动作和交互的通用规则。

    　　 一个Actor指的是一个最基本的计算单元。它能接收一个消息并且基于其执行计算。

    　　这个理念很像面向对象语言，一个对象接收一条消息（方法调用），然后根据接收的消息做事（调用了哪个方法）。

    　　Actors一大重要特征在于actors之间相互隔离，它们并不互相共享内存。这点区别于上述的对象。也就是说，一个actor能维持一个私有的状态，并且这个状态不可能被另一个actor所改变。

    　　actor是用来收发消息的，一个Actor就是一个类的实例。可以创建多个Actor实现并发，根据老师和学生的例子（老师和学生都是Actor），所有的Actor都要管理者（教学总监），创建Actor和管理Actor，以后发送的消息是带类型的，可以根据消息类型进行匹配，处理不同的逻辑。如果用Scala实现，用到的知识点：case class、case object

    2.4 Akka的通信流程

3 AKKA案例

需求：实现上述akka的通信流程

3.1 创建一个master，并向自己发送消息

过程：

  • 创建master类（因为是多实例的），并让其实现Actor特质（让其与Akka关联起来）
  • 重写Actor特质中的方法（receive），用来接收消息，其会一直等着接收消息，此方法的返回值的类型为偏函数，所以函数体中可以用偏函数的形式去匹配
  • 创建master的伴生对象
  • 在伴生对象中创建actor（通过ActorSystem）
  • 使用创建的actor发送消息

代码如下：

// Actor编程模型进行通信，需要让其与AKKA发生点关系（此处实现Actor特质）
class Master extends Actor {
override def receive: Receive = {
case "hah" => {
println("receive a hello msg")
}
}
}

object Master{
def main(args: Array[String]): Unit = {
val host = "localhost"
val port = 8888
val configStr =
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider" // 负责通信的核心类，有必要可以自己定义
|akka.remote.netty.tcp.hostname = "$host"
|akka.remote.netty.tcp.port = "$port"
""".stripMargin // 此方法负责切割
val conf = ConfigFactory.parseString(configStr)
val actorSystem = ActorSystem.apply("MASTER_ACTOR_SYSTEM", conf)
// 通过ActorSystem对象常见Actor（通过反射指定类型的Actor的实例）
val masterActor = actorSystem.actorOf(Props[Master], name = "MASTER_ACTOR")
// 向Actor发送消息
// ! 表示的是发送异步消息
masterActor.!(message = "hah")
}
}

3.2  创建一个worker，并向自己发送消息，过程同master

代码如下：

class Worker extends Actor {
// 重写用于接收消息的方法
override def receive: Receive = {
case "liZhi" => println("z这个世界会好吗")
}
}

object Worker{
def main(args: Array[String]): Unit = {
val host = "localhost"
val port = 7777
val configStr =
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider" // 负责通信的核心类，有必要可以自己定义
|akka.remote.netty.tcp.hostname = "$host"
|akka.remote.netty.tcp.port = "$port"
""".stripMargin // 此方法负责切割
val conf = ConfigFactory.parseString(configStr)
val workerActorSystem = ActorSystem("WORKER_ACTOR_SYSTEM", conf)
val workerActor: ActorRef = workerActorSystem.actorOf(Props[Worker], name = "WORKER_ACTOR")
workerActor.!("liZhi")
}
}

以上都是自己给自己发消息，如何是worker和master彼此间互发消息呢？

worker去连接master（老大）

3.3 Worker向Master建立连接并发送消息（测试）

以下只列出代码的改变处：

Worker部分：

Master部分（写接收worker发送消息的代码即可）

3.4 最终版

思路

  • worker在接收消息前，构造方法之后（preStart方法），创建与Master的连接，并发送注册信息（RegisterWorker:注册消息应该包括自己的id，内存情况，以及自己使用的cpu个数）

    　　　　发送的注册信息用case class（后面带有信息的传递都使用case class，此处创建了Messages类），因为case class既可以封装数据，又可以进行模式匹配，还实现了序列化接口

  • Master处定义receive方法，用来接收worker处发送过来的注册消息，并将该消息保存到内存中，然后向worker回复注册成功消息（RegisteredWorker）

    　　（1）定义一个包含所有注册信息的类（此处自己定义的是WorkInfo）

    　　 （2）创建一个map，并将该信息以worker的id为key，workerInfo类为value存入该map（即存入了内存）

  • worker处接收注册成功的信息，并启动一个定时器，定时向Master发送心跳信息

    　　　由于worker端的定时器不能将定时信息发送给Master处（worker与Master处于两个进程中），所以其先将心跳信息（Heartbeat）发送给自己，然后worker自己接收到这个信息再发送给Master

  • Master匹配并接收Worker发送过来的心跳汇报消息（此处涉及每次心跳时间的更新）

  • Master自己创建一个定时器，定期检查是否有deadWorker，用来剔除该worker

代码：

worker

package com._51doit.akka.rpc

import java.util.UUID

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

/**
* Worker Actor最好在构造方法执行之后，receive方法之前，向Master建立连接
*/
class Worker extends Actor {
var masterRef: ActorSelection = _

val WORKER_ID = UUID.randomUUID().toString
// 生命周期方法（一定并且按一定顺序执行的方法）
// 在构造方法之后，receive方法之前，执行一次preStart
override def preStart(): Unit = {
// Worker向Master建立网络连接，得到一个master代理对象
masterRef = context.actorSelection("akka.tcp://MASTER_ACTOR_SYSTEM@localhost:8888/user/MASTER_ACTOR")
// //Worker向Master发送注册的信息
masterRef ! RegisterWorker(WORKER_ID, 4096, 8)
}
// 重写用于接收消息的方法
override def receive: Receive = {
//Master反馈给Worker的消息
case RegisteredWorker => {

  //导入隐式转换  
  import context.dispatcher  
  //启动一个定时器，定期向Master发送心跳，使用Akka框架封装的定时器  
  //定期给自己发送消息，然后再给Master发送心跳  
  //参数依次为第一次的延迟时间，多少时间执行一次，消息发送给谁（此处找不到master，发送给masterRef代理对象也不行，），发送的消息  
  context.system.scheduler.schedule(0 millisecond, 5000 millisecond, self, SendHeartbeat)  
}  
//自己给自己发送的消息  
case SendHeartbeat => {

  //可以进行一些逻辑判断  
  //向Master发送心跳消息  
  masterRef ! Heartbeat(WORKER\_ID)  
}

}
}

object Worker{
def main(args: Array[String]): Unit = {
val host = "localhost"
val port = 7777
val configStr =
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider" // 负责通信的核心类，有必要可以自己定义
|akka.remote.netty.tcp.hostname = "$host"
|akka.remote.netty.tcp.port = "$port"
""".stripMargin // 此方法负责切割
val conf = ConfigFactory.parseString(configStr)
val workerActorSystem = ActorSystem("WORKER_ACTOR_SYSTEM", conf)
val workerActor: ActorRef = workerActorSystem.actorOf(Props[Worker], name = "WORKER_ACTOR")
workerActor.!("liZhi")
}
}

Master

package com._51doit.akka.rpc

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._

import scala.collection.mutable

// Actor编程模型进行通信，需要让其与AKKA发生点关系（此处实现Actor特质）
class Master extends Actor {
// 定义一个map，用来接收数据
val id2Worker = new mutable.HashMap[String, WorkerInfo]()

// 在preStart中启动定时器，定期检查超市的Worker，然后剔除
override def preStart(): Unit = {
import context.dispatcher

context.system.scheduler.schedule(0 millisecond, 10000 millisecond, self, CheckTimeOutWorker)

}
override def receive: Receive = {
// Master匹配并接收Worker发送过来的注册消息
case RegisterWorker(id, memory, cores) => {
//将数据封装起来，保存到内存中
val workerInfo: WorkerInfo = new WorkerInfo(id,memory,cores)
id2Worker(id) = workerInfo
//向Worker反馈一个注册成功的消息
sender() ! RegisteredWorker
}
// Master匹配并接收Worker发送过来的心跳汇报消息
case Heartbeat(workerId) => {
// 根据workId去map中查找相对应的WorkerInfo
if(id2Worker.contains(workerId)){
//根据ID取出WorkerInfo
val workerInfo = id2Worker(workerId)
//获取当前时间
val currentTime = System.currentTimeMillis()
//更新最近一次心跳时间
workerInfo.lastUpdateTime = currentTime
}
}
// 匹配定时器发送的内容，用于提出超时的worker
case CheckTimeOutWorker => {
val currentTime = System.currentTimeMillis()
// 取出map中的值，并计算出超时的worker
val values: Iterable[WorkerInfo] = id2Worker.values
val deadWorkers: Iterable[WorkerInfo] = values.filter(value => currentTime - value.lastUpdateTime > 10000)
// 移除所有超时的worker
deadWorkers.foreach(dw => id2Worker -= dw.id)
println("current alive worker is : " + id2Worker.size)
}
}
}

object Master {
def main(args: Array[String]): Unit = {
val host = "localhost"
val port = 8888
val configStr =
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider" // 负责通信的核心类，有必要可以自己定义
|akka.remote.netty.tcp.hostname = "$host"
|akka.remote.netty.tcp.port = "$port"
""".stripMargin // 此方法负责切割
val conf = ConfigFactory.parseString(configStr)
val actorSystem = ActorSystem.apply("MASTER_ACTOR_SYSTEM", conf)
// 通过ActorSystem对象常见Actor（通过反射指定类型的Actor的实例）
val masterActor = actorSystem.actorOf(Props[Master], name = "MASTER_ACTOR")
// 向Actor发送消息
// ! 表示的是发送异步消息
masterActor.!(message = "hah")
}
}

注意：当创建了master actor对象时，AKKA框架底层会自动调用声明周期方法

Message

package com._51doit.akka.rpc

// Worker发送给Master的注册消息，case class默认实现了序列化接口
case class RegisterWorker(id: String, memory: Int, cores: Int)

//Master发送给Worker注册成功的消息
case object RegisteredWorker

//Worker发送给Master的心跳消息
case class Heartbeat(workerId: String)

//Worker自己给自己发送的消息(内部消息)
case object SendHeartbeat

//Master发送给自己的消息，用于检查超时的Worker
case object CheckTimeOutWorker

WorkInfo

package com._51doit.akka.rpc

class WorkerInfo(val id: String, var memory: Int, var cores: Int) {
var lastUpdateTime: Long = _
　 // 重写toString方法
override def toString = s"WorkerInfo($id, $memory, $cores)"
}

手机扫一扫

移动阅读更方便

阿里云服务器
腾讯云服务器
七牛云服务器

你可能感兴趣的文章

Lagom 官方文档之随手记