主要的参考资料来源 《深度学习入门：基于Python的理论与实现 (斋藤康毅) 》（鱼书） B站Up主爆肝杰哥的相关视频 主要考虑到是入门以及快速能上手工程，没有基础，不能在数学等其余问题上过多的纠结，本身掌握程度不够，所以不会以李沐老师的课成为第一课程，后续才会去看 此外，因为是为了深度学习才接触的Python，并非专攻Python方向，所以对于Python的很多其他的特性也不会涉及，一切以深度学习的需要进行学习

前言

在深度学习领域，我们主要是利用Python以及几个库进行，分别是

• NumPy-为Python加上了关键的数组变量类型

• Pandas-在NumPy的基础上添加了类似Excel的行列标签

• Matplotlib-借鉴Matlab，为Python提供了绘图能力

• Scikit-learn-机器学习库，包含了分类，回归，聚类，将为等多种算法

• Tensorflow-Goole研发的深度学习框架

• PyTorch-FaceBook研发的深度学习框架

深度学习的相关概念

大家可能听说过人工智能；机器学习；神经网络；深度学习等多个专有名词，实际上这几个都是有联系的

• 人工智能是最大的一个概念，最顶层的，其中一个重要的分支就是机器学习

• 机器学习的算法多种多样，其中核心就是神经网络

• 神经网络的隐藏层如果够深，就可以被称之为深层神经网络，这也属于我们所说的深度学习

• 深度学习包括了深度神经网络；卷积神经网络；循环神经网络

变量类型与输出语句

总览

学习Python首先需要注意如下几点

1. Python的注释是采用#而不是 //，多行注释要采用三个双引号"""..."""

2. Python是动态语言类型不是静态，在声明变量的时候不需要指定变量类型，如果需要指定数据类型，可以使用类型注解（但是即使使用，也只是作为提示作用，不会受到编译器强制校验）

3. Python没有末尾分号，但是有比较烦人的缩进，所以不要在Python中随便使用空格

首先我们来看看如何定义一个变量

# 直接声明和初始化即可
a = 2  
# 如果需要指定类型 可以使用类型注解，即name:type的形式
b:int = 2  
# 但是类型注解只是起到提示的作用，编译器并不会校验，所以如下代码也不会报错
c:int = "abc"

Python原生的数据类型分为两大类：基本变量类型和高级变量类型，一共七种数据类型 基本数据类型包括：字符串，数字，布尔类型 ； 高级变量类型包括：集合，元组，列表，字典 七种类型展示如下，需要注意一下几点

• Python不能够声明但是不赋值，这是因为在Python中变量只有被赋值的时候才是真正被创建，在其他语言中，例如Java中允许int a;这是因为在Java中声明即创建了对应的内存空间，a已经存在只是没有数值

• Python中的布尔类型严格遵循大小写，只能是True和False，而不能是true或false

• Python四大高级变量类型，set集合是使用大括号{} ； list列表使用中括号[] ； tuple元组使用的小括号() ； dict字典是大括号但为Key-Value的形式

• list集合：有序，支持下标访问，地址不可变但是内容可变，支持增删改，允许重复，空列表使用[]或者list()

• tuple元组：有序，同列表可下标访问；不可变，一旦确认不能修改，不能增删改；允许重复；空元组使用()或者tuple()；如果是只有一个元素的元组创建，必须加一个逗号(1,)

• set集合：无序,不能下标查找；可变，支持增删，但不能修改已有元素；不可重复，自动去重；空集合使用set()；支持集合运算符交集(&),并集( | ),差集(-)等

• dict字典：3.7版本及其以上默认按照插入顺序排序；3.7之前无序 ； 可变 Key唯一，重复会覆盖原本的 ； Value可重复 ；Key必须是不可变类型(例如int,str,tuple；列表/集合/字典就不能作为Key) ； 空字典使用dict()或者{}，只能通过键值对访问而不能通过索引访问

# 声明但不赋值 这个在python中是不允许的
s:int
​
# 基本变量类型  
s = "zeal" # 字符串  
# 数字类型  
age = 22  
height = 1.75  
weight = 120.0  
# 布尔类型  
bool_t = True  
bool_f = False  
  
# 高级变量类型  
#集合set 利用大括号包裹{}  
set_v = {1,2,3}  
print(set_v)  
  
  
# 元组tuple 利用 小括号包裹()  
tuple_v = (1,2,3)  
print(tuple_v)  
  
# 列表  
list_v = [1,2,3,4,5]  
print(list_v)  
  
# 字典dict  
dict_v = {"name":"zeal","age":22}  
print(dict_v)

基本变量类型

字符串类型

Python中字符串类型，也就是我们别的编程语言中说的String类型，可以使用双引号，也可以使用单引号,支持两种方式的原因是为了保证当字符串本身有单引号或者双引号的时候可以灵活处理，如果字符串本身同时具备单引号和双引号，那么和别的编程语言一样，就需要使用转义了。

# 当变量中本身具备单引号  
s1 = "I'm a student."  # 使用双引号包裹字符串 防止字符串本身的单引号截断  
# 当变量中本身具备双引号  
s2 = 'He said, "I am a student."'  # 使用单引号包裹字符串 防止字符串本身的双引号截断  
# 当变量中同时具备单引号和双引号  
s3 = "He said, \"I'm a student.\""  # 使用转义字符 \ 来表示字符串中的双引号,当然，也可以转义单引号

当我们要想组合字符串类型的变量的时候，可以利用 f字符串，将需要引入的字符串变量在f后的字符串中通过大括号{...}包裹起来从而实现引用，在输出中我们常这么使用

name = "zeal"  
love = "play game"  
# 组合字符串为一个新字符串
result = f"My name is {name} and I love {love}"  
print(result)
# 直接在打印语句中组合字符串  
print(f"My name is {name} and I love {love}")

当然，print还有另外一个用法，直接利用逗号,进行拼接，但是因为默认,会用一个空格隔开，我们可以在最后用sep=设置间隔符从而实现无间隔拼接

print("my name is",name,"and age is",age,sep=" ")

数字类型

数字类型分为整型和浮点型，具体的区别我们在NumPy中再去介绍。 Python中常见运算规则如下

# 1. 算术运算符
print("-" * 40)  
a = 10  
b = 3  
print(f"a = {a}, b = {b}")  
  
print(f"加法: a + b = {a} + {b} = {a + b}")  
print(f"减法: a - b = {a} - {b} = {a - b}")  
print(f"乘法: a * b = {a} * {b} = {a * b}")  
#需要注意python中的单个/是完整的除法 也就是会计算完
#别的语言中 / 是整除，python中 // 才是整除
print(f"除法: a / b = {a} / {b} = {a / b}")  
print(f"整除: a // b = {a} // {b} = {a // b}")  
print(f"取余: a % b = {a} % {b} = {a % b}")  
print(f"幂运算: a ** b = {a} ** {b} = {a ** b}")  # a的b次方,即a^b
  
# 2. 比较运算符
# 等于为== 不等于为!= 可以直接使用>= 和 <=
print("-" * 40)  
x = 15  
y = 20  
print(f"x = {x}, y = {y}")  
  
print(f"等于: x == y -> {x} == {y} = {x == y}")  
print(f"不等于: x != y -> {x} != {y} = {x != y}")  
print(f"大于: x > y -> {x} > {y} = {x > y}")  
print(f"小于: x < y -> {x} < {y} = {x < y}")  
print(f"大于等于: x >= y -> {x} >= {y} = {x >= y}")  
print(f"小于等于: x <= y -> {x} <= {y} = {x <= y}")  
  
# 3. 赋值运算符 (Assignment Operators)print("\n【3】赋值运算符 (Assignment Operators)")  
print("-" * 40)  
c = 5  
print(f"初始值: c = {c}")  
  
c += 3      # c = c + 3  
print(f"加法赋值: c += 3 -> c = {c}")  
  
c -= 2      # c = c - 2  
print(f"减法赋值: c -= 2 -> c = {c}")  
  
c *= 4      # c = c * 4  
print(f"乘法赋值: c *= 4 -> c = {c}")  
  
c /= 2      # c = c / 2  
print(f"除法赋值: c /= 2 -> c = {c}")  
  
c //= 3     # c = c // 3  
print(f"整除赋值: c //= 3 -> c = {c}")  
  
c %= 4      # c = c % 4  
print(f"取余赋值: c %= 4 -> c = {c}")  
  
c **= 3     # c = c ** 3  
print(f"幂运算赋值: c **= 3 -> c = {c}")  
  
# 4. 位运算符 
print("-" * 40)  
m = 12      # 二进制: 1100  
n = 10      # 二进制: 1010  
# 利用bin(number) 可以得到number的二进制
print(f"m = {m} (二进制: {bin(m)}), n = {n} (二进制: {bin(n)})")  
  
print(f"按位与: m & n = {m} & {n} = {m & n} (二进制: {bin(m & n)})")  
print(f"按位或: m | n = {m} | {n} = {m | n} (二进制: {bin(m | n)})")  
print(f"按位异或: m ^ n = {m} ^ {n} = {m ^ n} (二进制: {bin(m ^ n)})")  
print(f"按位取反: ~m = ~{m} = {~m} (二进制: {bin(~m & 0b1111)})")  
print(f"左移: m << 2 = {m} << 2 = {m << 2} (二进制: {bin(m << 2)})")  
print(f"右移: m >> 2 = {m} >> 2 = {m >> 2} (二进制: {bin(m >> 2)})")  
  
# 5. 逻辑运算符 这里需要注意和别的语言的区别,python的与或非使用的为 and or not 而不是& | !
print("-" * 40)  
p = True  
q = False  
print(f"p = {p}, q = {q}")  
  
print(f"逻辑与: p and q = {p} and {q} = {p and q}")  
print(f"逻辑或: p or q = {p} or {q} = {p or q}")  
print(f"逻辑非: not p = not {p} = {not p}")  
  
