第十五章 客户信息管理系统
15.1 项目的开发流程
15.2 项目的需求分析
模拟实现基于文本界面的《客户信息管理软件》。 该软件 scala 能够实现对客户对象的插入、修改、删除、显示、查询（用 ArrayBuffer 或者 ListBuffer 实现），并能够打印客户明细表。
15.3 项目的界面
主界面
添加客户
修改客户
删除客户
客户列表
15.4 项目的设计-程序框架图
程序框架图：设计系统有多少个文件，以及文件之间的调用关系，可以帮助程序员实现模块的设计（清晰），便于程序员之间对项目交流分析。【业务优化，设计方案】
15.5 项目的功能实现
15.5.1 完成 Customer 类
根据需求文档或者页面，写出 Customer 类 Customer.scala
代码语言：javascript复制package com.atguigu.chapter15.customercrm.bean
class 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) {
this
this.id = id
this.name = name
this.gender = gender
this.age = age
this.tel = tel
this.email = email
}
}15.5.2 完成显示主菜单和退出软件功能
CustomerView.scala 功能分析： 1. 将主菜单的显示放入到 while 2. 用户可以根据输入，选择自己的操作 3. 如果输入5退出 CustomerView.scala
代码语言：javascript复制package com.atguigu.chapter15.customercrm.view
import scala.io.StdIn
class 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("你退出了系统...")
}
}示例代码如下：
代码语言：javascript复制package com.atguigu.chapter15.customercrm.app
import com.atguigu.chapter15.customercrm.view.CustomerView
object 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 方法
代码语言：javascript复制 override def toString: String = {
this.id "tt" this.name "tt" this.gender "tt" this.age "tt" this.tel "tt" this.email
}2、在 CustomerService.scala 中编写一个方法 list，返回当前系统有哪些客户
代码语言：javascript复制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
代码语言：javascript复制 val customerService = new CustomerService()
/*
---------------------------客户列表---------------------------
编号 姓名 性别 年龄 电话 邮箱
1 张三 男 30 010-56253825 abc@email.com
2 李四 女 23 010-56253825 lisi@ibm.com
3 王芳 女 26 010-56253825 wang@163.com
-------------------------客户列表完成-------------------------
*/
def list(): Unit = {
println()
println("---------------------------客户列表---------------------------")
println("编号tt姓名tt性别tt年龄tt电话tt邮箱")
// 遍历
// 调用 CustomerService 的方法 list
val customers = customerService.list()
for (customer <- customers) {
// 方式一：输出
// println(customer.id "tt" ...)
// 方式二：重写 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属性)
代码语言：javascript复制 // 辅助构造器(没有id属性)
def this(name: String, gender: Char, age: Short, tel: String, email: String) {
this
this.name = name
this.gender = gender
this.age = age
this.tel = tel
this.email = email
}2、在 CustomerService.scala 中编写一个方法 add，接收一个 Customer 对象，并设置 id 后再加入到 ArrayBuffer 中
代码语言：javascript复制 // 用于设置用户 id
var customerNum = 1
// 添加客户的方法
def add(customer: Customer): Boolean = {
// 设置 id
customerNum = 1
customer.id = customerNum
// 加入到 ArrayBuffer 中
customers.append(customer)
true
}3、在 CustomerView.scala 中 调用 CustomerService 的方法 add
代码语言：javascript复制 /*
---------------------添加客户---------------------
姓名：张三
性别：男
年龄：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 的方法 add
customerService.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，判断
代码语言：javascript复制 // 先根据 id 查找 用户的 index
def findIndexById(id: Int): Int = {
// 先假定一个索引，默认 -1，如果找到就改成对应的，如果没有找到就返回 -1
var index = -1
// 遍历 ArrayBuffer
breakable {
for (i <- 0 until customers.length) {
if (customers(i).id == id) {
index = i
break()
}
}
}
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)
代码语言：javascript复制 /*
---------------------删除客户---------------------
请选择待删除客户编号(-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().toLower
if (choice == 'y') {
if (customerService.del(id)) {
println("---------------------删除完成---------------------")
return
}
}
println("---------------------删除没有完成---------------------")
}15.5.6 完善退出确认功能
功能说明： 要求用户在退出时提示 "确认是否退出(Y/N)："，用户必须输入y/n，否则循环提示。且输入为y时，退出系统；输入为n时，不退出系统。 1、在 CustomerView.scala 中定义一个方法 isOut，并修改 key 所对应的函数。
代码语言：javascript复制 // 要求用户在退出时提示"确认是否退出(Y/N)："，用户必须输入y/n，否则循环提示。且输入为y时，退出系统；输入为n时，不退出系统。
def isOut(): Unit = {
println()
println("确认是否退出(Y/N)：")
key = StdIn.readChar().toLower
key match {
case 'y' => this.loop = false
case 'n' => this.loop = true
case _ => isOut()
}
}15.5.7 完善删除确认功能
功能说明： 要求用户在删除确认时提示 "确认是否删除(Y/N)："，用户必须输入y/n，否则循环提示。 1、在 CustomerView.scala 中，修改 del() 方法即可
代码语言：javascript复制 /*
---------------------删除客户---------------------
请选择待删除客户编号(-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().toLower
if (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 查找用户信息 的方法
代码语言：javascript复制 // 根据 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
代码语言：javascript复制 /*
---------------------修改客户---------------------
请选择待修改客户编号(-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.name
print(s"姓名(${name})：")
name = StdIn.readLine()
if (name.length == 0) name = customer.name
var gender = customer.gender
print(s"性别(${gender})：")
gender = StdIn.readChar()
var age = customer.age
print(s"年龄(${age})：")
age = StdIn.readShort()
var tel = customer.tel
print(s"电话(${tel})：")
tel = StdIn.readLine()
if (tel.length == 0) tel = customer.tel
var email = customer.email
print(s"邮箱(${email})：")
email = StdIn.readLine()
if (email.length == 0) email = customer.email
// 封装对象
val newCustomer = new Customer(id, name, gender, age, tel, email)
// 调用 CustomerService 的方法 update
customerService.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 项目步骤： 1) 创建项目 Mew -> New Project -> 选择 Maven 2) 给项目命名
3) 下一步 -> Finish 4) 会生成 pom.xml 文件(maven 文件， 项目包的依赖)
5) 将下面的 maven 配置模板拷贝到 pom.xml 文件中，新的 pom.xml 文件文件内容如下：
代码语言：javascript复制<?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>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>reference.conf</resource>
</transformer>
<!-- 指定main方法 -->
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>xxx</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>6) 因为按照配置模板的内容 "指定源码包和测试包的位置" 的部分：
代码语言：javascript复制 <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) 代码实现：
代码语言：javascript复制package com.atguigu.akka.actor
import 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("发送：hellott回应：hello too:)")
case "ok" => println("发送：okttt回应：ok too:)")
case "exit" => {
println("接收到exit~指令，退出系统...")
context.stop(self) // 停止自己的 ActorRef
context.system.terminate() // 关闭 ActorSystem
}
}
}
object SayHelloActor {
// 1. 先创建一个 ActorSystem，专门用于创建 Actor
private val actoryFactory = ActorSystem("actoryFactory")
// 2. 创建一个 Actor 的同时，返回 Actor 的 ActorRef
private 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~"
// 研究异步如何退出 ActorSystem
sayHelloActorRef ! "exit"
}
}输出结果如下：
代码语言：javascript复制发送：hello 回应：hello too:)
发送：ok 回应：ok too:)
接收到exit~指令，退出系统...9) 运行的效果
代码的示意图和小结
小结： 当程序执行 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
代码语言：javascript复制package com.atguigu.akka.actors
import akka.actor.{Actor, ActorRef}
class AActor(bActorRef: ActorRef) extends Actor {
var count = 0
override 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
代码语言：javascript复制package com.atguigu.akka.actors
import akka.actor.Actor
class BActor extends Actor {
var count = 0
override def receive: Receive = {
case "我打" => {
count = 1
println(s"BActor(乔峰) 挺猛 看我降龙十八掌 第${count}掌")
Thread.sleep(1000)
// 通过 sender() 方法，可以获取到发送消息的 Actor 的 ActorRef
sender() ! "我打"
}
}
}ActorApp.scala
代码语言：javascript复制package com.atguigu.akka.actors
import akka.actor.{ActorRef, ActorSystem, Props}
// 100招后，就退出
object ActorApp extends App {
// 创建 ActorSystem
val 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"
}输出结果如下：
代码语言：javascript复制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
代码语言：javascript复制package com.atguigu.akka.yellowchicken.server
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.atguigu.akka.yellowchicken.common.{ClientMessage, ServerMessage}
import com.typesafe.config.ConfigFactory
class YellowChickenServer extends Actor {
override def receive: Receive = {
case "start" => println("start 小黄鸡客服开始工作了...")
// 如果接收到了服务端的发来的消息，即 ClientMessage
case 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)
// 创建 ActorSystem
val serverActorSystem = ActorSystem("Server", config)
// 创建 YellowChickenServer 的 Actor 和 ActorRef
val yellowChickenServerActorRef: ActorRef = serverActorSystem.actorOf(Props[YellowChickenServer], "YellowChickenServer-01")
// 启动服务端
yellowChickenServerActorRef ! "start"
}CustomerActor.scala
代码语言：javascript复制package com.atguigu.akka.yellowchicken.client
import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.atguigu.akka.yellowchicken.common.{ClientMessage, ServerMessage}
import com.typesafe.config.ConfigFactory
import scala.io.StdIn
class CustomerActor(serverHost: String, serverPort: Int) extends Actor {
// 定义一个 YellowChickenServerRef
var 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)
// 创建 ActorSystem
val clientActorSystem = ActorSystem("Client", config)
// 创建 CustomerActor 的 Actor 和 ActorRef
val clientActorRef: ActorRef = clientActorSystem.actorOf(Props(new CustomerActor(serverHost, serverPort)), "CustomerActor-01")
// 启动客户端
clientActorRef ! "start"
// 客户端发送消息
while (true) {
val mes = StdIn.readLine()
clientActorRef ! mes
}
}MessageProtocol.scala
代码语言：javascript复制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
代码语言：javascript复制package com.atguigu.akka.sparkmasterworker.master
import akka.actor.{Actor, ActorSystem, Props}
import com.atguigu.akka.sparkmasterworker.common.{RegisterWorkerInfo, RegisteredWorkerInfo, WorkerInfo}
import com.typesafe.config.ConfigFactory
import scala.collection.mutable
class MasterActor extends Actor {
// 定义一个 mutable.HashMap 属性，用于管理 Worker
val workers = mutable.HashMap[String, WorkerInfo]()
override def receive: Receive = {
case "start" => println("Master服务器启动了...")
// 接收到 Worker 客户端注册的信息，保存进 HashMap
case RegisterWorkerInfo(id, cpu, ram) => {
if (!workers.contains(id)) {
// 创建 WorkerInfo
val workerInfo = new WorkerInfo(id, cpu, ram)
// 加入到 HashMap
workers = (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)
// 先创建 ActorSystem
val masterActorSystem = ActorSystem("Master", config)
// 再创建 Master 的 Actor 和 ActorRef
val masterActorRef = masterActorSystem.actorOf(Props[MasterActor], "MasterActor-01")
// 启动 Master
masterActorRef ! "start"
}
}WorkerActor.scala
代码语言：javascript复制package com.atguigu.akka.sparkmasterworker.worker
import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.atguigu.akka.sparkmasterworker.common.{RegisterWorkerInfo, RegisteredWorkerInfo}
import com.typesafe.config.ConfigFactory
class WorkerActor(serverHost: String, serverPort: Int) extends Actor {
// 定义一个 MasterActorRef
var 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)
// 创建 ActorSystem
val workerActorSystem = ActorSystem("Worker", config)
// 创建 WorkerActor 的 Actor 和 ActorRef
val workerActorRef: ActorRef = workerActorSystem.actorOf(Props(new WorkerActor(serverHost, serverPort)), "WorkerActor-01")
// 启动客户端
workerActorRef ! "start"
}
}MessageProtocol.scala
代码语言：javascript复制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 RegisteredWorkerInfo16.9.5 实现功能 2-Worker 定时发送心跳
功能要求：Worker 定时发送心跳给 Master，Master 能够接收到，并更新 Worker 上一次心跳时间。
示例代码如下： MessageProtocol.scala 中增加代码
代码语言：javascript复制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 SendHeartBeatMasterActor.scala 中增加代码
代码语言：javascript复制 case HeartBeat(id) => {
// 更新对应的 Worker 的心跳时间
// 1、先从 Worker 中取出 WorkerInfo
val workerInfo = workers(id)
workerInfo.lastHeartBeatTime = System.currentTimeMillis()
println("Master更新了 " id " 的心跳时间 ")
}WorkerActor.scala 中增加代码
代码语言：javascript复制 // 当客户端注册成功后，就定义一个定时器，每隔一定时间，发送 SendHeartBeat 给自己
import context.dispatcher
context.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 中增加代码
代码语言：javascript复制// Master 给自己发送一个触发检查超时 Worker 的信息
case object StartTimeOutWorker
// Master 给自己发消息，检测 Worker，对于心跳超时的
case object RemoveTimeOutWorkerMasterActor.scala 中增加代码
代码语言：javascript复制 case "start" => {
println("Master服务器启动了...")
// Master 启动定时任务，定时检测注册的 Worker 有哪些没有更新心跳，已经超时的 Worker，将其从 HashMap 中删除掉。
self ! StartTimeOutWorker
}
// 开启定时器，每隔一定时间检测是否有 Worker 的心跳超时
case StartTimeOutWorker => {
println("开启了定时检测Worker心跳的任务")
import context.dispatcher // 使用调度器时候必须导入dispatcher
context.system.scheduler.schedule(0 millis, 9000 millis, self, RemoveTimeOutWorker)
}
// 判断哪些 Worker 心跳超时（nowTime - lastHeartBeatTime），对已经超时的 Worker，将其从 HashMap 中删除掉。
case RemoveTimeOutWorker => {
// 首先获取所有 Workers 的所有 WorkerInfo
val workerInfos = workers.values
val 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 中修改代码
代码语言：javascript复制 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 和 ActorRef
val masterActorRef = masterActorSystem.actorOf(Props[MasterActor], s"${masterName}")WorkerActor.scala 中增修改代码
代码语言：javascript复制 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 和 ActorRef
val workerActorRef: ActorRef = workerActorSystem.actorOf(Props(new WorkerActor(serverHost, serverPort.toInt, masterName)), s"${workerName}")Master 配置参数截图：
Worker 配置参数截图：
