大数据技术之_16_Scala学习_11_客户信息管理系统+并发编程模型 Akka+Akka 网络编程-小黄鸡客服案例+Akka 网络编程-Spark Master Worker 进程通讯项目

大数据技术之_16_Scala学习_11

第十五章客户信息管理系统
- 15.1 项目的开发流程
- 15.2 项目的需求分析
- 15.3 项目的界面
- 15.4 项目的设计-程序框架图
- 15.5 项目的功能实现
- - 15.5.1 完成 Customer 类
  - 15.5.2 完成显示主菜单和退出软件功能
  - 15.5.3 完成显示客户列表的功能
  - 15.5.4 完成添加客户的功能
  - 15.5.5 完成删除客户的功能
  - 15.5.6 完善退出确认功能
  - 15.5.7 完善删除确认功能
  - 15.5.8 完成修改客户的功能
第十六章并发编程模型 Akka
- 16.1 Akka 的介绍
- 16.2 Actor 模型用于解决什么问题
- 16.3 Akka 中 Actor 模型详解
- 16.4 Actor 模型工作机制说明
- 16.5 Actor 模型应用实例
- - 16.5.1 Actor 自我通讯
  - 16.5.2 Actor 之间通讯
- 16.7 Akka 网络编程
- - 16.7.1 Akka 网络编程基本介绍
  - 16.7.2 协议(tcp/ip)
  - 16.7.3 OSI 与 Tcp/ip 参考模型
  - 16.7.4 ip 地址
  - 16.7.5 端口(port)
- 16.8 Akka 网络编程-小黄鸡客服案例
- - 16.8.1 需求分析 + 界面设计
  - 16.8.2 程序框架图
  - 16.8.3 功能实现
- 16.9 Akka 网络编程-Spark Master Worker 进程通讯项目
- - 16.9.1 项目意义
  - 16.9.2 项目需求分析
  - 16.9.3 项目界面设计
  - 16.9.4 实现功能 1-Worker 完成注册
  - 16.9.5 实现功能 2-Worker 定时发送心跳
  - 16.9.6 实现功能 3-Master 启动定时任务，定时检测注册的 Worker
  - 16.9.7 实现功能 4-Master,Worker 的启动参数运行时指定
  - 16.9.8 Master Worker 进行分布式部署

第十五章客户信息管理系统

15.1 项目的开发流程

15.2 项目的需求分析

模拟实现基于文本界面的《客户信息管理软件》。
该软件 scala 能够实现对客户对象的插入、修改、删除、显示、查询（用 ArrayBuffer 或者 ListBuffer 实现），并能够打印客户明细表。

15.3 项目的界面

主界面

添加客户

修改客户

删除客户

客户列表

15.4 项目的设计-程序框架图

程序框架图：设计系统有多少个文件，以及文件之间的调用关系，可以帮助程序员实现模块的设计（清晰），便于程序员之间对项目交流分析。【业务优化，设计方案】

15.5 项目的功能实现

15.5.1 完成 Customer 类

根据需求文档或者页面，写出 Customer 类
Customer.scala

package com.atguigu.chapter15.customercrm.beanclass Customer {// 属性var id: Int = _var name: String = _var gender: Char = _var age: Short = _var tel: String = _var email: String = _// 辅助构造器def this(id: Int, name: String, gender: Char, age: Short, tel: String, email: String) {thisthis.id = idthis.name = namethis.gender = genderthis.age = agethis.tel = telthis.email = email}
}

15.5.2 完成显示主菜单和退出软件功能

CustomerView.scala 功能分析：
1. 将主菜单的显示放入到 while
2. 用户可以根据输入，选择自己的操作
3. 如果输入5退出
CustomerView.scala

package com.atguigu.chapter15.customercrm.viewimport scala.io.StdInclass CustomerView {// 定义一个循环变量，控制是否退出var loop = true// 定义一个 key 用于接收用户输入的选项var key = ' 'def mainMenu(): Unit = {do {println("-----------------客户信息管理软件-----------------")println("                  1 添 加 客 户                  ")println("                  2 修 改 客 户                  ")println("                  3 删 除 客 户                  ")println("                  4 客 户 列 表                  ")println("                  5 退       出                  ")println("                  请选择(1-5):                   ")key = StdIn.readChar()key match {case '1' => println("添 加 客 户")case '2' => println("修 改 客 户")case '3' => println("删 除 客 户")case '4' => println("客 户 列 表")case '5' => this.loop = false}} while (loop)println("你退出了系统...")}
}

示例代码如下：

package com.atguigu.chapter15.customercrm.appimport com.atguigu.chapter15.customercrm.view.CustomerViewobject CustomerCrm {def main(args: Array[String]): Unit = {new CustomerView().mainMenu()}
}

15.5.3 完成显示客户列表的功能

CustomerView.scala 功能分析：
1. 接收4，显示客户列表
2. 调用 CustomerService 的方法 list
3. 需要一个 CustomerService 对象(属性)
CustomerService.sacla 功能分析：
1. 编写一个方法 list，返回当前系统有哪些客户
2. 客户放在哪？–> 内存 --> 可变集合 --> ArrayBuffer
1、在 Customer.sacla 中重写 toString 方法

  override def toString: String = {this.id + "\t\t" + this.name + "\t\t" + this.gender + "\t\t" + this.age + "\t\t" + this.tel + "\t\t" + this.email}