# 6. 成员运算符和身份运算符 
# 主要用于高级变量类型，检查是否在集合内，in 和 not in
# 元组，列表，集合，字典都可以使用，对于字典而言，检查的是key不是value
print("-" * 40)  
num_list = [1, 2, 3, 4, 5]  
num1 = 10  
num2 = 10  
print(f"列表: {num_list}, num1 = {num1}, num2 = {num2}")  
  
print(f"成员运算符 in: 3 in num_list = {3 in num_list}")  
print(f"成员运算符 not in: 6 not in num_list = {6 not in num_list}")  
print(f"身份运算符 is: num1 is num2 = {num1 is num2}")  
print(f"身份运算符 is not: num1 is not num_list = {num1 is not num_list}")  
  
# 7. 运算符优先级示例  
print("\n【7】运算符优先级示例")  
print("-" * 40)  
result1 = 10 + 5 * 3  
result2 = (10 + 5) * 3  
result3 = 2 ** 3 * 4  
result4 = 2 ** (3 * 4)  
  
print(f"乘法优先于加法: 10 + 5 * 3 = {result1}")  
print(f"括号改变优先级: (10 + 5) * 3 = {result2}")  
print(f"幂运算优先于乘法: 2 ** 3 * 4 = {result3}")  
print(f"括号改变优先级: 2 ** (3 * 4) = {result4}")  
  
# 8. 不同数字类型之间的运算  
print("\n【8】不同数字类型之间的运算")  
print("-" * 40)  
int_num = 10  
float_num = 3.5  
complex_num = 2 + 3j  
  
print(f"整数: {int_num}, 浮点数: {float_num}, 复数: {complex_num}")  
print(f"整数 + 浮点数: {int_num} + {float_num} = {int_num + float_num} (类型: {type(int_num + float_num)})")  
print(f"整数 + 复数: {int_num} + {complex_num} = {int_num + complex_num} (类型: {type(int_num + complex_num)})")  
print(f"浮点数 * 复数: {float_num} * {complex_num} = {float_num * complex_num} (类型: {type(float_num * complex_num)})")

类型转换

Python中，字符串，浮点数，整数，布尔类型之间都可以转换

• 需要转化为字符串，使用str()

• 需要转化为整数，使用int()

• 需要转化为浮点数，使用float()

• 需要转化为布尔类型，使用bool() 需要注意，上述的这些属于内置工具函数，而不是别的语言中的强转，但也要注意 浮点数字符串不能int(); 非0数字和非空字符串(包括空格组成的字符串) 使用bool()返回True; 0和串("")使用bool()返回False

str_v = "123.123"  
print(float(str_v))  
print(bool(str_v))  
  
number1 = 99  
number2 = 0  
print(str(number1))  
print(str(number2))  
print(bool(number2))  
  
bool_v1 = True  
bool_v2 = False  
print(str(bool_v1))  
print(str(bool_v2))  
print(int(bool_v1))  
print(int(bool_v2))
​

高级变量类型

Set集合

Set集合的特点：无序 ； 不可重复 ；可以使用-；|；&进行集合的运算；不能修改，只能删除和新增

set_1 = {"湖","南","科","专",1,2,3}  
print(set_1) # 输出 {1, 2, 3, '南', '专', '湖', '科'}  体现无序
set_2 = set(["湖","南","科","科",4,5,6])  # 类似上面的类型转换，可有元组得到
print(set_2) # 输出 {4, 5, 6, '南', '湖', '科'} 体现自动去重
​
print(set_1-set_2) # set1中有但是set2中没有的  
print(set_1|set_2) # set1或set2中的所有元素  
print(set_1&set_2) # set1和set2中的交集元素

我们可以使用 in 和 not in判断元素是否存在于集合中

可以通过list(set)或者tuple(set)的方式将集合转换为列表或者元组，从而实现遍历和查找元素 如果要修改元素，因为不支持修改，我们只能先删除再添加，删除我们常用的语法是remove()和discard()（前者删除不存在的元素的时候会报错，后者不会更加安全），新增则用add()

集合在深度学习的过程中遇到的概率比较低，会被Pands替代，所以不过多讲述

List列表

Python中的列表就可以比作Java中的ArrayList，边长的数组，可重复，有序，下标访问，只是基于Python动态语言的特性，一个List中可以包含任何数据类型，不存在类型限制，也可以List套List

list_v = [1,2,3,"ZhangSan",True,[4,5,6]]  
print(list_v)
print(list_v[0])

虽然List没有限制，但是实际上底层List还是会需要存储每个类型的数据类型，导致数组其实占用空间增大了。在NumPy中我们会用NumPy数组（仅仅接纳一种数据类型的列表）代替List，提升运行速度，降低内存压力

List支持负数下标索引，我们知道0索引位置对应的List的第一个元素，那么-1其实就是倒数第一个元素，以此类推，-2就是倒数第二个元素

list_v = [1,2,3,"ZhangSan",True,[4,5,6]]  
print(list_v[-1],list_v[-2],list_v[-3])  
list_v[-2]=100  
print(list_v[-1],list_v[-2],list_v[-3])

Python的List有着和Golang的数组一样的切片的特性，切片就是列表一部分，可以利用[start:end]的形式截取列表中的 start~end 之间的部分，注意左闭右开

list_v = [1,2,3,"ZhangSan",True,[4,5,6]]  
print(list_v[1:]) # 第二个元素到最后  
print(list_v[:]) # 全部元素  
print(list_v[1:2]) # 只有第二个元素  
print(list_v[:5]) # 前四个元素 等同于 print(list_v[:-1])

除此之外，Python的切片还能支持间隔切片。当我们明确隔n个元素采样一次的时候，可以使用[star:end:n]

list_v = [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15]  
print(list_v[1:-1:2]) # [2, 4, 6, 8, 10, 12, 14]

特别注意，Python的切片和原列表对象是完全独立的，也就是两个列表了实际上，所以对于几百万数据的大列表进行切片操作，对于主机而言是一个很大的考验，这点要和Go的切片区分开来，Go中的切片和原来的是共享的

list_v = [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15]  
list_v2 = list_v[1:-1:2]  
print(list_v2)   # [2, 4, 6, 8, 10, 12, 14]
list_v2[0] = 99  
print(list_v2)  # [99, 4, 6, 8, 10, 12, 14]
print(list_v) # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]

在NumPy的数组中这个就被舍弃了，NumPy中的切片和原数组是共享的，修改切片也会修改原数组，如果真的需要完全拷贝一份新的，就是借助copy()函数

List可以通过append()函数添加元素，也可以通过加法和乘法扩充和复制

list_v = [1,2,3,4]  
list_v.append(5)  
print(list_v)  #12345
print(list_v+[6])   #123456
print(list_v+[6,7,8])  #12345678
print(list_v*2) #1234512345

列表还有一个列表推导式，可以了解和学习一下 正常的我们创建由已知列表开平方得到新列表，我们需要借助for循环，代码如下

result = []
for i in [1,2,3,4]:
    result = result+[i**2]
print(result)

使用列表推导式之后可以这么写

result = [i**2 for i in [1,2,3,4]]

也还是加一点的if判断

result = []
for i in [1,2,3,4]:
    if i<3:
        result = result+[i**2]
print(result)
​
# 列表推导式写法
result = [i**2 for i in [1,2,3,4] if i<3]

tuple元组

元组的特点：有序；可重复；单个元素不可修改但是可重新赋值替换；下标访问

tuple_age = (1,2,3)  
# 这种创建方式极其简单方便
# 同时也可以看到，python中对于类型的要求没那么强
tuple_name = "张三","李四",99  tuple_null = ()  # 创建空元组
print(tuple_age)  
# tuple_age[0]=10 在运行的时候会报错
tuple_age = 3,3,3  
print(tuple_age[0]) # 下标访问单个元素
print(tuple_name)  
print(tuple_null)

除此之外，可以利用元组进行批量赋值，或者说就是Python的批量赋值的特性

a,b,c = 1,2,3  
e,f,g = (11,12.2,13.33)  
print(a,b,c)  
print(e,f,g)

当有多个数值的列表，然后你只需要部分数值的时候，你可以利用元组的快捷构造的特性进行截取

result1,result2,*other,result3 =1,2,3,4,5  
print(result1)  #1
print(result2)  #2
print(other)  #[3,4]
print(result3) #5

dict字典

一个Key-Value结构，不可重复，重复会覆盖；无序 ；Key唯一 ； Value可以为任何数据类型，Key只能是不可变的元素（数字，字符串，布尔，元组，不可变集合）

dict_v = {  
    "name":"zhangsan",  
    "age":18,  
    1:(1,2,3),  
    2:[1,2,3],  
    3:{1,2,3},  
    "4":True  
}  
# {'name': 'zhangsan', 'age': 18, 1: (1, 2, 3), 2: [1, 2, 3], 3: {1, 2, 3}, '4': True}
print(dict_v)

字典添加元素，可以直接通过dict[newKey]=newValue的方式进行新增

dict_v = {  
    "name":"zhangsan",  
    "age":18,  
}  
dict_v["newKey"] = "newValue"  
print(dict_v)

类型转换

集合，列表，元组，字典之间也可以进行转换，用到如下四个函数

• 转化为集合使用set()

• 转化为元组使用tuple()

• 转化为列表使用list()

• 转化为字典使用dist() 这里就展示图片，不附带代码了，可自行尝试，需要注意的是，其余的转化为字典的时候，因为字典是双列的，所以肯定还会需要一个单列集合，这个时候就需要借助zip(A,B)函数，该函数会将入参A和B进行一一对应，然后让A[0]作为Key，B[0]作为Value从而组成dict[0]，这里需要注意，传入的A和B如果是集合这种无序的，那么会造成匹配不一样按照预期一样

流程控制

if语句

在Python中，if语句的使用格式为如下，需要注意：

• Python中的函数，循环，if条件判断等都不会使用end来表示代码块的结束

• Python中使用缩进来表示代码块的范围

• 每一个判断条件的最后都有一个冒号，不要遗漏

if 判断条件:
    逻辑处理一
elif 判断条件:
    逻辑处理二
else:
    逻辑处理三

简单的分数区间判断案例

source = 85  
if source == 100:  
    print("满昏")  
elif source >= 90:  
    print("优秀")  
elif source >= 60:  
    print("及格")  
else:  
    print("不及格")

循环语句

Python中for循环的基本使用如下

for 变量名 in 可迭代对象:
    循环体逻辑
    
    
# 遍历列表，依次取出每个元素 
fruits = ["苹果", "香蕉", "橙子"] 
for fruit in fruits: 
    print(f"我喜欢吃：{fruit}") 
# 输出结果： # 我喜欢吃：苹果 # 我喜欢吃：香蕉 # 我喜欢吃：橙子

如果需要计数式循环，即例如Java中for(int i=0;i<10;i++),可以借助range(n)，该函数会创建一个0~n-1的整数序列供for循环使用,需要注意，range()函数也是左闭右开的区间范围

# 用法1：range(n) → 生成 0 ~ n-1 的整数序列  
for i in range(5):  
    print(i)  # 输出：0 1 2 3 4  
  
# 用法2：range(start, end) → 生成 start ~ end-1 的整数序列（左闭右开）  
for i in range(2, 7):  
    print(i)  # 输出：2 3 4 5 6  
  
# 用法3：range(start, end, step) → 步长为step，生成等差序列  
for i in range(0, 10, 2):  
    print(i)  # 输出：0 2 4 6 8（偶数）  
  
# 示例：通过索引遍历列表  
fruits = ["苹果", "香蕉", "橙子"]  
for i in range(len(fruits)):  
    print(f"索引{i}：{fruits[i]}")

for循环也常用于高级变量类型的遍历

# 1. 遍历列表 (List)print("\n【1】遍历列表 (List)")  
print("-" * 40)  
fruits = ["苹果", "香蕉", "橙子", "葡萄", "草莓"]  
print(f"列表内容: {fruits}")  
  
print("方法1: 直接遍历元素")  
for fruit in fruits:  
    print(f"  - {fruit}")  
  
print("\n方法2: 使用enumerate()获取索引和值")  
for index, fruit in enumerate(fruits):  
    print(f"  索引: {index}, 值: {fruit}")  
  
print("\n方法3: 使用range(len())遍历索引")  
for i in range(len(fruits)):  
    print(f"  索引: {i}, 值: {fruits[i]}")  
  
# 2. 遍历元组 (Tuple)print("\n【2】遍历元组 (Tuple)")  
print("-" * 40)  
colors = ("红色", "绿色", "蓝色", "黄色")  
print(f"元组内容: {colors}")  
  
print("方法1: 直接遍历元素")  
for color in colors:  
    print(f"  - {color}")  
  
print("\n方法2: 使用enumerate()获取索引和值")  
for index, color in enumerate(colors):  
    print(f"  索引: {index}, 值: {color}")  
  
# 3. 遍历集合 (Set)print("\n【3】遍历集合 (Set)")  
print("-" * 40)  
numbers = {1, 2, 3, 4, 5, 3, 2}  # 注意：集合会自动去除重复元素  
print(f"集合内容: {numbers}")  
  
print("方法1: 直接遍历元素")  
for num in numbers:  
    print(f"  - {num}")  
  
print("\n方法2: 对集合排序后遍历")  
for num in sorted(numbers):  
    print(f"  - {num} (排序后)")  
  
# 4. 遍历字典 (Dictionary)print("\n【4】遍历字典 (Dictionary)")  
print("-" * 40)  
student_scores = {  
    "张三": 95,  
    "李四": 87,  
    "王五": 92,  
    "赵六": 78  
}  
print(f"字典内容: {student_scores}")  
  
print("方法1: 遍历键(keys)")  
for name in student_scores:  
    print(f"  学生: {name}, 分数: {student_scores[name]}")  
  
print("\n方法2: 明确遍历键(keys)")  
for name in student_scores.keys():  
    print(f"  学生: {name}, 分数: {student_scores[name]}")  
  
print("\n方法3: 遍历值(values)")  
for score in student_scores.values():  
    print(f"  分数: {score}")  
  
print("\n方法4: 同时遍历键和值(items)")  
for name, score in student_scores.items():  
    print(f"  学生: {name}, 分数: {score}")  
  
# 5. 嵌套数据结构的遍历  
print("\n【5】嵌套数据结构的遍历")  
print("-" * 40)  
nested_data = {  
    "班级A": [85, 90, 78],  
    "班级B": [92, 88, 84],  
    "班级C": [79, 85, 91]  
}  
  
print("遍历嵌套字典和列表")  
for class_name, scores in nested_data.items():  
    print(f"  {class_name}:")  
    for i, score in enumerate(scores):  
        print(f"    学生{i+1}: {score}分")  
  
# 6. 使用条件语句的遍历  
print("\n【6】带条件判断的遍历")  
print("-" * 40)  
mixed_list = [1, "hello", 3.14, "world", 42, True, "python"]  
print(f"混合列表: {mixed_list}")  
  
print("只打印字符串类型的元素:")  
for item in mixed_list:  
    if isinstance(item, str):  
        print(f"  字符串: {item}")  
  
print("\n只打印数字类型的元素:")  
for item in mixed_list:  
    if isinstance(item, (int, float)):  
        print(f"  数字: {item}")  
  
print("\n" + "="*60)  
print("Python for循环遍历演示完成！")  
print("="*60)

while循环的基本格式如下

while 条件:
    循环体

break和continue的使用和其他语言一致，这里不过多讲述

函数

Python中的函数关键字为def，其格式如下

def 函数名(入参列表):
    """ 文档说明字符串 """
    函数体
    return 反参

首先我们来解释一下这个一开始文档说明字符串，这个需要搭配方法名.__doc__进行使用，当你在这个方法写了文档说明字符串的时候，你使用方法名.__doc__就能打印出这些内容，相当于做了方法说明文档，当然，这个不是必须的，如果不写，调用.__doc__会返回None

def add(x, y):  
    """计算x+y"""  
    return x + y  
  
print(add.__doc__) # 输出 计算x+y

函数的入参和反参可以是七种数据类型中的任意一种，并且Python的入参是值传递，也就是说函数入参是形参，不会对外部数据造成影响

def change(v):  
    v=v+1  
  
x = 10  
change(x)  
print(x) # 依旧是10

Python的函数可以多个入参和多个反参,多个反参可以利用多个变量接收，可以当作一个整体进行接收

def my_counter(a,b):  
    return a+b,a-b,a*b  
  
a,b,c= my_counter(1,2)  
print("和为",a,"差为",b,"积为",c)  
tuple_v = my_counter(1,2)  
print("和为",tuple_v[0],"差为",tuple_v[1],"积为",tuple_v[2])

需要传入可变数量的参数的时候，可以利用元组拆分法，利用*name作为入参

def my_counter(*args):  
    result = 0  
    for arg in args:  
        result += arg  
    return result  
  
print(my_counter(1, 2, 3, 4, 5))  #15
print(my_counter(1, 2, 3)) # 6

自然，也可以同时指定一个入参和剩下的不确定入参，需要注意的是，可变参数只能作为最后一个参数且一个函数只能有一个可变参数，虽然编译器不会报错，但是运行时会出错，因为无法确定边界

def my_counter(off,*args):  
    result = 0  
    for arg in args:  
        result += arg  
    return result+off  
  
print(my_counter(-10,1, 2, 3, 4, 5))  
print(my_counter(10,1, 2, 3))

上面通过*name实现不确定入参，可以理解为创建了一个name的列表然后允许我们一直往里面加东西，除此之外，还可以利用**name的方式，作为可变参数的键值对/字典

def plusDict(a,b,**dict_v):  
    """ 加入对Java和Golang的描述，也可以自定义加新的描述键值对 """    dict_v["Java"] = a  
    dict_v["Golang"] = b  
    return dict_v  
  
print(plusDict( "CRUD工程师","云原生的宠儿",Python="AI时代大哥大",Rust="高大上") )

在入参的时候，可以指定入参名从而不一定需要按照顺序传入参数

def my_counter(a, b, c):  
    return a + b * c  
print(my_counter(1,c=2,b=3))

在定义函数的时候，可以指定入参默认值，从而当不传入的时候，可以使用默认值

def yourSchool(name,level="本科"):  
    print("your school " + name + "is a " + level )  
  
yourSchool("HNUST")  # 不会报错
yourSchool("HNUST","985")

类

定义和普通使用

类的定义我们就不多说了，Python中的类一样具备封装，继承，多态的特性(据说在深度学习领域类一般是用来封装参数传递，其余特性用的少)，并且在Python中其实也是一切皆对象，就连基本数据类型int也是

a = 1
print(type(a)) # 输出：<class 'int'>

Python中类中的函数称之为方法，且约定俗成类的名称一般大写，Python的每个类必须包含一个__init__的初始化方法（简单理解为有参构造方法） init方法内的入参通过赋值就可以让类对象拥有成员a,b，也可以直接不通过入参来定义成员，这些成员具备默认值

class Counter:  
    """ 一个可以进行加减法的计算器 """    
    def __init__(self, a, b):  # 用于完成初始化的特殊方法  
        self.a = a  
        self.b = b
        self.name = "超级计算机"
  
  
    def plus(self):  # 类中的普通方法  
        return self.a + self.b  
    def minus(self):  # 类中的普通方法  
        return self.a - self.b
    def zwjs(self):  
        print("欢迎使用", self.name)

可以看到，无论是自定义普通方法还是init初始化方法，都有一个self参数，这个参数用于链接类内部和外部的信息，大家可以简单的理解为this指针，而且从上面也可也看到，Python的类的成员属性貌似可以不要定义，会根据__init__方法自动定义在self中，需要的话直接去self中拿就行了

类定义完成后，就可以创建类的对象以及调用其成员方法和成员属性

counter1 = Counter(1,2)  
print(counter1.a, counter1.b)  
print(counter1.plus())  
print(counter1.minus())  
counter1.zwjs()