2、在 CustomerService.scala 中编写一个方法 list，返回当前系统有哪些客户

class CustomerService {// customers 是存放客户用的，为了方便测试，我们先进行初始化val customers = ArrayBuffer(new Customer(1, "tom", '男', 20, "110", "tom@sohu.com"))// 查询客户列表的方法def list(): ArrayBuffer[Customer] = {this.customers}
}

3、在 CustomerView.scala 中调用 CustomerService 的方法 list

  val customerService = new CustomerService()/*
---------------------------客户列表---------------------------
编号  姓名       性别    年龄   电话            邮箱1    张三       男      30     010-56253825   abc@email.com2    李四       女      23     010-56253825    lisi@ibm.com3    王芳       女      26     010-56253825   wang@163.com
-------------------------客户列表完成-------------------------*/def list(): Unit = {println()println("---------------------------客户列表---------------------------")println("编号\t\t姓名\t\t性别\t\t年龄\t\t电话\t\t邮箱")// 遍历// 调用 CustomerService 的方法 listval customers = customerService.list()for (customer <- customers) {// 方式一：输出// println(customer.id + "\t\t" + ...)// 方式二：重写 Customer 的 toString 方法，返回信息，并且格式化println(customer)}println("-------------------------客户列表完成-------------------------")}

15.5.4 完成添加客户的功能

CustomerView.scala 功能分析：
1. 接收客户的信息，并封装成对应的 Customer 对象
2. 调用 CustomerService 的方法 add
CustomerService.sacla 功能分析：
1. 编写一个方法 add，接收一个 Customer 对象
2. 加入到 ArrayBuffer 中
3. 规定：以添加客户是第几个作为它的 id
1、在 Customer.sacla 中添加一个新的辅助构造器(没有id属性)

  // 辅助构造器(没有id属性)def this(name: String, gender: Char, age: Short, tel: String, email: String) {thisthis.name = namethis.gender = genderthis.age = agethis.tel = telthis.email = email}

2、在 CustomerService.scala 中编写一个方法 add，接收一个 Customer 对象，并设置 id 后再加入到 ArrayBuffer 中

  // 用于设置用户 idvar customerNum = 1// 添加客户的方法def add(customer: Customer): Boolean = {// 设置 idcustomerNum += 1customer.id = customerNum// 加入到 ArrayBuffer 中customers.append(customer)true}

3、在 CustomerView.scala 中调用 CustomerService 的方法 add

  /*
---------------------添加客户---------------------
姓名：张三
性别：男
年龄：30
电话：010-56253825
邮箱：zhang@abc.com
---------------------添加完成---------------------*/def add(): Unit = {println()println("---------------------添加客户---------------------")println("姓名：")val name = StdIn.readLine()println("性别：")val gender = StdIn.readChar()println("年龄：")val age = StdIn.readShort()println("电话：")val tel = StdIn.readLine()println("邮箱：")val email = StdIn.readLine()// 封装对象val customer = new Customer(name, gender, age, tel, email)// 调用 CustomerService 的方法 addcustomerService.add(customer)println("---------------------添加完成---------------------")}

15.5.5 完成删除客户的功能

CustomerView.scala 功能分析：
1. 接收客户 id，准备删除
2. 调用 CustomerService 的 del(id)
CustomerService.sacla 功能分析：
1. 编写一个方法 del，接收一个 id，先去调用另一个方法 findIndexById，判断
2. 编写一个方法 findIndexById（因为我们的 ArrayBuffer 索引和 id 并不是对应的）
3. 如果发现有，则删除，如果没有就返回 false
1、在 CustomerService.scala 中编写一个方法 del，接收一个 id，先去调用另一个方法 findIndexById，判断

  // 先根据 id 查找 用户的 indexdef findIndexById(id: Int): Int = {// 先假定一个索引，默认 -1，如果找到就改成对应的，如果没有找到就返回 -1var index = -1// 遍历 ArrayBufferbreakable {for (i <- 0 until customers.length) {if (customers(i).id == id) {index = ibreak()}}}index}// 再根据 id 删除用户def del(id: Int): Boolean = {val index = findIndexById(id)if (index != -1) {customers.remove(index)true} else {false}}

2、在 CustomerView.scala 中接收客户 id，调用 CustomerService 的 del(id)

  /*
---------------------删除客户---------------------
请选择待删除客户编号(-1退出)：1
确认是否删除(Y/N)：y
---------------------删除完成---------------------*/def del(): Unit = {println()println("---------------------删除客户---------------------")println("请选择待删除客户编号(-1退出)：")val id = StdIn.readInt()if (id == -1) {println("---------------------删除没有完成---------------------")return}println("确认是否删除(Y/N)：")val choice = StdIn.readChar().toLowerif (choice == 'y') {if (customerService.del(id)) {println("---------------------删除完成---------------------")return}}println("---------------------删除没有完成---------------------")}

15.5.6 完善退出确认功能

功能说明：
要求用户在退出时提示 “确认是否退出(Y/N)：”，用户必须输入y/n，否则循环提示。且输入为y时，退出系统；输入为n时，不退出系统。
1、在 CustomerView.scala 中定义一个方法 isOut，并修改 key 所对应的函数。

  // 要求用户在退出时提示"确认是否退出(Y/N)："，用户必须输入y/n，否则循环提示。且输入为y时，退出系统；输入为n时，不退出系统。def isOut(): Unit = {println()println("确认是否退出(Y/N)：")key = StdIn.readChar().toLowerkey match {case 'y' => this.loop = falsecase 'n' => this.loop = truecase _ => isOut()}}

15.5.7 完善删除确认功能

功能说明：
要求用户在删除确认时提示 “确认是否删除(Y/N)：”，用户必须输入y/n，否则循环提示。
1、在 CustomerView.scala 中，修改 del() 方法即可

  /*
---------------------删除客户---------------------
请选择待删除客户编号(-1退出)：1
确认是否删除(Y/N)：y
---------------------删除完成---------------------*/def del(): Unit = {println()println("---------------------删除客户---------------------")println("请选择待删除客户编号(-1退出)：")val id = StdIn.readInt()if (id == -1) {println("---------------------删除没有完成---------------------")return}println("确认是否删除(Y/N)：")var choice = ' '// 要求用户在删除确认时提示 "确认是否删除(Y/N)："，用户必须输入y/n，否则循环提示。breakable {do {choice = StdIn.readChar().toLowerif (choice == 'y' || choice == 'n') {break()}println("确认是否删除(Y/N)：")} while (true)}if (choice == 'y') {if (customerService.del(id)) {println("---------------------删除完成---------------------")return}}println("---------------------删除没有完成---------------------")}

15.5.8 完成修改客户的功能

1、在 CustomerService.scala 中定义一个方法根据 id 修改用户（更新用户）的方法和 // 根据 id 查找用户信息的方法

  // 根据 id 查找用户信息def findCustomerById(id: Int): Customer = {val index = findIndexById(id)if (index != -1) {customers(index)} else {null}}// 根据 id 修改用户（更新用户）def update(id: Int, customer: Customer): Boolean = {val index = findIndexById(id)customers.update(index, customer)true}

2、在 CustomerView.scala 中定义一个方法 update

  /*
---------------------修改客户---------------------
请选择待修改客户编号(-1退出)：1
姓名(张三)：<直接回车表示不修改>
性别(男)：
年龄(30)：
电话(010-56253825)：
邮箱(zhang@abc.com)：zsan@abc.com
---------------------修改完成---------------------*/def update(): Unit = {println()println("---------------------修改客户---------------------")println("请选择待修改客户编号(-1退出)：")var id = StdIn.readInt()if (id == -1) {println("---------------------修改没有完成---------------------")return}val customer = customerService.findCustomerById(id)if (customer == null) {println("---------------------修改没有完成---------------------")return}var name = customer.nameprint(s"姓名(${name})：")name = StdIn.readLine()if (name.length == 0) name = customer.namevar gender = customer.genderprint(s"性别(${gender})：")gender = StdIn.readChar()var age = customer.ageprint(s"年龄(${age})：")age = StdIn.readShort()var tel = customer.telprint(s"电话(${tel})：")tel = StdIn.readLine()if (tel.length == 0) tel = customer.telvar email = customer.emailprint(s"邮箱(${email})：")email = StdIn.readLine()if (email.length == 0) email = customer.email// 封装对象val newCustomer = new Customer(id, name, gender, age, tel, email)// 调用 CustomerService 的方法 updatecustomerService.update(id, newCustomer)println("---------------------修改完成---------------------")}

第十六章并发编程模型 Akka

16.1 Akka 的介绍

16.2 Actor 模型用于解决什么问题

16.3 Akka 中 Actor 模型详解

Actor 模型及其说明

对上图的详解如下：

16.4 Actor 模型工作机制说明

Actor模型工作机制说明(对照工作机制示意图理解)：

Actor 间传递消息机制(对照工作机制示意图理解)

16.5 Actor 模型应用实例

16.5.1 Actor 自我通讯

应用实例需求

代码实现
SayHelloActor 项目步骤：

创建项目 Mew -> New Project -> 选择 Maven
给项目命名
下一步 -> Finish
会生成 pom.xml 文件(maven 文件，项目包的依赖)
将下面的 maven 配置模板拷贝到 pom.xml 文件中，新的 pom.xml 文件文件内容如下：

<?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"><modelVersion>4.0.0</modelVersion><groupId>com.atguigu.akka</groupId><artifactId>SayHelloActor</artifactId><version>1.0-SNAPSHOT</version><!-- 定义一下常量 --><properties><encoding>UTF-8</encoding><scala.version>2.11.8</scala.version><scala.compat.version>2.11</scala.compat.version><akka.version>2.4.17</akka.version></properties><dependencies><!-- 添加scala的依赖 --><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version></dependency><!-- 添加akka的actor依赖 --><dependency><groupId>com.typesafe.akka</groupId><artifactId>akka-actor_${scala.compat.version}</artifactId><version>${akka.version}</version></dependency><!-- 多进程之间的Actor通信 --><dependency><groupId>com.typesafe.akka</groupId><artifactId>akka-remote_${scala.compat.version}</artifactId><version>${akka.version}</version></dependency></dependencies><!-- 指定插件--><build><!-- 指定源码包和测试包的位置 --><sourceDirectory>src/main/scala</sourceDirectory><testSourceDirectory>src/test/scala</testSourceDirectory><plugins><!-- 指定编译scala的插件 --><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.2.2</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals><configuration><args><arg>-dependencyfile</arg><arg>${project.build.directory}/.scala_dependencies</arg></args></configuration></execution></executions></plugin><!-- maven打包的插件 --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>2.4.3</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><filters><filter><artifact>*:*</artifact><excludes><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters><transformers><transformerimplementation="org.apache.maven.plugins.shade.resource.AppendingTransformer"><resource>reference.conf</resource></transformer><!-- 指定main方法 --><transformerimplementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass>xxx</mainClass></transformer></transformers></configuration></execution></executions></plugin></plugins></build>
</project>

因为按照配置模板的内容 “指定源码包和测试包的位置” 的部分：

    <sourceDirectory>src/main/scala</sourceDirectory><testSourceDirectory>src/test/scala</testSourceDirectory>

我们需要创建对应的 scala 目录，并 mark 为 Sources Root
7) 当修改后，第一次速度比较慢，因为 maven 需要 resolve 包的依赖，要下载相关的包。
注意：需要如图勾选，update snapshots，而且不需要联网，如果使用 maven 解决依赖后，仍然 pom.xml 有误，则只需要重启下 idea, 或者动一下 pom.xml 文件(不用改)，重新保存即可。

8) 代码实现：

package com.atguigu.akka.actorimport akka.actor.{Actor, ActorRef, ActorSystem, Props}// 1. 当我们继承 Actor 后，就是一个 Actor，需要重写该 Actor 的核心方法 receive
class SayHelloActor extends Actor {// 循环的接收消息// 1. receive方法，会被该 Actor 的 MailBox(实现了 Runnable 接口)调用// 2. 当该 Actor 的 MailBox 接收到消息，就会调用 receive 方法// 3. Receive 的底层：type Receive = PartialFunction[Any, Unit]override def receive: Receive = {// 接受消息并处理，如果接收到 exit，就退出case "hello" => println("发送：hello\t\t回应：hello too:)")case "ok" => println("发送：ok\t\t\t回应：ok too:)")case "exit" => {println("接收到exit~指令，退出系统...")context.stop(self) // 停止自己的 ActorRefcontext.system.terminate() // 关闭 ActorSystem}}
}object SayHelloActor {// 1. 先创建一个 ActorSystem，专门用于创建 Actorprivate val actoryFactory = ActorSystem("actoryFactory")// 2. 创建一个 Actor 的同时，返回 Actor 的 ActorRefprivate val sayHelloActorRef: ActorRef = actoryFactory.actorOf(Props[SayHelloActor], "sayHelloActor")// (1) Props[SayHelloActor] 创建了一个 SayHelloActor 实例，这里使用到了反射// (2) "sayHelloActor" 是 Actor 的名字// (3) sayHelloActorRef: ActorRef  =>是 Props[SayHelloActor] 的引用// (4) 创建的 SayHelloActor 实例被 ActorSystme 接管def main(args: Array[String]): Unit = {// 给 SayHelloActor 发消息(邮箱)sayHelloActorRef ! "hello"sayHelloActorRef ! "ok"sayHelloActorRef ! "ok~"// 研究异步如何退出 ActorSystemsayHelloActorRef ! "exit"}}

输出结果如下：

发送：hello      回应：hello too:)
发送：ok            回应：ok too:)
接收到exit~指令，退出系统...

运行的效果

代码的示意图和小结

小结：
当程序执行 private val sayHelloActorRef: ActorRef = actoryFactory.actorOf(Props[SayHelloActor], “sayHelloActor”) 会完成如下任务： [这是非常重要的方法]

1、actorFactory 是 ActorSystem(“actorFactory”) 创建的。
2、这里的 Props[SayHelloActor] 会使用反射机制，创建一个 SayHelloActor 对象，如果是 actorFactory.actorOf(Props(new SayHelloActor(其他代理对象的引用)), “sayHelloActor”) 形式，就是使用 new 的方式创建一个 SayHelloActor 对象。注意：Props() 是小括号。
3、会创建一个 SayHelloActor 对象的代理对象 sayHelloActorRef，使用 sayHelloActorRef 才能发送消息。
4、会在底层创建 Dispather Message，是一个线程池，用于分发消息，消息是发送到对应的 Actor 的 MailBox。
5、会在底层创建 SayHelloActor 的 MailBox 对象，该对象是一个队列，可接收 Dispatcher Message 发送的消息。
6、MailBox 实现了 Runnable 接口，是一个线程，一直运行并调用 Actor 的 receive 方法，因此当Dispather 发送消息到 MailBox 时，Actor 在r eceive 方法就可以得到信息。
7、SayHelloActorRef ! “hello” ，表示把 hello 消息发送到 SayHello Actor 的 Mailbox （通过Dispatcher Message 转发）。

16.5.2 Actor 之间通讯

应用实例需求

两个 Actor 的通讯机制原理图

代码实现
AActor.scala

package com.atguigu.akka.actorsimport akka.actor.{Actor, ActorRef}class AActor(bActorRef: ActorRef) extends Actor {var count = 0override def receive: Receive = {case "start" => {println("AActor 出招了，start ok")bActorRef ! "我打"}case "我打" => {count += 1// 给 BActor 发出消息// 这里需要持有 BActor 的引用(BActorRef)才可以println(s"AActor(黄飞鸿) 厉害！看我佛山无影脚 第${count}脚")Thread.sleep(1000)bActorRef ! "我打" // 给 BActor 发出消息}}
}

BActor.scala

package com.atguigu.akka.actorsimport akka.actor.Actorclass BActor extends Actor {var count = 0override def receive: Receive = {case "我打" => {count += 1println(s"BActor(乔峰) 挺猛 看我降龙十八掌 第${count}掌")Thread.sleep(1000)// 通过 sender() 方法，可以获取到发送消息的 Actor 的 ActorRefsender() ! "我打"}}
}

ActorApp.scala

package com.atguigu.akka.actorsimport akka.actor.{ActorRef, ActorSystem, Props}// 100招后，就退出
object ActorApp extends App {// 创建 ActorSystemval actorfactory = ActorSystem("actorfactory")// 先创建 BActor 的引用/代理val bActorRef: ActorRef = actorfactory.actorOf(Props[BActor], "bActor")// 创建 AActor 的引用时需要持有 BActor 的引用val aActorRef: ActorRef = actorfactory.actorOf(Props(new AActor(bActorRef)), "aActor")// aActor 先出招aActorRef ! "start"
}

输出结果如下：

AActor 出招了，start ok
BActor(乔峰) 挺猛 看我降龙十八掌 第1掌
AActor(黄飞鸿) 厉害！看我佛山无影脚 第1脚
BActor(乔峰) 挺猛 看我降龙十八掌 第2掌
AActor(黄飞鸿) 厉害！看我佛山无影脚 第2脚
BActor(乔峰) 挺猛 看我降龙十八掌 第3掌
AActor(黄飞鸿) 厉害！看我佛山无影脚 第3脚
BActor(乔峰) 挺猛 看我降龙十八掌 第4掌
AActor(黄飞鸿) 厉害！看我佛山无影脚 第4脚
BActor(乔峰) 挺猛 看我降龙十八掌 第5掌
AActor(黄飞鸿) 厉害！看我佛山无影脚 第5脚
BActor(乔峰) 挺猛 看我降龙十八掌 第6掌
AActor(黄飞鸿) 厉害！看我佛山无影脚 第6脚
......

代码的小结

1、两个 Actor 通讯机制和 Actor 自身发消息机制基本一样，只是要注意如下：
2、如果 A Actor 在需要给 B Actor 发消息，则需要持有 B Actor 的 ActorRef，可以通过创建 A Actor 时，传入 B Actor 的代理对象(ActorRef)。
3、当 B Actor 在 receive 方法中接收到消息，需要回复时，可以通过 sender() 获取到发送 Actor 的代理对象。

如何理解 Actor 的 receive 方法被调用？

1、每个 Actor 对应 MailBox。
2、MailBox 实现了 Runnable 接口，处于运行的状态。
3、当有消息到达 MailBox，就会去调用 Actor 的 receive 方法，即将消息推送给 receive 方法。

16.7 Akka 网络编程

看两个实际应用(socket/tcp/ip)
QQ、迅雷、百度网盘客户端、新浪网站、京东商城、淘宝

16.7.1 Akka 网络编程基本介绍

16.7.2 协议(tcp/ip)

TCP/IP(Transmission Control Protocol/Internet Protocol)的简写，中文译名为传输控制协议/因特网互联协议，又叫网络通讯协议，这个协议是Internet 最基本的协议、是 Internet 国际互联网络的基础，简单地说，就是由网络层的IP协议和传输层的TCP协议组成的。
TCP/IP 3本圣经级别书籍：xxx

16.7.3 OSI 与 Tcp/ip 参考模型

16.7.4 ip 地址

概述：每个 internet 上的主机和路由器都有一个 ip 地址，它包括网络号和主机号，ip 地址有 ipv4(32位) 或者 ipv6(128位)，可以通过 ipconfig（ifconfig）来查看。
一个小技巧：网络不通时，如何确定是哪一个路由(ip地址)出现问题？答：使用 tracert 指令。演示如下：

16.7.5 端口(port)

我们这里所指的端口不是指物理意义上的端口，而是特指TCP/IP协议中的端口，是逻辑意义上的端口。如果把 IP 地址比作一间房子，端口就是出入这间房子的门。真正的房子只有几个门，但是一个 IP 地址的端口可以有65535（即：256×256-1）个之多！端口是通过端口号来标记的。
端口(port)-分类

端口(port)-使用注意

socket 编程中客户端和服务器的网络分布

16.8 Akka 网络编程-小黄鸡客服案例

16.8.1 需求分析 + 界面设计

需求分析
1、服务端进行监听(9999)
2、客户端可以通过键盘输入，发送咨询问题给小黄鸡客服(服务端)
3、小黄鸡(服务端)回答客户的问题

界面设计
服务端：

客户端：

16.8.2 程序框架图

16.8.3 功能实现

代码结构：

示例代码如下：
YellowChickenServer.scala

package com.atguigu.akka.yellowchicken.serverimport akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.atguigu.akka.yellowchicken.common.{ClientMessage, ServerMessage}
import com.typesafe.config.ConfigFactoryclass YellowChickenServer extends Actor {override def receive: Receive = {case "start" => println("start 小黄鸡客服开始工作了...")// 如果接收到了服务端的发来的消息，即 ClientMessagecase ClientMessage(mes) => {println("客户咨询的问题是：" + mes)mes match {// 使用 match case 匹配（模糊匹配）case "大数据学费" => sender() ! ServerMessage("20000 RMB")case "学校地址" => sender() ! ServerMessage("北京市朝阳区青年路大悦城")case "学习什么技术" => sender() ! ServerMessage("大数据 前端 Python")case _ => sender() ! ServerMessage("你说的啥子:)")}}}
}// 主程序入口
object YellowChickenServerApp extends App {val host = "127.0.0.1" // 服务端ip地址val port = 9999 // 端口// 创建 config 对象，指定协议类型、监听的ip和端口val config = ConfigFactory.parseString(s"""|akka.actor.provider="akka.remote.RemoteActorRefProvider"|akka.remote.netty.tcp.hostname=$host|akka.remote.netty.tcp.port=$port""".stripMargin)// 创建 ActorSystemval serverActorSystem = ActorSystem("Server", config)// 创建 YellowChickenServer 的 Actor 和 ActorRefval yellowChickenServerActorRef: ActorRef = serverActorSystem.actorOf(Props[YellowChickenServer], "YellowChickenServer-01")// 启动服务端yellowChickenServerActorRef ! "start"
}

CustomerActor.scala

package com.atguigu.akka.yellowchicken.clientimport akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.atguigu.akka.yellowchicken.common.{ClientMessage, ServerMessage}
import com.typesafe.config.ConfigFactoryimport scala.io.StdInclass CustomerActor(serverHost: String, serverPort: Int) extends Actor {// 定义一个 YellowChickenServerRefvar serverActorRef: ActorSelection = _// 在 Actor 中有一个方法 preStart 方法，它会在 Actor 运行前执行// 在 Akka 开发中，通常将初始化的工作，放在 preStart 方法中override def preStart(): Unit = {this.serverActorRef = context.actorSelection(s"akka.tcp://Server@${serverHost}:${serverPort}/user/YellowChickenServer-01")println("this.serverActorRefer=" + this.serverActorRef)}override def receive: Receive = {case "start" => println("start 客户端运行，可以咨询问题")case mes: String => {// 发给服务端// serverActorRef ! mes // 不应该发送字符串，应该包装一把，应该发送一个(样例)对象（即协议）serverActorRef ! ClientMessage(mes) // 此时发送的是一个对象，该样例类默认实现了序列化 和 apply 方法}// 如果接受到了服务器端的消息case ServerMessage(mes) => {println(s"收到小黄鸡客服(Server)消息：$mes")}}
}// 主程序入口
object CustomerActorApp extends App {val (host, port, serverHost, serverPort) = ("127.0.0.1", 9990, "127.0.0.1", 9999)val config = ConfigFactory.parseString(s"""|akka.actor.provider="akka.remote.RemoteActorRefProvider"|akka.remote.netty.tcp.hostname=$host|akka.remote.netty.tcp.port=$port""".stripMargin)// 创建 ActorSystemval clientActorSystem = ActorSystem("Client", config)// 创建 CustomerActor 的 Actor 和 ActorRefval clientActorRef: ActorRef = clientActorSystem.actorOf(Props(new CustomerActor(serverHost, serverPort)), "CustomerActor-01")// 启动客户端clientActorRef ! "start"// 客户端发送消息while (true) {val mes = StdIn.readLine()clientActorRef ! mes}
}

MessageProtocol.scala

package com.atguigu.akka.yellowchicken.common// 使用样例类来构建协议// 1、客户端发送服务端的协议（序列化对象）
case class ClientMessage(mes: String)  // 回顾：样例类的构造器中的每一个参数都默认为 val ，即只可读。// 2、服务器端发送给客户端的协议
case class ServerMessage(mes: String)

16.9 Akka 网络编程-Spark Master Worker 进程通讯项目

16.9.1 项目意义

1、深入理解 Spark 的 Master 和 Worker 的通讯机制。
2、为了方便同学们看 Spark 的底层源码，命名的方式和源码保持一致(如：通讯消息类命名就是一样的)。
3、加深对主从服务心跳检测机制(HeartBeat)的理解，方便以后 spark 源码二次开发。

16.9.2 项目需求分析

16.9.3 项目界面设计

我们主要是通过应用实例，来剖析 Spark 的 Master 和 Worker 的通讯机制，因此功能比较简洁，设计的界面如下。看后面演示即可。

16.9.4 实现功能 1-Worker 完成注册

功能要求: Worker 注册到 Master，Master 完成注册，并回复 Worker 注册成功。

代码结构：

示例代码如下：
MasterActor.scala

package com.atguigu.akka.sparkmasterworker.masterimport akka.actor.{Actor, ActorSystem, Props}
import com.atguigu.akka.sparkmasterworker.common.{RegisterWorkerInfo, RegisteredWorkerInfo, WorkerInfo}
import com.typesafe.config.ConfigFactoryimport scala.collection.mutableclass MasterActor extends Actor {// 定义一个 mutable.HashMap 属性，用于管理 Workerval workers = mutable.HashMap[String, WorkerInfo]()override def receive: Receive = {case "start" => println("Master服务器启动了...")// 接收到 Worker 客户端注册的信息，保存进 HashMapcase RegisterWorkerInfo(id, cpu, ram) => {if (!workers.contains(id)) {// 创建 WorkerInfoval workerInfo = new WorkerInfo(id, cpu, ram)// 加入到 HashMapworkers += (id -> workerInfo)println("服务器的Workers= " + workers)// 回复客户端注册成功sender() ! RegisteredWorkerInfo}}}
}object MasterActorApp {def main(args: Array[String]): Unit = {val host = "127.0.0.1" // 服务端ip地址val port = 10005 // 端口// 创建 config 对象，指定协议类型、监听的ip和端口val config = ConfigFactory.parseString(s"""|akka.actor.provider="akka.remote.RemoteActorRefProvider"|akka.remote.netty.tcp.hostname=$host|akka.remote.netty.tcp.port=$port""".stripMargin)// 先创建 ActorSystemval masterActorSystem = ActorSystem("Master", config)// 再创建 Master 的 Actor 和 ActorRefval masterActorRef = masterActorSystem.actorOf(Props[MasterActor], "MasterActor-01")// 启动 MastermasterActorRef ! "start"}
}

WorkerActor.scala

package com.atguigu.akka.sparkmasterworker.workerimport akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.atguigu.akka.sparkmasterworker.common.{RegisterWorkerInfo, RegisteredWorkerInfo}
import com.typesafe.config.ConfigFactoryclass WorkerActor(serverHost: String, serverPort: Int) extends Actor {// 定义一个 MasterActorRefvar masterActorProxy: ActorSelection = _// 定义 Worker 的编号var id = java.util.UUID.randomUUID().toString// 在 Actor 中有一个方法 preStart 方法，它会在 Actor 运行前执行// 在 Akka 开发中，通常将初始化的工作，放在 preStart 方法中override def preStart(): Unit = {this.masterActorProxy = context.actorSelection(s"akka.tcp://Master@${serverHost}:${serverPort}/user/MasterActor-01")println("this.masterActorProxy=" + this.masterActorProxy)}override def receive = {case "start" => {println("Worker客户端启动运行")// 给服务器发送一个注册信息masterActorProxy ! RegisterWorkerInfo(id, 16, 16 * 1024)}case RegisteredWorkerInfo => {println("WorkedId= " + id + " 注册成功！")}}
}object WorkerActorApp {def main(args: Array[String]): Unit = {val (host, port, serverHost, serverPort) = ("127.0.0.1", 10001, "127.0.0.1", 10005)val config = ConfigFactory.parseString(s"""|akka.actor.provider="akka.remote.RemoteActorRefProvider"|akka.remote.netty.tcp.hostname=$host|akka.remote.netty.tcp.port=$port""".stripMargin)// 创建 ActorSystemval workerActorSystem = ActorSystem("Worker", config)// 创建 WorkerActor 的 Actor 和 ActorRefval workerActorRef: ActorRef = workerActorSystem.actorOf(Props(new WorkerActor(serverHost, serverPort)), "WorkerActor-01")// 启动客户端workerActorRef ! "start"}
}

MessageProtocol.scala

package com.atguigu.akka.sparkmasterworker.common// 使用样例类来构建协议// Worker 注册信息
case class RegisterWorkerInfo(id: String, cpu: Int, ram: Int)// 这个是 WorkerInfo，是保存在 Master 的 HashMap 中的，该 HashMap 用于管理 Worker
// 将来这个 WorkerInfo 会扩展，比如 增加 Worker 上一次的心跳时间
class WorkerInfo(val id: String, val cpu: Int, val ram: Int)// 当 Worker 注册成功，服务器返回一个 RegisteredWorkerInfo 对象
case object RegisteredWorkerInfo

16.9.5 实现功能 2-Worker 定时发送心跳

功能要求：Worker 定时发送心跳给 Master，Master 能够接收到，并更新 Worker 上一次心跳时间。

示例代码如下：
MessageProtocol.scala 中增加代码

package com.atguigu.akka.sparkmasterworker.common// 使用样例类来构建协议// Worker 注册信息
case class RegisterWorkerInfo(id: String, cpu: Int, ram: Int)// 这个是 WorkerInfo，是保存在 Master 的 HashMap 中的，该 HashMap 用于管理 Worker
// 将来这个 WorkerInfo 会扩展，比如 增加 Worker 上一次的心跳时间
class WorkerInfo(val id: String, val cpu: Int, val ram: Int) {// 新增属性：心跳时间var lastHeartBeatTime: Long = _
}// 当 Worker 注册成功，服务器返回一个 RegisteredWorkerInfo 对象
case object RegisteredWorkerInfo// 每隔一定时间定时器发送给 Master 一个心跳
case class HeartBeat(id: String)// Worker 每隔一定时间定时器发送给 自己 一个消息
case object SendHeartBeat

MasterActor.scala 中增加代码

    case HeartBeat(id) => {// 更新对应的 Worker 的心跳时间// 1、先从 Worker 中取出 WorkerInfoval workerInfo = workers(id)workerInfo.lastHeartBeatTime = System.currentTimeMillis()println("Master更新了 " + id + " 的心跳时间 ")}

WorkerActor.scala 中增加代码

      // 当客户端注册成功后，就定义一个定时器，每隔一定时间，发送 SendHeartBeat 给自己import context.dispatchercontext.system.scheduler.schedule(0 millis, 3000 millis, self, SendHeartBeat)case SendHeartBeat => {println("WorkedId= " + id + " 给Master发送心跳")masterActorProxy ! HeartBeat(id)}

16.9.6 实现功能 3-Master 启动定时任务，定时检测注册的 Worker

功能要求：Master 启动定时任务，定时检测注册的 Worker 有哪些没有更新心跳，已经超时的 Worker，将其从 HashMap 中删除掉。

示例代码如下：
MessageProtocol.scala 中增加代码

// Master 给自己发送一个触发检查超时 Worker 的信息
case object StartTimeOutWorker// Master 给自己发消息，检测 Worker，对于心跳超时的
case object RemoveTimeOutWorker

MasterActor.scala 中增加代码

    case "start" => {println("Master服务器启动了...")// Master 启动定时任务，定时检测注册的 Worker 有哪些没有更新心跳，已经超时的 Worker，将其从 HashMap 中删除掉。self ! StartTimeOutWorker}// 开启定时器，每隔一定时间检测是否有 Worker 的心跳超时case StartTimeOutWorker => {println("开启了定时检测Worker心跳的任务")import context.dispatcher // 使用调度器时候必须导入dispatchercontext.system.scheduler.schedule(0 millis, 9000 millis, self, RemoveTimeOutWorker)}// 判断哪些 Worker 心跳超时（nowTime - lastHeartBeatTime），对已经超时的 Worker，将其从 HashMap 中删除掉。case RemoveTimeOutWorker => {// 首先获取所有 Workers 的所有 WorkerInfoval workerInfos = workers.valuesval nowTime = System.currentTimeMillis()// 过滤出所有超时的 workerInfo 并删除即可workerInfos.filter(workerInfo => (nowTime - workerInfo.lastHeartBeatTime) > 6000).foreach(workerInfo => workers.remove(workerInfo.id))println("当前有 " + workers.size + " 个Worker存活")}

16.9.7 实现功能 4-Master,Worker 的启动参数运行时指定

功能要求：Master，Worker 的启动参数运行时指定，而不是固定写在程序中的。

MasterActor.scala 中修改代码

    if (args.length != 3) {println("请输入参数 host port MasterActor的名字")sys.exit()}val host = args(0)  // 服务端ip地址val port = args(1)  // 端口val masterName = args(2)  // MasterActor的名字......// 再创建 Master 的 Actor 和 ActorRefval masterActorRef = masterActorSystem.actorOf(Props[MasterActor], s"${masterName}")

WorkerActor.scala 中增修改代码

    if (args != 6) {println("请输入参数 host port WorkerActor的名字 serverHost serverPort MasterActor的名字")}val host = args(0)val port = args(1)val workerName = args(2)val serverHost = args(3)val serverPort = args(4)val masterName = args(5)......// 创建 WorkerActor 的 Actor 和 ActorRefval workerActorRef: ActorRef = workerActorSystem.actorOf(Props(new WorkerActor(serverHost, serverPort.toInt, masterName)), s"${workerName}")

Master 配置参数截图：

Worker 配置参数截图：

16.9.8 Master Worker 进行分布式部署

Master Worker 进行分布式部署：Linux 系统 -> 如何给 maven 项目打包 -> 上传Linux
步骤如下：
步骤一：先给 MasterActor 打包，修改 pom.xml 文件的 <mainClass>xxx</mainClass> 节点，指定我们程序 MasterActor 的主类
即修改成如下：<mainClass>com.atguigu.akka.sparkmasterworker.master.MasterActorApp</mainClass>

步骤二：给 MasterActor 打包

步骤三：将打好的 jar 包拷贝至某个目录里，并修改 jar 包名称为 MasterActor.jar，等待上传至 Linux 系统

步骤四：再给 WorkerActor 打包，修改 pom.xml 文件的 <mainClass>xxx</mainClass>节点，指定我们程序 WorkerActor 的主类
即修改成如下：<mainClass>com.atguigu.akka.sparkmasterworker.worker.WorkerActorApp</mainClass>

步骤五：给 WorkerActor 打包，操作同步骤二，注意：打包前我们需要先 clean 下

步骤六：将打好的 jar 包拷贝至某个目录里，并修改 jar 包名称为 WorkerActor.jar，等待上传至 Linux 系统

步骤七：将两个 jar 包上传至 Linux

步骤八：测试运行，，命令如下：

java -jar MasterActor.jar 127.0.0.1 10005 MasterActor-01
java -jar WorkerActor.jar 127.0.0.1 10001 WorkerActor-01 127.0.0.1 10005 MasterActor-01
java -jar WorkerActor.jar 127.0.0.1 10002 WorkerActor-02 127.0.0.1 10005 MasterActor-01