继承

利用继承，可以让子类继承父类的属性和方法 Python的继承的表现形式为

class 类名(父类名):

在Python中还能多继承

class 类名(fu1,fu2)

子类需要在__init__方法中调用super()函数从而实现继承属性和方法，对于父类中已有属性和方法子类中再次定义的话就会覆盖重写

class Counter2(Counter):  
    """ 可以进行加减乘除的计算器 """  
    def __init__(self, a, b):  
        super().__init__(a, b)  # 这个不写其实也没事，不写的话就不能使用父类的东西，只要不使用父类的东西就不会报错
        self.name = "超级计算机二代"  
  
    def multiply(self):  
        return self.a * self.b  
    def divide(self):  
        return self.a / self.b  
  
    def zwjs(self):  
        print("您好,欢迎使用", self.name)
        
        
counter = Counter2(1, 2)  
print(counter.plus())  
print(counter.minus())  
print(counter.multiply())  
print(counter.divide())  
counter.zwjs()

我们知道如下几个点

1. Nacos客户端会和服务端进行心跳同步

2. Nacos服务端在集群环境下采用单个责任节点+多个从节点的方式管理，责任节点负责负责健康检查，心跳处理等操作，从节点仅仅担任数据同步工作

3. 一个微服务实例连接集群的条件下，请求可能会因为负载策略发送到任意一个Nacos节点上

那么综合上述，我们可以引出一个问题：如果客户端的心跳请求发送到了非责任节点会怎么处理？

这个地方可以参考Nacos仓库下面的一个issue

server端收到client beat请求后，不更新last beat时间，导致实际被判定为unhealthy · Issue #5363 · alibaba/nacos

针对上述问题，我们来举个例子：假设现在我们有三个集群节点，member1、member2、member3，又假设 member1 是负责 sotck-service 健康检查任务的节点。
​
stock-serivce 客户端发送心跳请求的时候，是随机发送到集群中某一个节点上的，假设 stock-service 没有给 member1 发送心跳请求，如果 lastBeat 属性不同步给集群节点，那么 member1 在做健康检查的时候，是不是会判断 stock-service 为不健康的实例？

我们可以看一下issue上大佬们关于这个问题的讨论与回复

按照大佬的意思就是：心跳请求应该是随机发送到集群中某一台上，然后通过distro协议转发到该服务的责任节点去统一更新实例心跳时间。且心跳时间不会进行同步。

我们可以自己扒一下源码来了解一下

我们来看到服务端对于心跳的处理这一块，可以看到有一个@CanDistro的注解

这个注解是一个自定义注解，但是他的作用好像不是很明了，我们可以看一下哪个地方用到了这个注解，发现这里有个Filter过滤器中用到了

我们来看一下这个过滤器，可以看到最终其匹配的URL是所有的/v1/ns/*的路径

接下来我们就可以来看一下Filter中的过滤逻辑，代码很长，但是我们主要看一下重要的部分，即有我写了注释的那一行其实就行了

@Override
public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain)
        throws IOException, ServletException {
    ReuseHttpRequest req = new ReuseHttpServletRequest((HttpServletRequest) servletRequest);
    HttpServletResponse resp = (HttpServletResponse) servletResponse;
    
    String urlString = req.getRequestURI();
    
    if (StringUtils.isNotBlank(req.getQueryString())) {
        urlString += "?" + req.getQueryString();
    }
    
    try {
        String path = new URI(req.getRequestURI()).getPath();
        String serviceName = req.getParameter(CommonParams.SERVICE_NAME);
​
        if (StringUtils.isBlank(serviceName)) {
            serviceName = req.getParameter("dom");
        }
        
        if (StringUtils.isNotBlank(serviceName)) {
            serviceName = serviceName.trim();
        }
        Method method = controllerMethodsCache.getMethod(req);
        
        if (method == null) {
            throw new NoSuchMethodException(req.getMethod() + " " + path);
        }
        
        String groupName = req.getParameter(CommonParams.GROUP_NAME);
        if (StringUtils.isBlank(groupName)) {
            groupName = Constants.DEFAULT_GROUP;
        }
        
​
        String groupedServiceName = serviceName;
        if (StringUtils.isNotBlank(serviceName) && !serviceName.contains(Constants.SERVICE_INFO_SPLITER)) {
            groupedServiceName = groupName + Constants.SERVICE_INFO_SPLITER + serviceName;
        }
        
// 判断是否带有CanDistro注解 && responsible方法即判断是否为责任节点
        if (method.isAnnotationPresent(CanDistro.class) && !distroMapper.responsible(groupedServiceName)) {
            
            String userAgent = req.getHeader(HttpHeaderConsts.USER_AGENT_HEADER);
            
            if (StringUtils.isNotBlank(userAgent) && userAgent.contains(UtilsAndCommons.NACOS_SERVER_HEADER)) {
​
                Loggers.SRV_LOG.error("receive invalid redirect request from peer {}", req.getRemoteAddr());
                resp.sendError(HttpServletResponse.SC_BAD_REQUEST,
                        "receive invalid redirect request from peer " + req.getRemoteAddr());
                return;
            }
            // 得到目标的真正的责任节点
            final String targetServer = distroMapper.mapSrv(groupedServiceName);
            
            List<String> headerList = new ArrayList<>(16);
            Enumeration<String> headers = req.getHeaderNames();
            while (headers.hasMoreElements()) {
                String headerName = headers.nextElement();
                headerList.add(headerName);
                headerList.add(req.getHeader(headerName));
            }
            
            final String body = IoUtils.toString(req.getInputStream(), Charsets.UTF_8.name());
            final Map<String, String> paramsValue = HttpClient.translateParameterMap(req.getParameterMap());
            
            RestResult<String> result = HttpClient
                    .request("http://" + targetServer + req.getRequestURI(), headerList, paramsValue, body,
                            PROXY_CONNECT_TIMEOUT, PROXY_READ_TIMEOUT, Charsets.UTF_8.name(), req.getMethod());
            String data = result.ok() ? result.getData() : result.getMessage();
            try {
                WebUtils.response(resp, data, result.getCode());
            } catch (Exception ignore) {
                Loggers.SRV_LOG.warn("[DISTRO-FILTER] request failed: " + distroMapper.mapSrv(groupedServiceName)
                        + urlString);
            }
        } else {
            OverrideParameterRequestWrapper requestWrapper = OverrideParameterRequestWrapper.buildRequest(req);
            requestWrapper.addParameter(CommonParams.SERVICE_NAME, groupedServiceName);
            filterChain.doFilter(requestWrapper, resp);
        }
    } catch (AccessControlException e) {
        resp.sendError(HttpServletResponse.SC_FORBIDDEN, "access denied: " + ExceptionUtil.getAllExceptionMsg(e));
    } catch (NoSuchMethodException e) {
        resp.sendError(HttpServletResponse.SC_NOT_IMPLEMENTED,
                "no such api:" + req.getMethod() + ":" + req.getRequestURI());
    } catch (Exception e) {
        resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
                "Server failed," + ExceptionUtil.getAllExceptionMsg(e));
    }
    
}

那么可以看到，所有的标记了@CanDistro注解的请求，都会请求的时候判断：如果当前处理请求的集群节点，不是这个 service 的责任节点，那么这个过滤器就会转发到对应的责任节点上做处理。

那我们回到最开始的问题上：lastBeat 属性是如何同步的？

心跳检测在请求到达节点后就能自行检测是否责任节点上，如果不在则会直接发送给责任节点，而不是等待全体同步，所以可以理解为心跳时间是定向同步而非全体同步，心跳可能会发送到非责任节点，但是非责任节点一定会检测并且转发给责任节点，责任节点对于心跳的数据不会也没有必要去同步到别的节点上，因为别的节点完全不需要也不负责

整体结构分析

我们可以来看一下之前我们已经了解过的几个Event类对象，其实可以发现，都是继承自Event这个事件父类

同样的，我们再去看一下所有的Handler所在类，也可以发现，有些继承自SmartSubscriber，并且SmartSubscriber之上还有个顶级接口Subscriber，就是订阅者，在这里定义了顶级抽象方法

public abstract List<Class<? extends Event>> subscribeTypes();

该方法的主要作用就是返回该订阅者感兴趣的Event类型的List集合

除此之外在顶级接口中Subscriber中还有一个onEvent()方法，该方法就是当对应的Event来了之后进行处理回调的方法

在每个子类的构造方法上，还会调用一个registerSubscriber方法，该方法是NotifyCenter中的静态方法

NotifyCenter.registerSubscriber(this);
NotifyCenter.registerSubscriber(this, NamingEventPublisherFactory.getInstance());

根据如上的机构，其实我们完全可以自己基于Nacos提供的事件驱动架构，自己写一些事件和订阅者进行运行

如下，我们先写一个TestEvent测试事件类

import com.alibaba.nacos.common.notify.Event;
​
public class TestEvent extends Event {
}
​

然后写一个对应的订阅者TestSubscriber，实现subscribeTypes和onEvent方法，在构造方法中将该订阅者注册上去

​
import com.alibaba.nacos.common.notify.Event;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.notify.listener.SmartSubscriber;
import org.springframework.stereotype.Component;
​
import java.util.LinkedList;
import java.util.List;
​
@Component
public class TestSubscriber extends SmartSubscriber {
​
    public TestSubscriber() {
        NotifyCenter.registerSubscriber(this);
    }
​
    @Override
    public List<Class<? extends Event>> subscribeTypes() {
        List<Class<? extends Event>> result = new LinkedList<>();
        result.add(TestEvent.class);
        return result;
    }
​
    @Override
    public void onEvent(Event event) {
        System.out.println("TestSubscriber onEvent");
    }
}
​

我们写一个测试类，来测试一下上面的代码，可以看到，调用完publishEvent之后，对应的订阅者的onEvent方法也成功被调用且打印了输出

源码分析

事件发布

上面我们了解一下事件驱动架构的大致结构，我们现在就可以开始分析，我们先从事件的发布这块入手

我们知道，发布事件用的是NotifyCenter.publishEvent这个方法，我们来看下这个方法的源码

NotifyCenter.publishEvent(new TestEvent());
​
​
public static boolean publishEvent(final Event event) {
    try {
        // 跟主线任务
        return publishEvent(event.getClass(), event);
    } catch (Throwable ex) {
        LOGGER.error("There was an exception to the message publishing : ", ex);
        return false;
    }
}
​
​
private static boolean publishEvent(final Class<? extends Event> eventType, final Event event) {
    if (ClassUtils.isAssignableFrom(SlowEvent.class, eventType)) {
        return INSTANCE.sharePublisher.publish(event);
    }
    
    // 根据Event的类型的到事件类型Topic
    final String topic = ClassUtils.getCanonicalName(eventType);
    // 根据发布事件类型获取 EventPublisher 对象，该对象中会包含订阅者信息
    // 订阅者怎么来的？ 从这里可以看到肯定是从一个Map中获取的，具体的在下文章节会讲到，publisherMap其实就是一个以Event全类名为key，EventPublisher对象(该对象成员属性subscribers会保存对应的订阅者类)为value的map，
    EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    if (publisher != null) {
        // 跟主线任务
        return publisher.publish(event);
    }
    LOGGER.warn("There are no [{}] publishers for this event, please register", topic);
    return false;
}
​
​
@Override
public boolean publish(Event event) {
    checkIsStart();
    // 把事件放入到了一个 阻塞队列
    boolean success = this.queue.offer(event);
    // 如果没有放入成功则手动调用receiveEvent方法接收并通知Subscriber处理该事件
    if (!success) {
        LOGGER.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);
        receiveEvent(event);
        return true;
    }
    return true;
}
​
​

可以看到，又是利用到了阻塞队列，自然我们就可以去找到队列中拿任务的逻辑

可以发现就在DefaultPublisher中，在这一段代码中，我们有看到 subscribers 一个 Set 集合，注意，这里并不是全部的订阅者，只是当前发布的事件对应的订阅者，并不会通知全部的订阅者哈

for (; ; ) {
    if (shutdown) {
        break;
    }
    // 从阻塞队列取数据
    final Event event = queue.take();
    // 处理事件
    receiveEvent(event);
    UPDATER.compareAndSet(this, lastEventSequence, Math.max(lastEventSequence, event.sequence()));
}
​
​
protected final ConcurrentHashSet<Subscriber> subscribers = new ConcurrentHashSet<>();
​
void receiveEvent(Event event) {
    final long currentEventSequence = event.sequence();
    
    if (!hasSubscriber()) {
        LOGGER.warn("[NotifyCenter] the {} is lost, because there is no subscriber.", event);
        return;
    }
    
    // 循环遍历通知发布事件对应的订阅者
    for (Subscriber subscriber : subscribers) {
        // Whether to ignore expiration events
        if (subscriber.ignoreExpireEvent() && lastEventSequence > currentEventSequence) {
            LOGGER.debug("[NotifyCenter] the {} is unacceptable to this subscriber, because had expire",
                    event.getClass());
            continue;
        }
        
        // Because unifying smartSubscriber and subscriber, so here need to think of compatibility.
        // Remove original judge part of codes.
        //通知
        notifySubscriber(subscriber, event);
    }
}
​

然后最后调用了 notifySubscriber(subscriber, event); 方法来通知订阅者，代码如下:

通过这段代码，我们能够知道，最后是使用线程异步的方式，来通知订阅者的。 事件发布流程我们就分析完了，核心关键还是利用了 阻塞队列 + 异步任务 来进行实现的，那么接下来我们一起来看看，订阅者注册的流程。

@Override
public void notifySubscriber(final Subscriber subscriber, final Event event) {
    
    LOGGER.debug("[NotifyCenter] the {} will received by {}", event, subscriber);
    
    // 创建一个任务，就是调用订阅者的 onEvent 方法
    final Runnable job = () -> subscriber.onEvent(event);
    final Executor executor = subscriber.executor();
    
    if (executor != null) {
        // 执行任务
        executor.execute(job);
    } else {
        try {
            job.run();
        } catch (Throwable e) {
            LOGGER.error("Event callback exception: ", e);
        }
    }
}
​

订阅者注册

看完了事件的发布，我们来看订阅者这边

每个订阅者在构造方法中都需要先将自己注册，所以我们先来看注册部分的代码，也就是NotifyCenter.registerSubscriber()

可以看到，首先这个就是调用subscribeTypes获取当前订阅者订阅的Event类型的List列表，然后遍历这个列表，调用INSTANCE.sharePublisher.addSubscriber封装保存或者调用addSubscriber这个方法(两种情况的底层都将内容保存到了subscribers这个Set中)

public TestSubscriber() {
    NotifyCenter.registerSubscriber(this);
}
​
public static void registerSubscriber(final Subscriber consumer) {
    registerSubscriber(consumer, DEFAULT_PUBLISHER_FACTORY);
}
​
public static void registerSubscriber(final Subscriber consumer, final EventPublisherFactory factory) {
    // If you want to listen to multiple events, you do it separately,
    // based on subclass's subscribeTypes method return list, it can register to publisher.
    if (consumer instanceof SmartSubscriber) {
        // 在这里会调用 subscribeTypes 方法，来获取我们需要监听的事件，然后进行遍历
        for (Class<? extends Event> subscribeType : ((SmartSubscriber) consumer).subscribeTypes()) {
            // For case, producer: defaultSharePublisher -> consumer: smartSubscriber.
            if (ClassUtils.isAssignableFrom(SlowEvent.class, subscribeType)) {
                INSTANCE.sharePublisher.addSubscriber(consumer, subscribeType);
            } else {
                // For case, producer: defaultPublisher -> consumer: subscriber.
                // 添加订阅者
                addSubscriber(consumer, subscribeType, factory);
            }
        }
        return;
    }
    
    // 这里代码省略
}
​
​

接下来我们继续看addSubscriber方法

private static void addSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType,
        EventPublisherFactory factory) {
    
    final String topic = ClassUtils.getCanonicalName(subscribeType);
    synchronized (NotifyCenter.class) {
        // 这里很关键，会创建 EventPublisher 对象，一个事件会对应一个 EventPublisher 一个对象
        MapUtil.computeIfAbsent(INSTANCE.publisherMap, topic, factory, subscribeType, ringBufferSize);
    }
    // 获取事件对应的 EventPublisher 对象
    EventPublisher publisher = INSTANCE.publisherMap.get(topic);
    if (publisher instanceof ShardedEventPublisher) {
        ((ShardedEventPublisher) publisher).addSubscriber(consumer, subscribeType);
    } else {
        // 往 EventPublisher 对象添加订阅者信息
        publisher.addSubscriber(consumer);
    }
}
​

其中MapUtil.computeIfAbsent的逻辑如下，可以看到，在这里会利用传入的第三个参数，调用其apply方法，利用第四个和第五个参数创建出一个EventPublisher对象，这个对象内部会存储我们的订阅者信息，也就是 subscribers 属性

@NotThreadSafe
public static <K, C, V, T> V computeIfAbsent(Map<K, V> target, K key, BiFunction<C, T, V> mappingFunction, C param1,
        T param2) {
    
    Objects.requireNonNull(target, "target");
    Objects.requireNonNull(key, "key");
    Objects.requireNonNull(mappingFunction, "mappingFunction");
    Objects.requireNonNull(param1, "param1");
    Objects.requireNonNull(param2, "param2");
    
    V val = target.get(key);
    if (val == null) {
        V ret = mappingFunction.apply(param1, param2);
        target.put(key, ret);
        return ret;
    }
    return val;
}

在代码的最后，会调用 addSubscriber 方法往subscribers 属性中添加订阅者，完成注册。

总结

Nacos2.X中的事件驱动架构，整体基于Event(事件)-Subscriber(订阅者)-EventPublish(事件发布者)-NotifyCenter(事件中心)

订阅者通过注册中心进行注册，通过订阅者subscribeTypes()方法来获取其关注的Event事件类型的集合，对于非慢事件利用工厂创建对应的EventPublisher，并且将订阅者保存到EventPublisher 的 subscribers 集合，在publisherMap中保存事件-事件发布者的对应关系，EventPublisher 初始化时启动异步线程，监听阻塞队列，等待事件到来

通过事件中心发布事件的时候，根据事件的类型获得其全类名作为topic，利用topic从publisherMap中获取对应的EventPublisher，利用EventPublisher将任务写入到阻塞任务队列中

客户端的相关操作

怎么找到对应的代码块这边我们就不过多表述了，肯定还是主要围绕AbstractAutoServiceRegistration这个自动注册相关的类进行的操作

我们可以看到，在这个类下，有一个@PreDestroy注解标识的销毁方法，该方法内部就是调用了stop方法

stop方法的逻辑如下

可以看到，整体内容和Nacos1.4版本的逻辑操作相差不大，对客户端保存的注册实例遍历然后依次执行一些注销前置操作，注销后置操作等等

private List<RegistrationLifecycle<R>> registrationLifecycles = new ArrayList<>();
​
public void stop() {
    if (this.getRunning().compareAndSet(true, false) && isEnabled()) {
​
        this.registrationLifecycles.forEach(
                registrationLifecycle -> registrationLifecycle.postProcessBeforeStopRegister(getRegistration()));
        deregister();
        this.registrationLifecycles.forEach(
                registrationLifecycle -> registrationLifecycle.postProcessAfterStopRegister(getRegistration()));
        if (shouldRegisterManagement()) {
            this.registrationManagementLifecycles
                    .forEach(registrationManagementLifecycle -> registrationManagementLifecycle
                            .postProcessBeforeStopRegisterManagement(getManagementRegistration()));
            deregisterManagement();
            this.registrationManagementLifecycles
                    .forEach(registrationManagementLifecycle -> registrationManagementLifecycle
                            .postProcessAfterStopRegisterManagement(getManagementRegistration()));
        }
        this.serviceRegistry.close();
    }
}

在最后deregisterManagement()中最后会发送一个Rpc的请求发送给服务端

服务端处理Rpc请求

依旧是先找到对应的Handler，依旧是熟悉的switch-case对于不同类型的区分处理

对应的deregisterInstance()方法的代码如下

private InstanceResponse deregisterInstance(Service service, InstanceRequest request, RequestMeta meta) {
    clientOperationService.deregisterInstance(service, request.getInstance(), meta.getConnectionId());
    return new InstanceResponse(NamingRemoteConstants.DE_REGISTER_INSTANCE);
}
​
​
@Override
public void deregisterInstance(Service service, Instance instance, String clientId) {
    if (!ServiceManager.getInstance().containSingleton(service)) {
        Loggers.SRV_LOG.warn("remove instance from non-exist service: {}", service);
        return;
    }
    Service singleton = ServiceManager.getInstance().getSingleton(service);
    // 获取 client 对象
    Client client = clientManager.getClient(clientId);
    if (!clientIsLegal(client, clientId)) {
        return;
    }
    // 移除 instance 信息，在这里面会操作 publishers 那个Map，以及发 ClientChangedEvent 事件，来同步集群节点
    InstancePublishInfo removedInstance = client.removeServiceInstance(singleton);
    client.setLastUpdatedTime();
    if (null != removedInstance) {
        // 发布客户端注销事件
        NotifyCenter.publishEvent(new ClientOperationEvent.ClientDeregisterServiceEvent(singleton, clientId));
        NotifyCenter.publishEvent(
                new MetadataEvent.InstanceMetadataEvent(singleton, removedInstance.getMetadataId(), true));
    }
}
​

对于客户端注销事件的处理ClientDeregisterServiceEvent，同样的搜索对应的handler的相关逻辑

剩下的部分也就是对于内存注册表，订阅表等的一些remove操作以及同步集群了

总结-Nacos2.x注册中心部分

1. 对于临时实例，会采用RPC的方式进行通讯

2. 内存注册表的结构变化。由原本的嵌套式Map转化为了平铺式Map，结构更加i清晰

3. 客户端查询微服务实例列表的时候，调用的是客户端订阅的请求，在服务端会有订阅表，当我们被订阅的服务发生了变动，会来通知订阅者。

4. 可以明显感觉到，2.0中的事件发布操作明显更多了

注意，这里只是说服务端，即服务都安如何去检测客户端的健康状态，而不是客户端的心跳以及也不是集群中对于每个节点的健康检测。

我们可以来看到ConnectionManager，在这个类的内部有一个Map保存了所有的了所有的Connection连接对象，连接对象有个元数据的成员对象ConnectionMeta，他有一个成员属性为lastActiveTime(最后存活时间)

在ConnectionManager对象中，有个@PostConstruct注解标记的start方法，该方法会开启一个run任务，在该方法中，遍历上述的map并且用now和元数据中的lastActiveTime进行比较判断是否存活/过时,如果发现过时了，就会放入到outDatedConnections这个Set集合中

outDatedConnections在没有超出上限的时候，不会立马被清除，而是会统一在最后进行二次判断是否真正下线，如果没有下线，则会刷新activeTime并且加入到存活对象集合中，这个过程为了保证并发安全还是用了countDownLatch，这个被称之为探活机制

如果上述二次请求遍历完毕后依旧集合中依旧还有数据，则调用unregister()取消注册进行删除

unregister源码如下，首先会从connections这个管理了所有的连接的Map中remove，remove的时候会返回对应的Connection连接实例信息，根据实例信息中的Ip信息操作连接数，最后通知客户端注销连接

public synchronized void unregister(String connectionId) {
    // 移除客户端信息
    Connection remove = this.connections.remove(connectionId);
    if (remove != null) {
        String clientIp = remove.getMetaInfo().clientIp;
        AtomicInteger atomicInteger = connectionForClientIp.get(clientIp);
        if (atomicInteger != null) {
            int count = atomicInteger.decrementAndGet();
            if (count <= 0) {
                connectionForClientIp.remove(clientIp);
            }
        }
        remove.close();
        Loggers.REMOTE_DIGEST.info("[{}]Connection unregistered successfully. ", connectionId);
        // 通知客户端注销连接
        clientConnectionEventListenerRegistry.notifyClientDisConnected(remove);
    }
}

来看一下通知客户端注销连接的逻辑,可以看到最后是发布了一个ClientDisconnectEvent的事件，这个事件的处理我们在上一节的内容中分析过了，客户端会注销，同步到集群，当前节点关于该链接的数据，订阅者列表，注册表信息等等也都会被删除

// 1
clientConnectionEventListenerRegistry.notifyClientDisConnected(remove);
​
// 2
public void notifyClientDisConnected(final Connection connection) {
    
    for (ClientConnectionEventListener clientConnectionEventListener : clientConnectionEventListeners) {
        try {
        
            // 客户端注销，我们看这个实现类，ConnectionBasedClientManager
            clientConnectionEventListener.clientDisConnected(connection);
        } catch (Throwable throwable) {
            Loggers.REMOTE.info("[NotifyClientDisConnected] failed for listener {}",
                    clientConnectionEventListener.getName(), throwable);
        }
    }
}
​
// 3 ConnectionBasedClientManager 实现类
@Override
public void clientDisConnected(Connection connect) {
    clientDisconnected(connect.getMetaInfo().getConnectionId());
}
​
// 4
@Override
public boolean clientDisconnected(String clientId) {
    Loggers.SRV_LOG.info("Client connection {} disconnect, remove instances and subscribers", clientId);
    ConnectionBasedClient client = clients.remove(clientId);
    if (null == client) {
        return true;
    }
    client.release();
    // 最后发布客户端注销事件
    NotifyCenter.publishEvent(new ClientEvent.ClientDisconnectEvent(client));
    return true;
}
​

在看Nacos服务端处理服务注册的时候，我们最后是探索到了AbstractClient类下的addServiceInstance()方法

@Override
public boolean addServiceInstance(Service service, InstancePublishInfo instancePublishInfo) {
    // 把 instance 信息放入到 publishers 中
    if (null == publishers.put(service, instancePublishInfo)) {
        MetricsMonitor.incrementInstanceCount();
    }
    // 同步集群节点数据
    NotifyCenter.publishEvent(new ClientEvent.ClientChangedEvent(this));
    Loggers.SRV_LOG.info("Client change for service {}, {}", service, getClientId());
    return true;
}

在该方法的最后那个pushlishEvent()方法中，发布了一个ClientChangedEvent，从命名和业务逻辑来看，这一步肯定是发布客户端改变事件用于同步别的集群节点，所以我们这一篇的内容就是来了解一下这个

处理ClientChangedEvent

还是老样子，我们对于Event事件类型的处理，都是查找对应的处理Event的地方，利用Idea的全局搜索轻松实现

private void syncToAllServer(ClientEvent event) {
    Client client = event.getClient();
    // Only ephemeral data sync by Distro, persist client should sync by raft.
    // 临时实例是走 distro 协议，持久化实例走 raft 协议
    // isResponsibleClient 这里判断了只有该 client 的责任节点有权利进行集群数据同步
    if (null == client || !client.isEphemeral() || !clientManager.isResponsibleClient(client)) {
        return;
    }
    // 客户端注销集群节点同步事件
    if (event instanceof ClientEvent.ClientDisconnectEvent) {
        // 这里有常量默认值TYPE，其数值为 Nacos:Naming:v2:ClientData 
        DistroKey distroKey = new DistroKey(client.getClientId(), TYPE);
        distroProtocol.sync(distroKey, DataOperation.DELETE);
        
    // 客户端改变集群节点同步事件
    } else if (event instanceof ClientEvent.ClientChangedEvent) {
         // 这里有常量默认值TYPE，其数值为 Nacos:Naming:v2:ClientData 
        DistroKey distroKey = new DistroKey(client.getClientId(), TYPE);
        distroProtocol.sync(distroKey, DataOperation.CHANGE);
    }
}
​

通过上面这一段代码我们可以得知，不管是客户端注销了，还是客户端改变了，都会来同步集群节点，它们两个都是调用了 distroProtocol.sync 来进行同步，唯一的区别就是参数一个是 DELETE、一个是CHANGE。那么我们接下来一个看下同步的具体逻辑吧，代码如下：

public void sync(DistroKey distroKey, DataOperation action) {
    sync(distroKey, action, DistroConfig.getInstance().getSyncDelayMillis());
}
​
public void sync(DistroKey distroKey, DataOperation action, long delay) {
    // 遍历集群其他节点，除开自身节点
    for (Member each : memberManager.allMembersWithoutSelf()) {
        syncToTarget(distroKey, action, each.getAddress(), delay);
    }
}
​
public void syncToTarget(DistroKey distroKey, DataOperation action, String targetServer, long delay) {
    DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),
            targetServer);
    // 创建了 DistroDelayTask 任务，并且把 targetServer(即集群节点的address) 包装成 DistroKey 当作参数传入了进去
    DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);
   // 往执行引擎中添加任务
   distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask);
    if (Loggers.DISTRO.isDebugEnabled()) {
        Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, targetServer);
    }
}
​

可以看到，这里最后利用distroTaskEngineHolder.getDelayTaskExecuteEngine()去执行distroDelayTask任务，这个地方的逻辑我们在上面就说过，实际上执行的就需要看对应的processor，而DistroTaskEngineHolder中的执行器就是DistroDelayTaskProcessor，我们看看对应的处理方法

@Override
public boolean process(NacosTask task) {
    if (!(task instanceof DistroDelayTask)) {
        return true;
    }
    DistroDelayTask distroDelayTask = (DistroDelayTask) task;
    DistroKey distroKey = distroDelayTask.getDistroKey();
    switch (distroDelayTask.getAction()) {
        case DELETE:
            // 删除任务
            DistroSyncDeleteTask syncDeleteTask = new DistroSyncDeleteTask(distroKey, distroComponentHolder);
            distroTaskEngineHolder.getExecuteWorkersManager().addTask(distroKey, syncDeleteTask);
            return true;
        case CHANGE:
        case ADD:
            // 改变、新增都会走这一段逻辑，封装为DistroSyncChangeTask
            DistroSyncChangeTask syncChangeTask = new DistroSyncChangeTask(distroKey, distroComponentHolder);
            distroTaskEngineHolder.getExecuteWorkersManager().addTask(distroKey, syncChangeTask);
            return true;
        default:
            return false;
    }
}

这一段逻辑，不管是删除，还是改变和新增的操作，都是创建了一个任务，添加到了 distro 任务执行引擎，这个引擎和之前看的延时任务执行实现有点不一样。

从逻辑上找下去，我们首先需要看到DistroExecuteTaskExecuteEngine中的addTask()的逻辑

可以发现，就是将task加入到TaskExecuteWorker类型的数组中，然后调用TaskExecuteWorker.process方法执行task

然后我们来看看process中的逻辑

可以看到这里面的主要逻辑就是将task放入到队列中。

在延时任务执行引擎实现原理是有一个 Map tasks 任务池，很多对应的任务处理器，然后定时从任务池获取任务，执行任务。而这个 distroTaskEngineHolder实现原理使用阻塞队列 + 异步任务的方式来实现的，所以最终这个任务肯定会被拿取并且执行

DistroSyncChangeTask

我们回到DistroDelayTaskProcessor中的process逻辑，可以知道无论是ADD还是CHANGE对应的都是封装为了DistroSyncChangeTask任务类型，我们查看其设计，可以发现继承自AbstractDistroExecuteTask，继续往上看其实就是实现了Runnable接口，所以我们来看看对应的run逻辑，因为我们是CHANGE类型，所以我们这里也肯定去看DistroSyncChangeTask下的doExecuteWithCallback的逻辑

@Override
protected void doExecuteWithCallback(DistroCallback callback) {
    
    String type = getDistroKey().getResourceType();
    // 获取请求数据
    DistroData distroData = getDistroData(type);
    if (null == distroData) {
        Loggers.DISTRO.warn("[DISTRO] {} with null data to sync, skip", toString());
        return;
    }
    // syncData 同步集群节点
    getDistroComponentHolder().findTransportAgent(type)
            .syncData(distroData, getDistroKey().getTargetServer(), callback);
}
​

此时我们应该去看syncData的逻辑，该方法也是接口中的方法，需要在对应的实现类中查找对应的实现，这里有两个实现类

怎么知道是哪个呢？我们可以回头来看一下，调用syncData的是findTransportAgent(type)的返回指，这个返回值其实在之前的run的逻辑中也有被调用，其返回类型是DistroTransportAgent，而DistroClientTransportAgent是DistroTransportAgent的子类

看到这里我们知道了，接下来要看的内容是DistroClientTransportAgent的syncData()的逻辑

@Override
public void syncData(DistroData data, String targetServer, DistroCallback callback) {
    if (isNoExistTarget(targetServer)) {
        callback.onSuccess();
        return;
    }
    // 创建请求对象
    DistroDataRequest request = new DistroDataRequest(data, data.getType());
    
    // 找到集群节点
    Member member = memberManager.find(targetServer);
    try {
        // 发送 rpc 异步请求
        clusterRpcClientProxy.asyncRequest(member, request, new DistroRpcCallbackWrapper(callback, member));
    } catch (NacosException nacosException) {
        callback.onFailed(nacosException);
    }
}
​

所以看到这里我们差不多知道了整个某个节点上的服务实例修改后，如果通知到其余节点的过程，捋一捋，一开始我们是从 ClientChangedEvent 事件开始分析，我们找到了处理这个事件的逻辑，调用了 sync 方法，在这个方法中会遍历其他集群节点进行数据同步，同步的逻辑如下。

1. 先是创建了 DistroDelayTask 延迟任务，放入到了延迟任务执行引擎，由 DistroDelayTaskProcessor 处理器来处理。

2. 在 DistroDelayTaskProcessor 处理器，处理的逻辑中又创建了 DistroSyncChangeTask 线程任务，执行引擎是先调用了 AbstractDistroExecuteTask 父类中的 run 方法，在 run 方法中又调用了子类的doExecuteWithCallback 方法。

3. doExecuteWithCallback 方法中会去获取最新的微服务实例列表，通过 rpc 发起异步请求进行数据同步。

集群其余节点处理DistroDataRequest

上述内容中，我们已经清楚了集群节点发其同步请求，那么现在我们来看看别的集群节点如何处理这个同步请求的。我们知道，对应的请求类型是DistroDataRequest,老规矩，利用IDEA全局搜索找到对应的Handler处理器

可以发现，无论是ADD 还是 CHANGE 还是 DELETE 的逻辑，最终都是执行的handlerSyncData(...)的逻辑，跟下去整个逻辑链路大概是handleSyncData()->onReceive()->processData->handlerClientSyncData()

private DistroDataResponse handleSyncData(DistroData distroData) {
    // 创建response对象
    DistroDataResponse result = new DistroDataResponse();
    // onReceive方法处理请求数据distroData
    if (!distroProtocol.onReceive(distroData)) {
        result.setErrorCode(ResponseCode.FAIL.getCode());
        result.setMessage("[DISTRO-FAILED] distro data handle failed");
    }
    return result;
}
​
​
/**
 * Receive synced distro data, find processor to process.
 *
 * @param distroData Received data
 * @return true if handle receive data successfully, otherwise false
 */
public boolean onReceive(DistroData distroData) {
    Loggers.DISTRO.info("[DISTRO] Receive distro data type: {}, key: {}", distroData.getType(),
            distroData.getDistroKey());
    String resourceType = distroData.getDistroKey().getResourceType();
    // 利用type找到对应的processor处理器
    DistroDataProcessor dataProcessor = distroComponentHolder.findDataProcessor(resourceType);
    if (null == dataProcessor) {
        Loggers.DISTRO.warn("[DISTRO] Can't find data process for received data {}", resourceType);
        return false;
    }
    // 调用处理器的processData方法处理请求数据
    return dataProcessor.processData(distroData);
}
​

processData也是接口中的方法，一样的有两个实现类，这里我们需要根据resourceType来进行确认，这个resourceType的值大家可以回过头看这一章的最开始那个地方，在syncToAllServer中，也就是我们Event创建和传入的地方，我们当前这个resourceType和DistroKey其实一开始就绑定了，那么其实这个resourceType就是我们上面提到的常量TYPE

然后这个地方通过findDataProcessor查找DistroComponentHolder中的Map成员dataProcessorMap，以TYPE为key的对应的value

private final Map<String, DistroDataProcessor> dataProcessorMap = new HashMap<>();
​
public DistroDataProcessor findDataProcessor(String processType) {
    return dataProcessorMap.get(processType);
}

到这里我们知道了存储在哪里以及对应的key，但是对应的value该如何查找呢？大家可以继续看到DistroComponentHolder中，可以发现这个Map是private私有的，那么就说明外部不能随便进行赋值，有且只能通过registerDataProcessor方法进行赋值

那么我们查找一下这个方法使用的地方，看看是否能查找到部分线索，可以发现就两个地方

我们可以分别看一下这两个地方中，因为我们需要其processType为TYPE，那么也就是对应的processType()方法返回值为TYPE，可以看到，符合条件的是DistroClientComponentRegistry,另外一个的返回不满足条件(另外一个的就不贴图了，大家自己可以去看一下,他的Key是另外一个常量)

那么确定了对应的返回值类型是DistroClientComponentRegistry,那么我们就可以看其中的processData()方法的实现逻辑

@Override
public boolean processData(DistroData distroData) {
    switch (distroData.getType()) {
        case ADD:
        case CHANGE:
            // 添加和改变走这个逻辑
            // 这一步是将请求数据利用序列化器进行对应的反序列化的操作
            ClientSyncData clientSyncData = ApplicationUtils.getBean(Serializer.class)
                    .deserialize(distroData.getContent(), ClientSyncData.class);
            handlerClientSyncData(clientSyncData);
            return true;
        case DELETE:
            // 删除走这个逻辑
            String deleteClientId = distroData.getDistroKey().getResourceKey();
            Loggers.DISTRO.info("[Client-Delete] Received distro client sync data {}", deleteClientId);
            clientManager.clientDisconnected(deleteClientId);
            return true;
        default:
            return false;
    }
}

可以看到主要的逻辑就是handlerClientSyncData方法，而在这个方法内的主要逻辑是upgradeClient方法

private void handlerClientSyncData(ClientSyncData clientSyncData) {
    Loggers.DISTRO.info("[Client-Add] Received distro client sync data {}", clientSyncData.getClientId());
    clientManager.syncClientConnected(clientSyncData.getClientId(), clientSyncData.getAttributes());
    Client client = clientManager.getClient(clientSyncData.getClientId());
    upgradeClient(client, clientSyncData);
}
​
// 传入client即客户端ID对应的客户端对象 和 clientSyncData为对应的客户端数据
private void upgradeClient(Client client, ClientSyncData clientSyncData) {
    List<String> namespaces = clientSyncData.getNamespaces();
    List<String> groupNames = clientSyncData.getGroupNames();
    List<String> serviceNames = clientSyncData.getServiceNames();
    List<InstancePublishInfo> instances = clientSyncData.getInstancePublishInfos();
    // 同步数据集合
    Set<Service> syncedService = new HashSet<>();
    // 遍历传入参数中的所有服务信息
    for (int i = 0; i < namespaces.size(); i++) {
        // 创建对应的Service对象
        Service service = Service.newService(namespaces.get(i), groupNames.get(i), serviceNames.get(i));
        // 获取内存中保存的对应的service单例对象
        Service singleton = ServiceManager.getInstance().getSingleton(service);
        // 加入到集合中，方便后续比较
        syncedService.add(singleton);
        InstancePublishInfo instancePublishInfo = instances.get(i);
        // 如果从client中获取的和同步数据中的内容不一样，则发布新增事件，client需要同步新数据
        if (!instancePublishInfo.equals(client.getInstancePublishInfo(singleton))) {
            client.addServiceInstance(singleton, instancePublishInfo);
            NotifyCenter.publishEvent(
                    new ClientOperationEvent.ClientRegisterServiceEvent(singleton, client.getClientId()));
        }
    }
    // 如果某个服务不在同步数据中（即不在 syncedService 集合中），则从client中移除该服务实例并发布注销事件
    for (Service each : client.getAllPublishedService()) {
        if (!syncedService.contains(each)) {
            client.removeServiceInstance(each);
            NotifyCenter.publishEvent(
                    new ClientOperationEvent.ClientDeregisterServiceEvent(each, client.getClientId()));
        }
    }
}
​

在这里会解析传过来的参数，最后对于新增的数据，还是发布了 ClientRegisterServiceEvent 事件，这个事件我们在讲解注册源码的时候已经分析过了，这一块注册逻辑就结束了。

我们会过来看看processData中对于删除操作的逻辑处理，可以发现核心在于clientManager.clientDisconnected(deleteClientId);一样得需要找对应的实现类

这个实现类怎么找到的呢，这个有点难度。
从代码的角度分析的话，DistroClientComponentRegistry中的这个clientManager，本质上是来源于DistroClientComponentRegistry中的clientManager，而DistroClientComponentRegistry是个Bean对象
那么我们来看看clientManager，他对应的接口类型是ClientManagerDelegate，这个ClientManagerDelegate中也同样实现了clientDisconnected这个方法
ClientManagerDelegate 持有三种不同的 ClientManager 实现：
ConnectionBasedClientManager：基于连接的客户端管理器
EphemeralIpPortClientManager：临时IP端口客户端管理器
PersistentIpPortClientManager：持久IP端口客户端管理器
​
而ClientManagerDelegate中对于clientDisconnected的实现如下
@Override
public boolean clientDisconnected(String clientId) {
    return getClientManagerById(clientId).clientDisconnected(clientId);
}
与这个方法相关联的方法如下
private ClientManager getClientManagerById(String clientId) {
    if (isConnectionBasedClient(clientId)) {
        return connectionBasedClientManager;
    }
    return clientId.endsWith(ClientConstants.PERSISTENT_SUFFIX) ? persistentIpPortClientManager : ephemeralIpPortClientManager;
}
​
private boolean isConnectionBasedClient(String clientId) {
   return !clientId.contains(IpPortBasedClient.ID_DELIMITER);
}
​
也就是说，我们确定这个clientDisconnected的实现类，最终是要依靠clientId，而本质上这个clientId是客户端注册的时候生成的，比较难追踪，而且你会发现很多的相关的信息都会回到这个clientId上来
​
所以我们这里其实最好的方法是debug的形式，或者我们也可以从理解的角度来看，一般的客户端都是临时IP端口，所以默认为EphemeralIpPortClientManager的实现

这个方法我们看 EphemeralIpPortClientManager 临时实例的实现类

@Override
public boolean clientDisconnected(String clientId) {
    Loggers.SRV_LOG.info("Client connection {} disconnect, remove instances and subscribers", clientId);
    // 移除客户端信息
    IpPortBasedClient client = clients.remove(clientId);
    if (null == client) {
        return true;
    }
    // 发布客户端注销事件
    NotifyCenter.publishEvent(new ClientEvent.ClientDisconnectEvent(client));
    client.release();
    return true;
}
​

ClientDisconnectEvent 发布客户端注销事件，会在多个地方处理该事件。

第一个是DistroClientDataProcessor，客户端注销后会同步集群节点。第二个是ClientServiceIndexesManager，移除本身节点该 client 的数据。

客户端注销同步集群节点没什么好说的了，和注册同步逻辑类似，我们来看下 ClientServiceIndexesManager 类处理注销事件，代码如下：

​
@Override
public void onEvent(Event event) {
    if (event instanceof ClientEvent.ClientDisconnectEvent) {
        // 处理客户端注销
        handleClientDisconnect((ClientEvent.ClientDisconnectEvent) event);
    } else if (event instanceof ClientOperationEvent) {
        handleClientOperation((ClientOperationEvent) event);
    }
}
​
private void handleClientDisconnect(ClientEvent.ClientDisconnectEvent event) {
    Client client = event.getClient();
    for (Service each : client.getAllSubscribeService()) {
        // 移除订阅者信息 其实就是操作subscriberIndexes
        removeSubscriberIndexes(each, client.getClientId());
    }
    for (Service each : client.getAllPublishedService()) {
        // 移除注册表信息 其实就是操作publisherIndexes
        removePublisherIndexes(each, client.getClientId());
    }
}

我们在Nacos源码学习计划-Day20-Nacos2.x-服务端处理客户端gRPC注册请求 的最后面有看到，服务端处理gRPC的注册请求的最后，即addPublisherIndexes()的最后一行，可以看到是发布了一个ServiceChangedEvent服务变更的事件

private void addPublisherIndexes(Service service, String clientId) {
    // 把 clientId 写入注册表
    publisherIndexes.computeIfAbsent(service, (key) -> new ConcurrentHashSet<>());
    publisherIndexes.get(service).add(clientId);
    // 发布服务改变事件
    NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true));
}

今天就围绕这个来看看，服务变更后是如何通知对应的订阅的客户端的

初看ServiceChangedEvent

老规矩，对于Event事件处理，优先需要找到对应的handler处理器，还是利用IDEA的全局搜索能轻松找到

看到这个地方，大家可以发现，这个if部分的逻辑其实就是将对应的ServiceChangedEvent封装成为PushDelayTask类型，然后调用delayTaskEngine.addTask()方法，将其放入到任务队列中去执行

如果有阅读了上一节内容的UU就会发现，在这里Nacos使用这个的 delayTaskEngine.addTask(),在上一节的最后，也就是ClientSubscribeServiceEvent客户端订阅服务事件的处理中也是利用这个，这个其实是Nacos2.X中很重要的延迟任务处理引擎

所以接下来，我们来好好分析一下这个玩意儿

PushDelayTaskExecuteEngine

延迟任务执行引擎，顾名思义是执行延迟任务，可以往执行引擎中添加任务，然后该任务会被延时执行。

这个delayTaskEngine的数据类型是PushDelayTaskExecuteEngine,我们可以先来查看一下他的相关的继承和实现关系

既然我们知道了最终会将任务放入到一个任务队列中，那么我们尝试找找从队列中取任务的逻辑

但是实际搜索会发现，好像并没有这种一个一个拿的相关逻辑

这个时候我们再回来看看整个PushDelayTaskExecuteEngine的代码，通过方法命名和整个组织的结构来看，可以猜测到，整个逻辑任务的实现可能依靠的是processTasks()这个方法，但是整个链路逻辑还是不是很明了，从代码来看，我们需要网上再去看其父类是如何实现的

我们来看其父类NacosDelayTaskExecuteEngine的逻辑

可以从父类的构造方法中看到，构建的时候会利用内部的线程池成员开启ProcessRunnable 线程任务，而这个线程任务的底层逻辑就是执行processTasks()方法

这个方法在子类中是有重写的，但是我们去看子类中的实现，就是加了一定的判断后再调用父类中的processTasks方法，所以我们直接来看父类中的实现即可

可以看到，首先从tasks中获取到所有的任务key进行遍历，通过 taskKey 获取到具体的任务之后，再通过 taskKey 获取对应的处理器，最终调用处理器(通过getProcessor()方法获取)的process()处理方法

从getProcessor()的逻辑可以看到，如果给对应的key即特定的task设置了特定的processor处理器，则会保存在taskProcessors这个Map中，如果没有则使用默认的处理器，在顶级抽象接口中为NacosTaskProcessor，而PushDelayTaskExecuteEngine的构造方法中调用了set方法将这个处理器设置为了PushDelayTaskProcessor

​
// 任务池
protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks;
​
// 这个不是NacosDelayTaskExecuteEngine的，而是来自顶级抽象类AbstractNacosTaskExecuteEngine  存储了特定的task-processor处理器的映射关系
private final ConcurrentHashMap<Object, NacosTaskProcessor> taskProcessors = new ConcurrentHashMap<>();
​
/**
 * process tasks in execute engine.
 */
protected void processTasks() {
    // 获取全部的任务，进行遍历
    Collection<Object> keys = getAllTaskKeys();
    for (Object taskKey : keys) {
        
        // 通过任务 key，获取具体的任务，并且从任务池中移除掉
        AbstractDelayTask task = removeTask(taskKey);
        if (null == task) {
            continue;
        }
        // 通过任务 key 获取对应的处理器
        NacosTaskProcessor processor = getProcessor(taskKey);
        if (null == processor) {
            getEngineLog().error("processor not found for task, so discarded. " + task);
            continue;
        }
        try {
            // 调用处理器的处理方法
            if (!processor.process(task)) {
                retryFailedTask(taskKey, task);
            }
        } catch (Throwable e) {
            getEngineLog().error("Nacos task execute error ", e);
            retryFailedTask(taskKey, task);
        }
    }
}
​
​
@Override
public NacosTaskProcessor getProcessor(Object key) {
  // 如果处理器中没有对应的 key，那么就返回默认的任务处理器
  return taskProcessors.containsKey(key) ? taskProcessors.get(key) : defaultTaskProcessor;
}

而我们整个任务的存储结构tasks这个map中的元素，就是来自于我们上面类似ServiceChangedEvent的各种事件处理代码中，会通过addTask()方法放入到tasks中从而完成任务的放入

@Override
public void addTask(Object key, AbstractDelayTask newTask) {
    lock.lock();
    try {
        // 往任务池添加任务
        AbstractDelayTask existTask = tasks.get(key);
        if (null != existTask) {
            newTask.merge(existTask);
        }
        tasks.put(key, newTask);
    } finally {
        lock.unlock();
    }
}