< div style = "color:#16b0ff;font-size:50px;font-weight: 900;text-shadow: 5px 5px 10px var(--theme-color);font-family: 'Comic Sans MS';" > Middleware< / div >
< span style = "color:#16b0ff;font-size:20px;font-weight: 900;font-family: 'Comic Sans MS';" > Introduction</ span > ：收纳技术相关的基础知识 `Redis` 、`RocketMQ`、`Zookeeper`、`Netty`、`Tomcat` 等总结！
[TOC]
# SPI
SPI 全称为 Service Provider Interface，
## Java SPI
JDK 中 提供了一个 `SPI` 的功能，核心类是 `java.util.ServiceLoader` 。其作用就是，可以通过类名获取在 `META-INF/services/` 下的多个配置实现文件。为了解决上面的扩展问题，现在我们在`META-INF/services/`下创建一个`com.github.yu120.test.SuperLoggerConfiguration`文件（没有后缀）。文件中只有一行代码，那就是我们默认的`com.github.yu120.test.XMLConfiguration`（注意，一个文件里也可以写多个实现，回车分隔）。然后通过 ServiceLoader 获取我们的 SPI 机制配置的实现类：
```java
// META-INF/services/com.github.test.test.SuperLoggerConfiguration:
com.github.yu120.test.XMLConfiguration
ServiceLoader< SuperLoggerConfiguration > serviceLoader = ServiceLoader.load(SuperLoggerConfiguration.class);
Iterator< SuperLoggerConfiguration > iterator = serviceLoader.iterator();
SuperLoggerConfiguration configuration;
while(iterator.hasNext()) {
// 加载并初始化实现类
configuration = iterator.next();
}
// 对最后一个configuration类调用configure方法
configuration.configure(configFile);
```
## [Dubbo SPI](https://github.com/apache/dubbo/tree/master/dubbo-common/src/main/java/org/apache/dubbo/common/extension)
Dubbo SPI 的相关逻辑被封装在了 ExtensionLoader类中， ， ， ：
- 通过配置文件，解耦拓展接口和拓展实现类
- 通过IOC自动注入依赖的拓展实现类对象
- 通过URL参数，
Dubbo SPI 所需的配置文件需放置在 META-INF/dubbo 路径下， （ ）
```properties
optimusPrime = org.apache.spi.OptimusPrime
bumblebee = org.apache.spi.Bumblebee
```
与 Java SPI 实现类配置不同， @SPI 注解。
## [Motan SPI](https://github.com/weibocom/motan/tree/master/motan-core/src/main/java/com/weibo/api/motan/core/extension)
Motan使用SPI机制来实现模块间的访问， ，
Motan的SPI的实现在 `motan-core/com/weibo/api/motan/core/extension` 中。组织结构如下：
```tex
motan-core/com.weibo.api.motan.core.extension
|-Activation： ，
|-ActivationComparator：
|-ExtensionLoader： ，
|-Scope： ，
|-Spi： ， ，
|-SpiMeta： ， ，
```
## SpringBoot SPI
Spring 的 SPI 配置文件是一个固定的文件 - `META-INF/spring.factories` ，功能上和 JDK 的类似，每个接口可以有多个扩展实现，使用起来非常简单：
```java
// 获取所有factories文件中配置的LoggingSystemFactory
List< LoggingSystemFactory > > factories = SpringFactoriesLoader.loadFactories(LoggingSystemFactory.class, classLoader);
```
下面是一段 Spring Boot 中 spring.factories 的配置
```properties
# Logging Systems
org.springframework.boot.logging.LoggingSystemFactory=\
org.springframework.boot.logging.logback.LogbackLoggingSystem.Factory,\
org.springframework.boot.logging.log4j2.Log4J2LoggingSystem.Factory
# PropertySource Loaders
org.springframework.boot.env.PropertySourceLoader=\
org.springframework.boot.env.PropertiesPropertySourceLoader,\
org.springframework.boot.env.YamlPropertySourceLoader
```
# Redis
**Java类型所占字节数（ ）
| 类型 | 存储(byte) | bit数(bit) | 取值范围 |
| ------- | ---------- | ---------- | ------------------------------------------------------------ |
| int | 4字节 | 4×
| short | 2字节 | 2×
| long | 8字节 | 8×
| byte | 1字节 | 1× ，
| float | 4字节 | 4× （ ： ）
| double | 8字节 | 8× （ ：
| boolean | 1字节 | 1×
| char | 2字节 | 2× ， ， ，
**注意**：
- 英文的数字、字母或符号：
- 中文的数字、字母或符号：
- 计算机的基本单位： ，
- 1TB=1024GB， ， ， （ ， ） ， （ ， ）
## 线程模型
Redis内部使用文件事件处理器`File Event Handler`， ：
- **多个Socket**
- **I/O多路复用程序**
- **Scocket队列**
- **文件事件分派器**
- **事件处理器**（连接应答处理器、命令请求处理器、命令回复处理器）
### 通信流程
客户端与redis的一次通信过程：
- **请求类型1**：
- 服务端会产生一个`AE_READABLE`事件，`I/O`多路复用程序接收到`server socket`事件后，将该`socket`压入队列中
- 文件事件分派器从队列中获取`socket`，交给连接应答处理器，创建一个可以和客户端交流的`socket01`
- 将`socket01`的`AE_READABLE`事件与命令请求处理器关联
- **请求类型2**：
- `socket01` 产生`AE_READABLE`事件，`socket01`压入队列
- 将获取到的`socket01`与命令请求处理器关联
- 命令请求处理器读取`socket01`中的`key value`，并在内存中完成对应的设置
- 将`socket01`的`AE_WRITABLE`事件与命令回复处理器关联
- **请求类型3**：
- `Redis` 中的`socket01`会产生一个`AE_WRITABLE`事件，压入到队列中
- 将获取到的`socket01`与命令回复处理器关联
- 回复处理器对`socket01`输入操作结果，如`ok`。之后解除`socket01`的`AE_WRITABLE`事件与命令回复处理器的关联
### 文件事件处理器
- **基于 Reactor 模式开发了自己的网络事件处理器（ ， ）
- 文件事件处理器 ** 使用 I/O 多路复用（ ）
- 当被监听的套接字准备好执行连接应答（ ） （ ） （ ） （ ） ，
- 文件事件处理器以单线程方式运行， 但通过使用 I/O 多路复用程序来监听多个套接字， 文件事件处理器既实现了高性能的网络通信模型， 又可以很好地与 redis 服务器中其他同样以单线程方式运行的模块进行对接， 这保持了 Redis 内部单线程设计的简单性
### I/O多路复用
I/O多路复用的I/O是指网络I/O， ） ， ，
I/O多路复用使用两个系统调用(select/poll/epoll和recvfrom)， ； ， ， ， ， ， ，
**select机制**
**基本原理**
客户端操作服务器时就会产生这三种文件描述符(简称fd)： ， ； ，
**优点**
- 几乎在所有的平台上支持，跨平台支持性好
**缺点**
- 由于是采用轮询方式全盘扫描，
- 每次调用 select()，需要把 fd 集合从用户态拷贝到内核态，并进行遍历(消息传递都是从内核到用户空间)
- 默认单个进程打开的FD有限制是1024个， ，
**poll机制**
基本原理与select一致， ； （ ）
**epoll机制**
**基本原理**
没有fd个数限制， ， ， ， ：
- `epoll_create()` ： ， ， ，
- `epoll_ctl()` ： ， ，
- `epoll_wait()` ： ，
**优点**
- 没fd这个限制， ，
- 效率提高， ，
- 内核和用户空间mmap同一块内存实现(mmap是一种内存映射文件方法，
例子： ， ，
- `select` ： ， ，
- `poll` ： ， ， ，
- `epoll` : 请求进来时就创建fd并绑定一个callback， ，
### 执行效率高
**Redis是单线程模型为什么效率还这么高？
- `纯内存操作` ： ， ，
- `非阻塞的I/O多路复用机制` ： ， ， ， ，
- `C语言实现` ：距离操作系统更近，执行速度会更快
- `单线程避免切换开销` ：单线程避免了多线程上下文切换的时间开销，预防了多线程可能产生的竞争问题
## 数据类型
在 Redis 中，常用的 5 种数据类型和应用场景如下：
- **String：
- **List：
- **Hash：
- **Set：
- **Zset：
### String(字符串)
String 数据结构是简单的 key-value 类型， ， （
### Hash(字典)
在 Memcached 中，我们经常将一些结构化的信息打包成 hashmap， （ ） ， ， （ ） ， ， ， （ ） ， ， （ ）
- 存储、读取、修改用户属性
**实战场景**
- 缓存： 经典使用场景， ， ， ， ， ，
- 计数器： ， ，
- session：
### List(列表)
List 说白了就是链表（ ） ， ） ， ，
- 微博 TimeLine
- 消息队列
使用列表的技巧
- lpush+lpop=Stack(栈)
- lpush+rpop=Queue（ ）
- lpush+ltrim=Capped Collection（ ）
- lpush+brpop=Message Queue（ ）
实战场景：
- timeline： ， ， ，
### Set(集合)
Set 就是一个集合，集合的概念就是一堆不重复值的组合。利用 Redis 提供的 Set 数据结构，可以存储一些集合性的数据。比如在微博应用中，可以将一个用户所有的关注人存在一个集合中，将其所有粉丝存在一个集合。因为 Redis 非常人性化的为集合提供了求交集、并集、差集等操作，那么就可以非常方便的实现如共同关注、共同喜好、二度好友等功能，对上面的所有集合操作，你还可以使用不同的命令选择将结果返回给客户端还是存集到一个新的集合中。
- 共同好友、二度好友
- 利用唯一性，可以统计访问网站的所有独立 IP
- 好友推荐的时候，根据 tag 求交集，大于某个 threshold 就可以推荐
实战场景;
- 标签（ ）
- 点赞， ， ，
### Sorted Set(有序集合)
和Sets相比， ， ， ， ，
- 带有权重的元素，比如一个游戏的用户得分排行榜
- 比较复杂的数据结构，一般用到的场景不算太多
实战场景：
- 排行榜：有序集合经典使用场景。例如小说视频等网站需要对用户上传的小说视频做排行榜，榜单可以按照用户关注数，更新时间，字数等打分，做排行。
## 特殊数据结构
### HyperLogLog(基数统计)
HyperLogLog 主要的应用场景就是进行基数统计。实际上不会存储每个元素的值， ， ，
| 命令 | 作用 |
| ----------------------------- | ----------------------- |
| pfadd key element ... | 将所有元素添加到key中 |
| pfcount key | 统计key的估算值(不精确) |
| pgmerge new_key key1 key2 ... | 合并key至新key |
**应用案例**
如何统计 Google 主页面每天被多少个不同的账户访问过？
对于 Google 这种访问量巨大的网页而言，其实统计出有十亿的访问量或十亿零十万的访问量其实是没有太多的区别的，因此，在这种业务场景下，为了节省成本，其实可以只计算出一个大概的值，而没有必要计算出精准的值。
对于上面的场景，可以使用`HashMap`、`BitMap`和`HyperLogLog`来解决。对于这三种解决方案，这边做下对比：
- `HashMap` ：算法简单，统计精度高，对于少量数据建议使用，但是对于大量的数据会占用很大内存空间
- `BitMap` ：位图算法，具体内容可以参考我的这篇，统计精度高，虽然内存占用要比`HashMap`少，但是对于大量数据还是会占用较大内存
- `HyperLogLog` ：存在一定误差，占用内存少，稳定占用 12k 左右内存，可以统计 2^64 个元素，对于上面举例的应用场景，建议使用
### Geo(地理空间信息)
Geo主要用于存储地理位置信息， （ ） ， ：
| 命令 | 作用 |
| ------------------------------------ | ------------------------------------------------------------ |
| geoadd key latitude longitude member | 添加成员位置(纬度、经度、名称)到key中 |
| geopos key member ... | 获取成员geo坐标 |
| geodist key member1 member2 [unit] | 计算成员位置间距离。若两个位置之间的其中一个不存在， 那返回空值 |
| georadius | 基于经纬度坐标范围查询 |
| georadiusbymember | 基于成员位置范围查询 |
| geohash | 计算经纬度hash |
**GEORADIUS**
```properties
GEORADIUS key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count]
```
以给定的经纬度为中心， 返回键包含的位置元素当中， 与中心的距离不超过给定最大距离的所有位置元素。范围可以使用以下其中一个单位：
- **m** 表示单位为米
- **km** 表示单位为千米
- **mi** 表示单位为英里
- **ft** 表示单位为英尺
在给定以下可选项时， 命令会返回额外的信息：
- `WITHDIST` : 在返回位置元素的同时， 将位置元素与中心之间的距离也一并返回。 距离单位和范围单位保持一致
- `WITHCOORD` : 将位置元素的经度和维度也一并返回
- `WITHHASH` : 以 52 位有符号整数的形式， 返回位置元素经过原始 geohash 编码的有序集合分值。 这个选项主要用于底层应用或者调试， 实际中的作用并不大
命令默认返回未排序的位置元素。 通过以下两个参数， 用户可以指定被返回位置元素的排序方式：
- `ASC` : 根据中心的位置， 按照从近到远的方式返回位置元素
- `DESC` : 根据中心的位置， 按照从远到近的方式返回位置元素
在默认情况下， GEORADIUS 命令会返回所有匹配的位置元素。 虽然用户可以使用 **COUNT `<count>`** 选项去获取前 N 个匹配元素， 但是因为命令在内部可能会需要对所有被匹配的元素进行处理， 所以在对一个非常大的区域进行搜索时， 即使只使用 `COUNT` 选项去获取少量元素， 命令的执行速度也可能会非常慢。 但是从另一方面来说， 使用 `COUNT` 选项去减少需要返回的元素数量， 对于减少带宽来说仍然是非常有用的。
### Pub/Sub(发布订阅)
发布订阅类似于广播功能。redis发布订阅包括 发布者、订阅者、Channel。常用命令如下：
| 命令 | 作用 | 时间复杂度 |
| ----------------------- | ---------------------------------- | ------------------------------------------------------------ |
| subscribe channel | 订阅一个频道 | O(n) |
| unsubscribe channel ... | 退订一个/多个频道 | O(n) |
| publish channel msg | 将信息发送到指定的频道 | O(n+m)，
| pubsub CHANNELS | 查看订阅与发布系统状态(多种子模式) | O(n) |
| psubscribe | 订阅多个频道 | O(n) |
| unsubscribe | 退订多个频道 | O(n) |
### Bitmap(位图)
Bitmap就是位图， （ ） ， （ ） ， ， ， ：
| 命令 | 作用 | 时间复杂度 |
| --------------------------- | --------------------------------------------------- | ---------- |
| setbit key offset val | 给指定key的值的第offset赋值val | O(1) |
| getbit key offset | 获取指定key的第offset位 | O(1) |
| bitcount key start end | 返回指定key中[start,end]中为1的数量 | O(n) |
| bitop operation destkey key | 对不同的二进制存储数据进行位运算(AND、OR、NOT、XOR) | O(n) |
**应用案例**
有1亿用户， ， ， ：
| 数据类型 | 每个 userid 占用空间 | 需要存储的用户量 | 全部占用内存量 |
| -------- | ------------------------------------------------------------ | ---------------- | ---------------------------- |
| set | 32位也就是4个字节（ ， ）
| Bitmap | 1 位（ ）
**应用场景**
- 用户在线状态
- 用户签到状态
- 统计独立用户
### BloomFilter(布隆过滤)
当一个元素被加入集合时， （ ， ， ， ） ， ， （ ） ：
- 如果这些点有任何一个为0，
- 如果都是1， ， ，
**应用场景**
- **解决缓存穿透**：
- **黑名单校验**：假设黑名单的数量是数以亿计的，存放起来就是非常耗费存储空间的，布隆过滤器则是一个较好的解决方案。把所有黑名单都放在布隆过滤器中，再收到邮件时，判断邮件地址是否在布隆过滤器中即可
- **Web拦截器**：用户第一次请求，将请求参数放入布隆过滤器中，当第二次请求时，先判断请求参数是否被布隆过滤器命中，从而提高缓存命中率
#### 基于Bitmap数据结构
```java
import com.google.common.base.Preconditions;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@Service
public class RedisService {
@Resource
private RedisTemplate< String , Object > redisTemplate;
/**
* 根据给定的布隆过滤器添加值
*/
public < T > void addByBloomFilter(BloomFilterHelper< T > bloomFilterHelper, String key, T value) {
Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");
int[] offset = bloomFilterHelper.murmurHashOffset(value);
for (int i : offset) {
redisTemplate.opsForValue().setBit(key, i, true);
}
}
/**
* 根据给定的布隆过滤器判断值是否存在
*/
public < T > boolean includeByBloomFilter(BloomFilterHelper< T > bloomFilterHelper, String key, T value) {
Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");
int[] offset = bloomFilterHelper.murmurHashOffset(value);
for (int i : offset) {
if (!redisTemplate.opsForValue().getBit(key, i)) {
return false;
}
}
return true;
}
}
```
#### 基于RedisBloom模块
RedisBloom模块提供了四种数据类型：
- **Bloom Filter (布隆过滤器)**
- **Cuckoo Filter（ ）
- **Count-Mins-Sketch**
- **Top-K**
`Bloom Filter` 和 `Cuckoo` 用于确定（以给定的确定性）集合中是否存在某项。使用 `Count-Min Sketch` 来估算子线性空间中的项目数，使用 `Top-K` 维护K个最频繁项目的列表。
```shell
# 1.git 下载
[root@test ~]# git clone https://github.com/RedisBloom/RedisBloom.git
[root@test ~]# cd redisbloom
[root@test ~]# make
# 2.wget 下载
[root@test ~]# wget https://github.com/RedisBloom/RedisBloom/archive/v2.0.3.tar.gz
[root@test ~]# tar -zxvf RedisBloom-2.0.3.tar.gz
[root@test ~]# cd RedisBloom-2.0.3/
[root@test ~]# make
# 3.修改Redis Conf
[root@test ~]#vim /etc/redis.conf
# 在文件中添加下行
loadmodule /root/RedisBloom-2.0.3/redisbloom.so
# 4.启动Redis server
[root@test ~]# /redis-server /etc/redis.conf
# 或者启动服务时加载os文件
[root@test ~]# /redis-server /etc/redis.conf --loadmodule /root/RedisBloom/redisbloom.so
# 5.测试RedisBloom
[root@test ~]# redis-cli
127.0.0.1:6379> bf.add bloomFilter foo
127.0.0.1:6379> bf.exists bloomFilter foo
127.0.0.1:6379> cf.add cuckooFilter foo
127.0.0.1:6379> cf.exists cuckooFilter foo
```
## 底层数据结构
当然是为了追求速度，不同数据类型使用不同的数据结构速度才得以提升。每种数据类型都有一种或者多种数据结构来支撑，底层数据结构有 6 种。
### SDS(简单动态字符)
C 语言字符串结构与 SDS 字符串结构对比图如上所示：
- SDS 中 len 保存这字符串的长度，
- 空间预分配：
- 惰性空间释放：当对 SDS 进行缩短操作时，程序并不会回收多余的内存空间，而是使用 free 字段将这些字节数量记录下来不释放，后面如果需要 append 操作，则直接使用 free 中未使用的空间，减少了内存的分配
### hash表(字典)
Redis 整体就是一个 哈希表来保存所有的键值对，无论数据类型是 5 种的任意一种。哈希表，本质就是一个数组，每个元素被叫做哈希桶，不管什么数据类型，每个桶里面的 entry 保存着实际具体值的指针。
整个数据库就是一个全局哈希表，而哈希表的时间复杂度是 O(1)，只需要计算每个键的哈希值，便知道对应的哈希桶位置，定位桶里面的 entry 找到对应数据，这个也是 Redis 快的原因之一。
**那 Hash 冲突怎么办？**
当写入 Redis 的数据越来越多的时候，哈希冲突不可避免，会出现不同的 key 计算出一样的哈希值。Redis 通过链式哈希解决冲突：也就是同一个 桶里面的元素使用链表保存。但是当链表过长就会导致查找性能变差可能，所以 Redis 为了追求快，使用了两个全局哈希表。用于 rehash 操作，增加现有的哈希桶数量，减少哈希冲突。开始默认使用 hash 表 1 保存键值对数据，哈希表 2 此刻没有分配空间。
### linkedList(双端列表)
后续版本对列表数据结构进行了改造，使用 quicklist 代替了 ziplist 和 linkedlist。quicklist 是 ziplist 和 linkedlist 的混合体，它将 linkedlist 按段切分，每一段使用 ziplist 来紧凑存储，多个 ziplist 之间使用双向指针串接起来。这也是为何 Redis 快的原因，不放过任何一个可以提升性能的细节。
### zipList(压缩列表)
压缩列表是 List 、hash、 sorted Set 三种数据类型底层实现之一。当一个列表只有少量数据的时候，并且每个列表项要么就是小整数值，要么就是长度比较短的字符串，那么 Redis 就会使用压缩列表来做列表键的底层实现。
ziplist 是由一系列特殊编码的连续内存块组成的顺序型的数据结构， ， ， ，
```go
struct ziplist< T > {
int32 zlbytes; // 整个压缩列表占用字节数
int32 zltail_offset; // 最后一个元素距离压缩列表起始位置的偏移量，用于快速定位到最后一个节点
int16 zllength; // 元素个数
T[] entries; // 元素内容列表，挨个挨个紧凑存储
int8 zlend; // 标志压缩列表的结束，值恒为 0xFF
}
```
如果我们要查找定位第一个元素和最后一个元素，可以通过表头三个字段的长度直接定位，复杂度是 O(1)。而查找其他元素时，就没有这么高效了，只能逐个查找，此时的复杂度就是 O(N)。
### skipList(跳跃表)
sorted set 类型的排序功能便是通过「跳跃列表」数据结构来实现。跳跃表（ ） ， ，
跳跃表支持平均 O（ ） （ ） ， ， ， ，
### intset(整数数组)
当一个集合只包含整数值元素， ，
```go
typedef struct intset{
//编码方式
uint32_t encoding;
//集合包含的元素数量
uint32_t length;
//保存元素的数组
int8_t contents[];
}intset;
```
contents 数组是整数集合的底层实现：整数集合的每个元素都是 contents 数组的一个数组项（ ） ， ，
## 持久化机制
通常来说，应该同时使用两种持久化方案，以保证数据安全：
- 如果数据不敏感，且可以从其他地方重新生成，可以关闭持久化
- 如果数据比较重要， ， ，
- 如果是用做内存数据， ，
- 如果只用AOF， ，
当RDB与AOF两种方式都开启时， ，
### RDB模式(内存快照)
`RDB` （ ，
**优点**
- RDB快照是一个压缩过的非常紧凑的文件。保存着某个时间点的数据集， ，
- 可最大化Redis的的性能。在保存RDB文件， ，
- 与AOF相比，
**缺点**
- RDB的数据安全性是不如AOF的， ， ， ，
- Redis数据集较大时，
**① 创建**
当 Redis 持久化时，程序会将当前内存中的**数据库状态**保存到磁盘中。创建 RDB 文件主要有两个 Redis 命令：`SAVE` 和 `BGSAVE` 。
**② 载入**
服务器在载入 RDB 文件期间，会一直处于**阻塞**状态，直到载入工作完成为止。
#### save同步保存
`save` 命令是同步操作，执行命令时，会 ** 阻塞** Redis 服务器进程，拒绝客户端发送的命令请求。
具体流程如下：
由于 `save` 命令是同步命令， ， ， ， ，
#### bgsave异步保存
`bgsave` 命令是异步操作，执行命令时，子进程执行保存工作，服务器还可以继续让主线程**处理**客户端发送的命令请求。
具体流程如下：
Redis使用Linux系统的`fock()`生成一个子进程来将DB数据保存到磁盘， ，
#### 自动保存
可通过配置文件对 Redis 进行设置， 让它在“ N 秒内数据集至少有 M 个改动”这一条件被满足时， 自动进行数据集保存操作:
```shell
# RDB自动持久化规则
# 当 900 秒内有至少有 1 个键被改动时，自动进行数据集保存操作
save 900 1
# 当 300 秒内有至少有 10 个键被改动时，自动进行数据集保存操作
save 300 10
# 当 60 秒内有至少有 10000 个键被改动时，自动进行数据集保存操作
save 60 10000
# RDB持久化文件名
dbfilename dump-< port > .rdb
# 数据持久化文件存储目录
dir /var/lib/redis
# bgsave发生错误时是否停止写入，
stop-writes-on-bgsave-error yes
# rdb文件是否使用压缩格式
rdbcompression yes
# 是否对rdb文件进行校验和检验，
rdbchecksum yes
```
#### 默认配置
RDB 文件默认的配置如下：
```properties
################################ SNAPSHOTTING ################################
#
# Save the DB on disk:
# 在给定的秒数和给定的对数据库的写操作数下，自动持久化操作。
# save <seconds> <changes>
#
save 900 1
save 300 10
save 60 10000
#bgsave发生错误时是否停止写入，
stop-writes-on-bgsave-error yes
#持久化时是否使用LZF压缩字符串对象?
rdbcompression yes
#是否对rdb文件进行校验和检验，
rdbchecksum yes
# RDB持久化文件名
dbfilename dump.rdb
#持久化文件存储目录
dir ./
```
### AOF模式(日志追加)
AOF（ ， ，
**优点**
- 数据更完整， ， （ ， ， ）
- AOF文件是一个只进行追加的日志文件， ， ，
**缺点**
- 对于相同的数据集， ，
- 根据所使用的fsync策略， ，
#### 持久化流程
**① 命令追加**
若 AOF 持久化功能处于打开状态，服务器在执行完一个命令后，会以协议格式将被执行的写命令**追加**到服务器状态的 `aof_buf` 缓冲区的末尾。
**② 文件同步**
服务器每次结束一个事件循环之前，都会调用 `flushAppendOnlyFile` 函数，这个函数会考虑是否需要将 `aof_buf` 缓冲区中的内容**写入和保存**到 AOF 文件里。`flushAppendOnlyFile` 函数执行以下流程：
- WRITE： ，
- SAVE： ，
这个函数是由服务器配置的 `appendfsync` 的三个值：`always、everysec、no` 来影响的，也被称为三种策略：
- **always**：每条**命令**都会 fsync 到硬盘中，这样 redis 的写入数据就不会丢失。
- **everysec**：
- **no**：根据当前**操作系统**的规则决定什么时候刷新到硬盘中，不需要我们来考虑。
**数据加载**
- 创建一个不带网络连接的伪客户端
- 从 AOF 文件中分析并读取出一条写命令
- 使用伪客户端执行被读出的写命令
- 一直执行步骤 2 和 3，
#### 文件重写
**为何需要文件重写**
- 为了解决 AOF 文件**体积膨胀**的问题
- 通过重写创建一个新的 AOF 文件来替代现有的 AOF 文件，新的 AOF 文件不会包含任何浪费空间的冗余命令
**文件重写的实现原理**
- 不需要对现有的 AOF 文件进行任何操作
- 从数据库中直接读取键现在的值
- 用一条命令记录键值对，从而**代替**之前记录这个键值对的多条命令
**后台重写**
为不阻塞父进程， ， ：
- 执行客户端发来的命令
- 将执行后的写命令追加到 AOF 缓冲区
- 将执行后的写命令追加到 AOF 重写缓冲区
#### 默认配置
AOF 文件默认的配置如下：
```properties
############################## APPEND ONLY MODE ###############################
#开启AOF持久化方式
appendonly no
#AOF持久化文件名
appendfilename "appendonly.aof"
#每秒把缓冲区的数据fsync到磁盘
appendfsync everysec
# appendfsync no
#是否在执行重写时不同步数据到AOF文件
no-appendfsync-on-rewrite no
# 触发AOF文件执行重写的增长率
auto-aof-rewrite-percentage 100
#触发AOF文件执行重写的最小size
auto-aof-rewrite-min-size 64mb
#redis在恢复时，
aof-load-truncated yes
#是否打开混合开关
aof-use-rdb-preamble yes
```
## 过期策略
**过期策略用于处理过期缓存数据**。Redis采用过期策略： `定期删除` 。memcached采用过期策略：
### 定时过期
每个设置过期时间的key都需要创建一个定时器， ， ；
### 惰性过期
只有当访问一个key时， ， ， ， ，
### 定期过期
每隔一定的时间， ， ， ， ，
## 淘汰策略
Redis淘汰机制的存在是为了更好的使用内存， ， ， `maxmemory-policy` 的清理缓存方式。
```properties
# 配置最大内存限制
maxmemory 1000mb
# 配置淘汰策略
maxmemory-policy volatile-lru
```
### LRU(最近最少使用)
- `volatile-lru` ： ，
- `allkeys-lru` ： ，
### LFU(最近最不经常使用)
- `volatile-lfu` ： ，
- `allkeys-lfu` ： ，
### Random(随机淘汰)
- `volatile-random` ： ，
- `allkeys-random` ： ，
### TTL(过期时间)
- `volatile-ttl` ： ，
- `allkeys-random` ： ，
### No-Enviction(驱逐)
- `noenviction（ ） ：当达到最大内存时直接返回错误，不覆盖或逐出任何数据
## 部署架构
### 单节点(Single)
**优点**
- 架构简单，部署方便
- 高性价比：缓存使用时无需备用节点(单实例可用性可以用 supervisor 或 crontab 保证)，当然为了满足业务的高可用性，也可以牺牲一个备用节点，但同时刻只有一个实例对外提供服务
- 高性能
**缺点**
- 不保证数据的可靠性
- 在缓存使用，进程重启后，数据丢失，即使有备用的节点解决高可用性，但是仍然不能解决缓存预热问题，因此不适用于数据可靠性要求高的业务
- 高性能受限于单核CPU的处理能力(Redis是单线程机制)， ， ，
### 主从复制(Replication)
**基本原理**
主从复制模式中包含一个主数据库实例（ ） （ ） ， ：
.png)
优缺点.png)
### 哨兵(Sentinel)
Sentinel主要作用如下：
- **监控**：
- **通知**： ，
- **自动故障转移**： ， ， ，
.png)
优缺点.png)
### 集群(Cluster)
.png)
优缺点.png)
## 环境搭建
**Redis安装及配置**
Redis的安装十分简单， http://redis.io ](http://redis.io/ ) 。
- 下载一个最新版本的安装包，如 redis-version.tar.gz
- 解压 `tar zxvf redis-version.tar.gz`
- 执行 make (执行此命令可能会报错， ，
如果是 mac 电脑， ， ，
```shell
port 6379 # 指定端口为 6379，
daemonize yes # 指定后台运行
```
### 单节点(Single)
安装好redis之后，
```shell
$ < redishome > /bin/redis-server path/to/redis.config
```
假设我们没有配置后台运行（ ： ）
```shell
93825:C 20 Jan 2019 11:43:22.640 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
93825:C 20 Jan 2019 11:43:22.640 # Redis version=5.0.3, bits=64, commit=00000000, modified=0, pid=93825, just started
93825:C 20 Jan 2019 11:43:22.640 # Configuration loaded
93825:S 20 Jan 2019 11:43:22.641 * Increased maximum number of open files to 10032 (it was originally set to 256).
_._
_.-``__ ''-._
_.-`` `. `_ . ''-._ Redis 5.0.3 (00000000/0) 64 bit
.-`` .-```. ```\/ _.,_ ''-._
( ' , .-` | `, ) Running in standalone mode
|`-._`-...-` __ ...-.``-._|'` _.-'| Port: 6380
| `-._ ` ._ / _.-' | PID: 93825
`-._ ` -._ `-./ _.-' _ .-'
|`-._`-._ `-.__.-' _.-'_ .-'|
| `-._` -._ _.-'_ .-' | http://redis.io
`-._ ` -._`-.__.-'_.-' _.-'
|`-._`-._ `-.__.-' _.-'_ .-'|
| `-._` -._ _.-'_ .-' |
`-._ ` -._`-.__.-'_.-' _.-'
`-._ ` -.__.-' _.-'
`-._ _.-'
`-.__.-'
```
### 主从复制(Replication)
Redis主从配置非常简单， ：
**第一步**： （ ， ， ）
```shell
mkdir redis-master-slave
cp path/to/redis/conf/redis.conf path/to/redis-master-slave master.conf
cp path/to/redis/conf/redis.conf path/to/redis-master-slave slave.conf
```
**第二步**：修改配置
```shell
## master.conf
port 6379
## master.conf
port 6380
slaveof 127.0.0.1 6379
```
**第三步**：
```shell
redis-server path/to/redis-master-slave/master.conf
redis-server path/to/redis-master-slave/slave.conf
```
启动之后，打开两个命令行窗口，分别执行 `telnet localhost 6379` 和 `telnet localhost 6380` ，然后分别在两个窗口中执行 `info` 命令,可以看到：
```shell
# Replication
role:master
# Replication
role:slave
master_host:127.0.0.1
master_port:6379
```
主从配置没问题。然后在master 窗口执行 set 之后， ， ， ，
```shell
-READONLY You can't write against a read only replica.
```
因为从节点是只读的。
### 哨兵(Sentinel)
Sentinel是用来监控主从节点的健康情况。客户端连接Redis主从的时候， ， ， ， ， ，
**哨兵sentinel配置**
为了说明sentinel的用处， （ ） ， ：
```shell
mkdir redis-sentinel
cd redis-sentinel
cp redis/path/conf/redis.conf path/to/redis-sentinel/redis01.conf
cp redis/path/conf/redis.conf path/to/redis-sentinel/redis02.conf
cp redis/path/conf/redis.conf path/to/redis-sentinel/redis03.conf
touch sentinel.conf
```
上我们创建了 3个redis配置文件， ， ，
```shell
vim redis01.conf
port 63791
vim redis02.conf
port 63792
slaveof 127.0.0.1 63791
vim redis03.conf
port 63793
slaveof 127.0.0.1 63791
vim sentinel.conf
daemonize yes
port 26379
sentinel monitor mymaster 127.0.0.1 63793 1 # 下面解释含义
```
上面的主从配置都熟悉，只有哨兵配置 sentinel.conf， ：
```shell
mymaster # 为主节点名字，可以随便取，后面程序里边连接的时候要用到
127.0.0.1 63793 # 为主节点的 ip,port
1 # 后面的数字 1 表示选举主节点的时候，
```
**启动哨兵**
上面我们配置好了redis主从， ， ， ：
```shell
redis-server path/to/redis-sentinel/redis01.conf
redis-server path/to/redis-sentinel/redis02.conf
redis-server path/to/redis-sentinel/redis03.conf
redis-server path/to/redis-sentinel/sentinel.conf --sentinel
```
启动之后，可以分别连接到 3个redis上，
**模拟主节点宕机情况**
运行上面的程序（**注意， ， ， ， ） ， ，
```shell
redis-cli -p 63791 shutdown
```
### 集群(Cluster)
上述所做的这些工作只是保证了数据备份以及高可用， ， ， ， ：
- 单个redis并发有限
- 单个redis接收所有数据， ， ，
所以需要redis cluster 即redis集群。Redis 集群是一个提供在**多个Redis间节点间共享数据**的程序集。Redis集群并不支持处理多个keys的命令， ，
- 自动分割数据到不同的节点上
- 整个集群的部分节点失败或者不可达的情况下能够继续处理命令
为了配置一个redis cluster,我们需要准备至少6台redis， ？ ：
> Note that the minimal cluster that works as expected requires to contain at least three master nodes.
因为最小的redis集群， ， ， ， ，
**配置 redis cluster 集群**
上面提到， ：
```shell
# 主：
# 主：
# 主：
mkdir redis-cluster
cd redis-cluster
mkdir redis01 到 redis06 6个文件夹
cp redis.conf 到 redis01 ... redis06
# 修改端口, 分别配置3组主从关系
```
**启动redis集群**
上面的配置完成之后， ， ：
```shell
redis-5.0.3/src/redis-cli --cluster create 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375 127.0.0.1:6376 --cluster-replicas 1
```
**注意**， ，
```shell
ERR Invalid node address specified: localhost:6371
```
执行成功之后， ，
```shell
cluster_state:ok
cluster_slots_assigned:16384
cluster_slots_ok:16384
cluster_slots_pfail:0
cluster_slots_fail:0
cluster_known_nodes:6
cluster_size:3
cluster_current_epoch:6
cluster_my_epoch:1
cluster_stats_messages_ping_sent:1515
cluster_stats_messages_pong_sent:1506
cluster_stats_messages_sent:3021
cluster_stats_messages_ping_received:1501
cluster_stats_messages_pong_received:1515
cluster_stats_messages_meet_received:5
cluster_stats_messages_received:3021
```
我们可以看到`cluster_state:ok`， ，
## 拓展方案
### 分区(Partitioning)
指在面临**单机**的**存储空间**瓶颈时， ， ，
**使用方式**
- 客户端处理
和传统的数据库分库分表一样，可以从**key**入手，先进行计算，找到对应数据存储的实例在进行操作。
- **范围角度**， ，
- **哈希计算**，就像我们的**hashmap**一样， ， ，
- 使用代理中间件
我们可以开发独立的代理中间件， ， ， ， ，
**缺点**
- **无缘多key操作**， ，
- **维护成本**，由于每个实例在物理和逻辑上，都属于单独的一个节点，缺乏统一管理
- **灵活性有限**， ， ， ，
### 主从(Master-Slave)
分区暂时能解决**单点**无法容纳的**数据量问题**，
**使用方式**
- 作为主节点的Redis实例， ，
- 作为从节点的实例，使用配置文件或命令方式`REPLICAOF 主节点Host 主节点port`即可完成主从配置
**缺点**
- slave节点都是**只读**的，如果**写流量**大的场景，就有些力不从心
- **故障转移**不友好，主节点挂掉后，写处理就无处安放，需要**手工**的设定新的主节点，如使用`REPLICAOF no one` 晋升为主节点， ，
### 哨兵(Sentinel)
**主从**的手工故障转移，肯定让人很难接受，自然就出现了高可用方案-**哨兵**（ ） ，
**使用方式**
**Sentinel**的最小配置，一行即可：
```properties
sentinel monitor < 主节点别名> < 主节点host> < 主节点端口> < 票数>
```
只需要配置master即可， < 配置文件>`` 命令即可启用。哨兵数量建议在三个以上且为奇数。
**使用场景问题**
- 故障转移期间短暂的不可用，但其实官网的例子也给出了`parallel-syncs`参数来指定并行的同步实例数量，以免全部实例都在同步出现整体不可用的情况，相对来说要比手工的故障转移更加方便
- 分区逻辑需要自定义处理， ， ，
- 既然是还是主从， ， ， ，
### 集群(Cluster)
**Cluster**在分区管理上，使用了“**哈希槽**”(hash slot)这么一个概念，一共有**16384**个槽位，每个实例负责一部分**槽**，通过`CRC16（ ） & 16383`这样的公式，
**使用方式**
配置文件
```properties
cluster-enabled yes
cluster-config-file "redis-node.conf"
```
启动命令
```bash
redis-cli --cluster create 127.0.0.1:7000 127.0.0.1:7001 \
127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 \
--cluster-replicas 1
```
**存在问题**
- 虽然是对分区良好支持， ，
- 在select命令页有提到, 集群模式下只能使用一个库，虽然平时一般也是这么用的，但是要了解一下
- 运维上也要谨慎，俗话说得好，“**使用越简单底层越复杂**”，启动搭建是很方便，使用时面对带宽消耗，数据倾斜等等具体问题时，还需人工介入，或者研究合适的配置参数
## 常见问题
**题目**： ， ，
- **保留热点数据：**对于保留 Redis 热点数据来说，我们可以使用 Redis 的内存淘汰策略来实现，可以使用**allkeys-lru淘汰策略，
- **保证 Redis 只存20w的数据： ， ， ， （ ） ，
**题目： ， ，
限定 Redis 占用的内存，
**题目： ， ， ？
使用 keys 指令可以扫出指定模式的 key 列表。对方接着追问：如果这个 Redis 正在给线上的业务提供服务，那使用 keys 指令会有什么问题？这个时候你要回答 Redis 关键的一个特性： ，
# Kafka
# RocketMQ
RocketMQ 是阿里巴巴开源的分布式消息中间件。支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。它里面有几个区别于标准消息中件间的概念，
**功能优势**
- **削峰填谷**：主要解决瞬时写压力大于应用服务能力导致消息丢失、系统奔溃等问题
- **应用解耦**：解决不同重要程度、不同能力级别系统之间依赖导致一死全死
- **提升性能**：当存在一对多调用时，可以发一条消息给消息系统，让消息系统通知相关系统
- **蓄流压测**：线上有些链路不好压测，可以通过堆积一定量消息再放开来压测
- **异步处理**：不需要同步执行的远程调用可以有效提高响应时间
## 架构设计
### 部署模型
### 角色
#### Broker
- 理解成RocketMQ本身
- Broker主要用于Producer和Consumer接收和发送消息
- Broker会定时向NameSrver提交自己的信息
- 是消息中间件的消息存储、转发服务器
- 每个Broker节点在启动时都会遍历NameServer列表， ， ，
#### NameServer
- 理解成Zookeeper的效果， ，
- 底层由Netty实现， ，
- NameServer是服务发现者， （ ） ， ， ，
- NameServer可以部署多个， ， ， ，
- NameServer集群间互不通信，
- NameServer内存式存储， ，
#### Producer
- 消息的生产者
- 随机选择其中一个NameServer节点建立长连接， （ ， ）
- 接下来向提供Topic服务的Master建立长连接（ ） ，
#### Consumer
- 消息的消费者
- 通过NameServer集群获得Topic的路由信息，
- 由于Master和Slave都可以读取消息，
### 核心流程
- Broker都注册到Nameserver上
- Producer发消息的时候会从Nameserver上获取发消息的Topic信息
- Producer向提供服务的所有Master建立长连接，
- Consumer通过NameServer集群获得Topic的路由信息
- Consumer会与所有的Master和所有的Slave都建立连接进行监听新消息
### 实现原理
RocketMQ由NameServer注册中心集群、Producer生产者集群、Consumer消费者集群和若干Broker（ ） ， ：
- Broker在启动的时候去向所有的NameServer注册， ，
- Producer在发送消息的时候从NameServer获取Broker服务器地址，
- Conusmer消费消息的时候同样从NameServer获取Broker地址，
## 核心概念
### Message(消息)
消息载体。Message发送或者消费的时候必须指定Topic。Message有一个可选的Tag项用于过滤消息，
### Topic(主题)
消息的逻辑分类， ，
### Queue(队列)
1个Topic会被分为N个Queue， ， ， ， ， ， ， ， ， ，
### Tag(标签)
Tag 是 Topic 的进一步细分， ， ， ，
### 消费模式(Message Model)
消息模型： （ ） （ ）
#### 集群模式(Clustering)
生产者往某个队列里面发送消息，一个队列可以存储多个生产者的消息，一个队列也可以有多个消费者，但是消费者之间是竞争关系，即每条消息只能被一个消费者消费。
- 每条消息只需要被处理一次，
- 在消息重投时，不能保证路由到同一台机器上
- 消费状态由Broker维护
#### 广播模式(Broadcasting)
**为了解决一条消息能被多个消费者消费的问题**，发布/订阅模型就来了。该模型是将消息发往一个`Topic`即主题中，所有订阅了这个 `Topic` 的订阅者都能消费这条消息。
- 消费进度由Consumer维护
- 保证每个消费者都消费一次消息
- 消费失败的消息不会重投
### Message Order(消息顺序)
消息顺序： （ ） （ ）
#### 顺序（ ）
#### 并发（ ）
### Producer Group(生产组)
消息生产者组。标识发送同一类消息的Producer， ， ， ， ， ， ， （ ）
### Consumer Group(消费组)
消息消费者组。标识一类Consumer的集合名称， ， ， ， ， ，
注： RocketMQ要求同一个Consumer Group的消费者必须要拥有相同的注册信息，
### Offset
在 Topic 的消费过程中，由于消息需要被不同的组进行多次消费，所以消费完的消息并不会立即被删除，这就需要 RocketMQ 为每个消费组在每个队列上维护一个消费位置（ ） ， ， ， ， ， ，
## 核心设计
### 消息清理
Broker中的消息被消费后不会立即删除， ， ， （ ） ：
- 检查这个文件最后访问时间
- 判断是否大于过期时间
- 指定时间删除，
```java
/**
* {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#isTimeToDelete()}
*/
private boolean isTimeToDelete() {
// when = "04";
String when = DefaultMessageStore.this.getMessageStoreConfig().getDeleteWhen();
// 是04点，
if (UtilAll.isItTimeToDo(when)) {
return true;
}
// 不是04点，
return false;
}
/**
* {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#deleteExpiredFiles()}
*/
private void deleteExpiredFiles() {
// isTimeToDelete()这个方法是判断是不是凌晨四点，是的话就执行删除逻辑。
if (isTimeToDelete()) {
// 默认是72， ，
long fileReservedTime = 48 * 60 * 60 * 1000;
deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(72 * 60 * 60 * 1000, xx, xx, xx);
}
}
/**
* {@link org.apache.rocketmq.store.CommitLog#deleteExpiredFile()}
*/
public int deleteExpiredFile(xxx) {
// 这个方法的主逻辑就是遍历查找最后更改时间+过期时间， （ ）
return this.mappedFileQueue.deleteExpiredFileByTime(72 * 60 * 60 * 1000, xx, xx, xx);
}
```
### push or pull
RocketMQ没有真正意义的push， ， ， ， `longPollingEnable` 标记是否开启长轮询，默认开启。源码如下：
```java
// {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()}
// 拉取消息，
this.pullAPIWrapper.pullKernelImpl(pullCallback);
```
**为什么要主动拉取消息而不使用事件监听方式？**
事件驱动方式是建立好长连接， （ ） ， ， ， ，
### 负载均衡
RocketMQ通过Topic在多Broker中分布式存储实现。
#### Producer端
发送端指定message queue发送消息到相应的broker，
- 提升写入吞吐量， ，
- 消息分布在多broker中，
**默认策略是随机选择：**
- producer维护一个index
- 每次取节点会自增
- index向所有broker个数取余
- 自带容错策略
**其他实现：**
- SelectMessageQueueByHash
- - hash的是传入的args
- SelectMessageQueueByRandom
- SelectMessageQueueByMachineRoom 没有实现
也可以自定义实现**MessageQueueSelector**接口中的select方法
```java
MessageQueue select(final List< MessageQueue > mqs, final Message msg, final Object arg);
```
#### Consumer端
采用的是平均分配算法来进行负载均衡。
**其他负载均衡算法**
- 平均分配策略(默认)(AllocateMessageQueueAveragely)
- 环形分配策略(AllocateMessageQueueAveragelyByCircle)
- 手动配置分配策略(AllocateMessageQueueByConfig)
- 机房分配策略(AllocateMessageQueueByMachineRoom)
- 一致性哈希分配策略(AllocateMessageQueueConsistentHash)
- 靠近机房策略(AllocateMachineRoomNearby)
**当消费负载均衡Consumer和Queue不对等的时候会发生什么？
Consumer和Queue会优先平均分配， ， ， ， ， ， ，
## 最佳实践
### Producer
- **Topic**： ，
- **Tag**： ， ，
- **key**：每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引(哈希索引)，应用可以通过 topic， ， ，
- **日志**：消息发送成功或者失败，要打印消息日志，务必要打印 send result 和key 字段
- **send**： ， ， ，
- **SEND_OK**：消息发送成功
- **FLUSH_DISK_TIMEOUT**：消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
- **FLUSH_SLAVE_TIMEOUT**： ， ， ， ，
- **SLAVE_NOT_AVAILABLE**： ， ， ， ，
- **订阅关系一致**
多个Group ID订阅了多个Topic，
### Consumer
- **消费幂等**
为了防止消息重复消费导致业务处理异常， ， ：
- **发送时消息重复**
- **投递时消息重复**
- **负载均衡时消息重复**（ ）
- **日志**：消费时记录日志，以便后续定位问题
- **批量消费**：尽量使用批量方式消费方式，可以很大程度上提高消费吞吐量
### 事务消息
MQ与DB一致性原理（ ）
事务消息就是MQ提供的类似XA的分布式事务能力， ， ， ：
- 生产者先发送一条半事务消息到MQ
- MQ收到消息后返回ack确认
- 生产者开始执行本地事务
- 如果事务执行成功发送commit到MQ，
- 如果MQ长时间未收到生产者的二次确认commit或者rollback，
- 生产者查询事务执行最终状态
- 根据查询事务状态再次提交二次确认
如果MQ收到二次确认commit， ， ，
## 保证顺序
RocketMQ的消息是存储到Topic的Queue里面的， （ ）
顺序消息（
- **顺序发布**
- **顺序消费**
顺序消息包含两种类型：
- **分区顺序**：
- **全局顺序**：
对于两个订单的消息的原始数据： （ ） ：
- 在发送时， ， ， ，
- 在存储时， ，
### 保持顺序发送
消息被发送时保持顺序。
### 保持顺序发送存储
消息被存储时保持和发送的顺序一致。
### 保持顺序消费
消息被消费时保持和存储的顺序一致。
#### MQPullConsumer
MQPullConsumer由用户控制线程， ， ，
#### MQPushConsumer
对于PushConsumer， ，
## 消息不丢失
一条消息从生产到被消费，将会经历三个阶段：
- 生产阶段：
- 存储阶段：消息将会存储在 Broker 端磁盘中
- 消息阶段：
以上任一阶段都可能会丢失消息，我们只要找到这三个阶段丢失消息原因，采用合理的办法避免丢失，就可以彻底解决消息丢失的问题。
### 生产阶段
Producer通过网络将消息发送给Broker， ，
失败会自动重试， ， ， ， ， ， ，
总结：**同步发送+自动重试机制+多个Master节点**
#### 同步发送
有三种send方法， ，
- **同步发送**： ， ， ，
- **异步发送**：在回调方法里可以得知是否发送成功
- **单向发送（ ）
```java
/**
* {@link org.apache.rocketmq.client.producer.DefaultMQProducer}
*/
// 同步发送
public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {}
// 异步发送，
public void send(Message msg,SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {}
// 单向发送，不关心发送结果，最不靠谱
public void sendOneway(Message msg) throws MQClientException, RemotingException, InterruptedException {}
```
#### 失败重试
发送消息如果失败或者超时了， ， ， ：
```java
producer.setRetryTimesWhenSendFailed(10);
```
底层源码逻辑如下：
```java
/**
* {@link org.apache.rocketmq.client.producer.DefaultMQProducer#sendDefaultImpl(Message, CommunicationMode, SendCallback, long)}
*/
// 自动重试次数， ， ， ，
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
for (; times < timesTotal ; times + + ) {
// 选择发送的消息queue
MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
if (mqSelected != null) {
try {
// 真正的发送逻辑，
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
switch (communicationMode) {
case ASYNC:
return null;
case ONEWAY:
return null;
case SYNC:
// 如果发送失败了， ， ，
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
continue;
}
}
// 发送成功的话，将发送结果返回给调用者
return sendResult;
default:
break;
}
} catch (RemotingException e) {
continue;
} catch (...) {
continue;
}
}
}
```
#### 故障切换
假设Broker宕机了， ， ， ， ， ， ， ， ，
### Broker存储阶段
若想很严格的保证Broker存储消息阶段消息不丢失， ， ：
```properties
# master 节点配置
flushDiskType = SYNC_FLUSH
brokerRole=SYNC_MASTER
# slave 节点配置
brokerRole=slave
flushDiskType = SYNC_FLUSH
```
#### 设置Broker同步刷盘策略
**设置Broker同步刷盘策略**。默认情况下，消息只要到了 Broker 端，将会优先保存到内存中，然后立刻返回确认响应给生产者。随后 Broker 定期批量的将一组消息从内存异步刷入磁盘。这种方式减少 I/O 次数，可以取得更好的性能，但是如果发生机器断电，异常宕机等情况，消息还未及时刷入磁盘，就会出现丢失消息的情况。
若想保证 Broker 端不丢消息，保证消息的可靠性，我们需要将消息保存机制修改为同步刷盘方式，即消息**存储磁盘成功**，才会返回响应。修改 Broker 端配置如下：
```properties
# 默认情况为 ASYNC_FLUSH
flushDiskType = SYNC_FLUSH
```
若 Broker 未在同步刷盘时间内（**默认为 5s**）完成刷盘，将会返回 `SendStatus.FLUSH_DISK_TIMEOUT` 状态给生产者。
#### 等待Master和Slave刷盘完
等待Master和Slave刷盘完。即使Broker设置了同步刷盘策略， ， ， ， ，
```properties
# 默认为 ASYNC_MASTER
brokerRole=SYNC_MASTER
```
### 消费阶段
消费失败了其实也是消息丢失的一种变体。
只有当消费模式为 **MessageModel.CLUSTERING(集群模式)** 时， ， ，
#### 手动ACK确认
消费者会先把消息拉取到本地， ， ，
```java
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List< MessageExt > msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
try{
for (MessageExt msg : msgs) {
String str = new String(msg.getBody());
System.out.println(str);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} catch(Throwable t){
log.error("消费异常:{}", msgs, t);
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
});
```
#### 消费异常自动重试
- 业务消费方返回 ConsumeConcurrentlyStatus.RECONSUME_LATER
- 业务消费方返回 null
- 业务消费方主动/被动抛出异常
针对以上3种情况下， （ ） ， `broker.conf` 配置文件中配置 Consumer 侧重试次数及时间间隔（**距离第1次发送的时间间隔**）
```properties
messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
```
消费者客户端，首先判断消费端有没有显式设置最大重试次数 MaxReconsumeTimes， ，
```java
private int getMaxReconsumeTimes() {
// default reconsume times: 16
if (this.defaultMQPushConsumer.getMaxReconsumeTimes() == -1) {
return 16;
} else {
return this.defaultMQPushConsumer.getMaxReconsumeTimes();
}
}
```
#### 消费超时无线重试
如果是消费超时情况，
#### 死信队列
死信的处理逻辑：
- 首先判断消息当前重试次数是否大于等于 16，
- 只要满足上述的任意一个条件，设置新的 topic（ ） ：
- 进行前置属性的添加
- 将死信消息投递到上述步骤 2 建立的死信 topic 对应的死信队列中并落盘，使消息持久化
最后单独启动一个死信队列的消费者进行消费，然后进行人工干预处理失败的消息。
## 消息幂等
在所有消息系统中消费消息有三种模式：`at-most-once`（最多一次）、`at-least-once`（最少一次）和 `exactly-only-once` （精确仅一次），分布式消息系统都是在三者间取平衡，前两者是可行的并且被广泛使用。
- `at-most-once` ： ， ， ，
- `at-lease-once` ：消息投递后，消费完成后，向服务器返回 ACK， ， ， ，
- `exactly-only-once` ： ， ， ：
- 发送消息阶段，不允许发送重复消息
- 消费消息阶段，不允许消费重复的消息
# Zookeeper
**下载地址：**http://www.apache.org/dist/zookeeper/
## 节点角色
### Leader
- 负责处理客户端发送的`读、写事务请求`。这里的事务请求可以理解这个请求具有事务的 ACID 特性
- 同步`写事务请求`给其他节点，且需要保证事务的顺序性
- 状态为 LEADING
### Follower
- 负责处理客户端发送的读请求
- 转发写事务请求给 Leader
- 参与 Leader 的选举
- 状态为 FOLLOWING
### Observer
和 Follower 一样，唯一不同的是，不参与 Leader 的选举，且状态为 OBSERING。可以用来线性扩展读的 QPS。
## 核心流程设计
### 启动阶段 Leader选举
Zookeeper 刚启动的时候，多个节点需要找出一个 Leader。怎么找呢， ， ， ：
- 节点 A 先`投票给自己`，投票信息包含`节点 id（ `ZXID` ，如 （ ） ， ，
- 节点 B 先`投票给自己`， （ ）
- 然后节点 A 和 B 将自己的投票信息`投票`给集群中`所有节点`
- 节点 A 收到节点 B 的投票信息后，`检查`下节点 B 的状态是否是`本轮投票`，以及是否是`正在选举(LOOKING)`的状态
- 投票 PK： ， ， （ ， ） ， `不需要更新` 投票信息，但是下一轮还需要再次将投票发出去。这个时候节点 A 的投票信息为（ ， ） ， ：
- 统计投票：每一轮投票，都会统计每台节点收到的投票信息，判断是否有`过半的节点`收到了`相同`的投票信息。节点 A 和 节点 B 收到的投票信息都为（ ， ） ， ， ，
- 更新节点状态：节点 A 作为 `Follower` ，更新状态为 FOLLOWING， `Leader` ，更新状态为 LEADING
### 运行期间Leader宕机处理
在 Zookeeper 运行期间， ， ：
- 剩下的 Follower 进行选举，
- 投票信息中的 zxid 用的是本地磁盘日志文件中的。如果这个节点上的 zxid 较大，就会被当选为 Leader。如果 Follower 的 zxid 都相同，则 Follower 的节点 id 较大的会被选为 Leader
### 节点之间同步数据流程
不同的客户端可以分别连接到主节点或备用节点。而客户端发送读写请求时是不知道自己连的是Leader 还是 Follower， （ ） ， ：
- **第一阶段**： ，
- **第二阶段**： ， ，
我们来看下 Leader 同步数据的流程：
- ① 客户端发送写事务请求
- ② Leader 收到写请求后，转化为一个 "proposal01：
- ③ 发送 proposal 给其他 Follower
- ④ Follower 收到 proposal 后，
接着我们看下 Follower 收到 Leader 发送的 proposal 事务请求后，怎么处理的：
- ⑤ Follower 返回 ack 给 Leader
- ⑥ Leader 收到超过一半的 ack，
- ⑦ Leader 将磁盘中的日志文件的 proposal 加载到 znode 内存数据结构中
- ⑧ Leader 发送 commit 消息给所有 Follower 和 Observer
- ⑨ Follower 收到 commit 消息后，将 磁盘中数据加载到 znode 内存数据结构中
现在Leader和Follower的数据都是在内存数据中的， ，
### ZAB顺序一致性原理
Leader 发送 proposal 时，其实会为每个 Follower 创建一个队列，都往各自的队列中发送 proposal。如下图所示是 Zookeeper 的消息广播流程：
客户端发送了三条写事务请求，对应的 proposal 为
```properties
proposal01:zxid1
proposal02:zxid2
proposal03:zxid3
```
Leader 收到请求后，依次放到队列中，然后 Follower 依次从队列中获取请求，这样就保证了数据的顺序性。
### Zookeeper是否是强一致性
> 官方定义：顺序一致性。
不保证强一致性，为什么呢？
因为 Leader 再发送 commit 消息给所有 Follower 和 Observer 后，它们并不是同时完成 commit 的。比如因为`网络原因`，不同节点收到的 commit 较晚，那么提交的时间也较晚，就会出现多个节点的数据不一致，但是经过短暂的时间后，所有节点都 commit 后，数据就保持同步了。另外 Zookeeper 支持强一致性，就是手动调用 sync 方法来保证所有节点都 commit 才算成功。
这里有个问题：如果某个节点 commit 失败，那么 Leader 会进行重试吗？如何保证数据的一致性？
### Leader 宕机数据丢失问题
**第一种情况**：假设 Leader 已经将消息写入了本地磁盘，但是还没有发送 proposal 给 Follower， ，
- zxid 的高 32 位自增 1 一次，高 32 位代表 Leader 的版本号。
- zxid 的低 32 位自增 1，
当老 Leader 恢复后，会转成 Follower， ，
**第二种情况**：如果 Leader 成功发送了 commit 消息给 Follower， ，
那么就需要选出磁盘日志中 zxid 最大的 Follower， ，
## ZK特性
Zookeeper主要靠其 ** 分布式数据一致性** 为集群提供 ** 分布式协调服务**，即指在集群的节点中进行可靠的消息传递，来协调集群的工作。主要具有如下特点：
- **最终一致性：**Client无论连接到哪个Server， ，
- **可靠性：**如果一个消息或事物被一台Server接受，
- **实时性：**Zookeeper不能保证强一致性， ， ， ， ，
- **原子性：**更新只能成功或者失败，没有中间状态
- **顺序性：**包括 ** 全序（ ） ** 因果顺序（ ）
- **全序：**如果消息a在消息b之前发送，
- **因果顺序：**如果消息a在消息b之前发生（ ） ， ，
## ZK角色
- **Server（ ）
- **Leader（ ） ： ， ，
- **Learner（ ）
- **Follower（ ） ：
- **Observer（ ） ： ， ， ， ，
- **Client（ ） ：
每个Server在工作过程中有三种状态：
- **LOOKING： ，
- **LEADING：
- **FOLLOWING： ，
Zookeeper集群中，
- **领导者（ ）
Leader服务器是整个ZooKeeper集群工作机制中的核心， ：
- 事务请求的唯一调度和处理者，保证集群事务处理的顺序性
- 集群内部各服务的调度者
- **跟随者（ ）
Follower服务器是ZooKeeper集群状态的跟随者， ：
- 处理客户端非事务请求，
- 参与事务请求Proposal的投票
- 参与Leader选举投票
- **观察者（ ） ，
Observer是3.3.0 版本开始引入的一个服务器角色， ：
- 处理客户端的非事务请求，转发事务请求给 Leader 服务器
- 不参与任何形式的投票
## 数据模型
Zookeeper 的数据模型：
- 层次化的目录结构， ，
- 每个节点在zookeeper中叫做znode,并且其有一个唯一的路径标识
- 节点Znode可以包含数据和子节点，
- Znode中的数据可以有多个版本， ，
- 客户端应用可以在节点上设置监视器
- 节点不支持部分读写，而是一次性完整读写
## Server工作状态
服务器具有四种状态，分别是 LOOKING、FOLLOWING、LEADING、OBSERVING。
- LOOKING： ， ，
- FOLLOWING：
- LEADING：
- OBSERVING：
## 运行模式
Zookeeper 有三种运行模式：单机模式、伪集群模式和集群模式。
- **单机模式**：这种模式一般适用于开发测试环境，一方面我们没有那么多机器资源，另外就是平时的开发调试并不需要极好的稳定性。
- **集群模式**：一个 ZooKeeper 集群通常由一组机器组成，一般 3 台以上就可以组成一个可用的 ZooKeeper 集群了。组成 ZooKeeper 集群的每台机器都会在内存中维护当前的服务器状态，并且每台机器之间都会互相保持通信。
- **伪集群模式**： ， ， ， ， ，
## Leader选举
### 服务器启动的Leader选举
zookeeper集群初始化阶段， （ ） ，
- 服务器1（ ） ， ，
- 服务器2（ ） ， ，
- 每个服务器发出一个投票
服务器1 和 服务器2都将自己作为Leader服务器进行投票， ： ， （ ， ） ， ， （ ） ， （ ） ，
- 接受来自各个服务器的投票
每个服务器都会接受来自其他服务器的投票。同时， ，
- 处理投票
收到其他服务器的投票， ， ：
- 优先检查ZXID。ZXID比较大的服务器优先作为leader。
- 如果ZXID相同的话， ， （ ） ， （ ） ， ， ， ， （ ） ， ， ，
- 统计投票
每次投票后， ， ， （ ） ，
- 服务器3（ ） ，
跟前面流程一致， ， ， ， （ ） ， ， ；
- 服务器4启动，
此时服务器1， ， ， ： ， ；
- 服务器5启动，
同理， ， ；
- 投票结束，
### 服务器运行期间的Leader选举
zookeeper集群的五台服务器（ ） ， ， ，
- 变更状态
Leader 服务器挂了之后， ，
- 每个服务器发起投票
每个服务器都把票投给自己， ， ， （ ） ， （ ， ） ， （ ） （ ） ，
- 接受来自各个服务器的投票
- 处理投票
投票规则是跟Zookeeper集群启动期间一致的， ， ，
- 统计投票
- 改变服务器状态
## 节点（ ）
**① 节点组成**
每个znode由4部分组成:
- **path：
- **stat： ， ，
- **data：
- **children：
**② 节点类型**
- **PERSISTENT（ ）
客户端与zookeeper断开连接后，
- **PERSISTENT_SEQUENTIAL（ ）
客户端与zookeeper断开连接后， ，
- **EPHEMERAL（ ）
客户端与zookeeper断开连接后，
- **EPHEMERAL_SEQUENTIAL（ ）
客户端与zookeeper断开连接后， ，
## 关键词
- **zxid（ ）
ZooKeeper状态的每一次改变， ， ， ， ， ，
- **session（ ）
在Client和Server通信之前， ， ， ， ，
## Watcher监听机制
Zookeeper 允许客户端向服务端的某个Znode注册一个Watcher监听， ， ，
- ZooKeeper的Watcher机制主要包括客户端线程、客户端 WatcherManager、Zookeeper服务器三部分
- 客户端向ZooKeeper服务器注册Watcher的同时，
- 当zookeeper服务器触发watcher事件后， ，
**Watcher特性总结**
- **「一次性:」** 一个Watch事件是一个一次性的触发器。一次性触发，
- **「异步的：」** Zookeeper服务器发送watcher的通知事件到客户端是异步的， ， ，
- **「轻量级：」** Watcher 通知非常简单，它只是通知发生了事件，而不会传递事件对象内容
- **「客户端串行：」** 执行客户端 Watcher 回调的过程是一个串行同步的过程
- 注册 watcher用getData、exists、getChildren方法
- 触发 watcher用create、delete、setData方法
## ZXID
ZXID有两部分组成：
- 任期： ， ，
- 事务计数器：单调递增，每生效一次写入，计数器加一
ZXID的低32位是计数器， ， ， ， ，
## 工作流程
**① 客户端发起的操作的主要流程**
Leader可以执行增删改查操作， ， ， ， ， ， ，
**② 客户端的请求什么时候才能算处理成功？为什么说集群过半机器宕机后无法再工作？**
Leader在收到客户端提交过来的任务后， ， ， ， ， ， ， ， ， ，
- **Leader工作流程**
Leader主要有三个功能：
- 恢复数据
- 维持与Learner的心跳，
- Learner的消息类型主要有如下四种， ， ：
- **PING消息：
- **REQUEST消息： ，
- **ACK消息： ， ，
- **REVALIDATE消息：
- **Follower工作流程**
Follower主要有四个功能：
- 向Leader发送消息请求（ ）
- 接收Leader消息并进行处理
- 接收Client的请求， ，
- 返回Client结果
Follower的消息循环处理如下几种来自Leader的消息：
- **PING消息：
- **PROPOSAL消息： ，
- **COMMIT消息：
- **UPTODATE消息：
- **REVALIDATE消息： ，
- **SYNC消息： ， ，
- **Observer工作流程**
对于Observer的流程不再叙述，
## ZAB协议
ZooKeeper并没有完全采用Paxos算法， （ ， ） ， ：
- **崩溃恢复模式（ ） ： ， ， ，
- **消息广播模式（ ） ： ， ， （ ： ）
**① 崩溃恢复模式（ ）
为了使Leader挂了后系统能正常工作， ：
- 已经被处理的消息不能丢
- 被丢弃的消息不能再次出现
**② 消息广播模式（ ）
广播的过程实际上是一个简化的二阶段提交过程：
- Leader 接收到消息请求后，将消息赋予一个全局唯一的 64 位自增 id， ： ，
- Leader 通过先进先出队列（通过 TCP 协议来实现，以此实现了全局有序这一特性）将带有 zxid 的消息作为一个提案（ ）
- 当 follower 接收到 proposal，
- 当 leader 接收到合法数量的 ACKs 后，
- 当 follower 收到消息的 COMMIT 命令时，就会执行该消息
**总结**
个人认为 Zab 协议设计的优秀之处有两点， ， ； ， ， ， ， （ ）
**③ ZAB 的四个阶段**
- **Leader election（ ） ： ， ， （ ， ） ，
- **Discovery（ ） ： ， ， ，
- **Synchronization（ ） ： ， ，
- **Broadcast（ ） ： ， ， ，
**④ JAVA版ZAB协议**
- **Fast Leader Election**：
前面提到 FLE 会选举拥有最新提议历史（ ） ，
- 选epoch最大的
- epoch相等，
- epoch和zxid都相等， （ ）
- **Recovery Phase**： ， ：
- **Broadcast Phase**：暂无
## ZK选举过程
最开始集群启动时，
当Leader崩溃或者Leader失去大多数的Follower， ， ， ：
- 选举线程由当前Server发起选举的线程担任， ，
- 选举线程首先向所有Server发起一次询问（ ）
- 选举线程收到回复后，验证是否是自己发起的询问(验证zxid是否一致)， ， ，
- 收到所有Server回复以后， ，
- 线程将当前zxid最大的Server设置为当前Server要推荐的Leader， ， ， ， ， ，
**分析结论**
要使Leader获得多数Server的支持， ， ，
## Zookeeper安装
### 单机模式
**第一步：安装部署**
```shell
# 下载解压
wget http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz
tar -zxvf zookeeper-3.4.9.tar.gz
# 设置全局变量
vim ~/.bash_profile
# 最后一行加入
export ZOOKEEPER_HOME=/home/zookeeper/zookeeper-3.4.9
export PATH=$ZOOKEEPER_HOME/bin:$PATH
# 使之生效
source ~/.bash_profile
# 复制配置文件
cp /home/zookeeper/zookeeper-3.4.9/conf/zoo_sample.cfg /home/zookeeper/zookeeper-3.4.9/conf/zoo.cfg
```
**第二步：配置信息**
```shell
# 心跳间隔
tickTime=2000
# 保存数据目录
dataDir=/home/zookeeper/zookeeper-3.4.9/dataDir
# 保存日志目录
dataLogDir=/home/zookeeper/zookeeper-3.4.9/dataLogDir
# 客户端连接Zookeeper的端口
clientPort=2181
# 允许follower连接并同步到leader的初始化连接时间(心跳倍数)，超过则连接失败
initLimit=5
# 表示leader和follower之间发送消息时, 请求和应答的最大时间长度
syncLimit=2
```
### 集群模式
**第一步：安装部署**
安装方式参考单机模式。
**第二步：配置信息**
在dataDir根目录下新建myid文件， ：
```shell
# 1表示当前集群的id号
echo "1" > myid
```
在单机配置情况下，新增下述参数：
```shell
# 格式：
# X表示myid(范围1~255)
# A是该server所在的IP地址
# B配置该server和集群中的leader交换消息所使用的端口,即数据同步端口
# C配置选举leader时所使用的端口
server.1=10.24.1.62:2888:3888
server.2=10.24.1.63:2888:3888
server.3=10.24.1.64:2888:3888
```
### 运维命令
```shell
# 启动ZK服务器
./zkServer.sh start
# 使用ZK Client连接指定服务器
./zkCli.sh -server 127.0.0.1:2181
# 查看ZK服务状态
./zkServer.sh status
# 停止ZK服务
./zkServer.sh stop
# 重启ZK服务
./zkServer.sh restart
```
### zoo.cfg配置参数
```shell
# 客户端连接server的端口，
clientPort=2181
# 存储快照文件snapshot的目录
dataDir=/User/lry/zookeeper/data
# ZK中的最小时间单元，
tickTime=5000
# 事务日志输出目录
dataLogDir=/User/lry/zookeeper/datalog
# Server端最大允许的请求堆积数，
globalOutstandingLimit=1000
# 每个事务日志文件的大小，
preAllocSize=64
# 每进行snapCount次事务日志输出后，
snapCount=100000
# 单个客户端与单台服务器之间的连接数的限制， ， ， ，
maxClientCnxns=60
# 对于多网卡的机器，
clientPortAddress=10.24.22.56
# Session超时时间限制， ，
minSessionTimeoutmaxSessionTimeout
# Follower在启动过程中， ，
initLimit
# 在运行过程中， ， ， ， ，
syncLimit
# 默认情况下， ， ， ， ， ，
leaderServes=yes
# server.[myid]=[hostname]:[数据同步和其它通讯端口]:[选举投票通讯]
server.x=[hostname]:nnnnn[:nnnnn]
# 对机器分组和权重设置
group.x=nnnnn[:nnnnn]weight.x=nnnnn
# Leader选举过程中， ，
cnxTimeout
# 每个节点最大数据量，
jute.maxbuffer
```
# Netty
## Netty逻辑架构
Netty内部逻辑的流转：
### 网络通信层
网络通信层的职责是**执行网络I/O的操作**， ， ， ：
- **BootStrap和ServerBootStrap**
主要负责整个Netty程序的启动、初始化、服务器连接等过程， ， ， ：
- Bootstrap可用于连接远端服务器，
- ServerBootStrap则用于服务端启动绑定本地端口， ， （ ， ）
- **Channel**
提供了基本的API用于网络I/O操作， ， ， ， ， ，
### 事件调度层
事件调度层的职责是通过Reactor线程模型对各类事件进行聚合处理， （ ） ， ：
- **EventLoopGroup、EventLoop**
EventLoopGroup是Netty的核心处理引擎， ， ， ， ， ，
- 一个 EventLoopGroup 往往包含一个或者多个 EventLoop。EventLoop 用于处理 Channel 生命周期内的所有 I/O 事件，如 accept、connect、read、write 等 I/O 事件
- EventLoop 同一时间会与一个线程绑定，每个 EventLoop 负责处理多个 Channel
- 每新建一个 Channel，
其实EventLoopGroup是Netty Reactor线程模型的具体实现方式， ， ：
- **单线程模型**： ，
- **多线程模型**： ，
- **主从多线程模型**： ， ， ， ， ，
### 服务编排层
服务编排层的职责是负责组装各类服务， ， ：
- **ChannelPipeline**
ChannelPipeline 是 Netty 的核心编排组件，负责组装各种 ChannelHandler， ，
ChannelPipeline 是线程安全的，因为每一个新的 Channel 都会对应绑定一个新的 ChannelPipeline。一个 ChannelPipeline 关联一个 EventLoop，
ChannelPipeline、ChannelHandler 都是高度可定制的组件。开发者可以通过这两个核心组件掌握对 Channel 数据操作的控制权。下面我们看一下 ChannelPipeline 的结构图：
ChannelPipeline中包含入站ChannelInboundHandler和出站 ChannelOutboundHandler两种处理器， ：
- **ChannelHandler & ChannelHandlerContext**
数据的编解码工作以及其他转换工作实际都是通过 ChannelHandler 处理的。ChannelHandlerContext 用于保存 ChannelHandler 上下文，通过 ChannelHandlerContext 可以知道 ChannelPipeline 和 ChannelHandler 的关联关系。ChannelHandlerContext 可以实现 ChannelHandler 之间的交互，
每创建一个 Channel 都会绑定一个新的 ChannelPipeline，
## Reactor线程模型
Reactor线程模型运行机制的四个步骤，
- 连接注册： ，
- 事件轮询：
- 事件分发：
- 任务处理： ，
### 单Reactor单线程
上图描述了 Reactor 的单线程模型结构，在 Reactor 单线程模型中，所有 I/O 操作（包括连接建立、数据读写、事件分发等），都是由一个线程完成的。单线程模型逻辑简单，缺陷也十分明显：
- 一个线程支持处理的连接数非常有限，
- 当多个事件被同时触发时，只要有一个事件没有处理完，其他后面的事件就无法执行，这就会造成消息积压及请求超时
- 线程在处理 I/O 事件时，
- 如果 I/O 线程一直处于满负荷状态，很可能造成服务端节点不可用
### 单Reactor多线程
由于单线程模型有性能方面的瓶颈， ， ， ，
### 主从Reactor多线程
主从多线程模型由多个 Reactor 线程组成，每个 Reactor 线程都有独立的 Selector 对象。MainReactor 仅负责处理客户端连接的 Accept 事件，连接建立成功后将新创建的连接对象注册至 SubReactor。再由 SubReactor 分配线程池中的 I/O 线程与其连接绑定，它将负责连接生命周期内所有的 I/O 事件。
Netty 推荐使用主从多线程模型，这样就可以轻松达到成千上万规模的客户端连接。在海量客户端并发请求的场景下，主从多线程模式甚至可以适当增加 SubReactor 线程的数量，从而利用多核能力提升系统的吞吐量。
## 核心设计
### Netty EventLoop原理
EventLoop 这个概念其实并不是 Netty 独有的，它是一种事件等待和处理的程序模型，可以解决多线程资源消耗高的问题。例如 Node.js 就采用了 EventLoop 的运行机制，不仅占用资源低，而且能够支撑了大规模的流量访问。
EventLoop 可以说是 Netty 的调度中心， ：
#### EventLoop运行模式
上图展示了 EventLoop 通用的运行模式。每当事件发生时，应用程序都会将产生的事件放入事件队列当中，然后 EventLoop 会轮询从队列中取出事件执行或者将事件分发给相应的事件监听者执行。事件执行的方式通常分为**立即执行、延后执行、定期执行**几种。
#### NioEventLoop原理
在 Netty 中 EventLoop 可以理解为 Reactor 线程模型的事件处理引擎，每个 EventLoop 线程都维护一个 Selector 选择器和任务队列 taskQueue。它主要负责处理 I/O 事件、普通任务和定时任务。Netty 中推荐使用 NioEventLoop 作为实现类，那么 Netty 是如何实现 NioEventLoop 的呢？首先我们来看 NioEventLoop 最核心的 run() 方法源码：
```java
protected void run() {
for (;;) {
try {
try {
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.BUSY_WAIT:
case SelectStrategy.SELECT:
// 轮询 I/O 事件
select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
selector.wakeup();
}
default:
}
} catch (IOException e) {
rebuildSelector0();
handleLoopException(e);
continue;
}
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
// 处理 I/O 事件
processSelectedKeys();
} finally {
// 处理所有任务
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
// 处理 I/O 事件
processSelectedKeys();
} finally {
final long ioTime = System.nanoTime() - ioStartTime;
// 处理完 I/O 事件，再处理异步任务队列
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Throwable t) {
handleLoopException(t);
}
}
}
```
上述源码的结构比较清晰， ，
#### 事件处理机制
结合Netty的整体架构， ， ：
- **BossEventLoopGroup** 和 **WorkerEventLoopGroup** 包含一个或者多个 NioEventLoop
BossEventLoopGroup 负责监听客户端的 Accept 事件，当事件触发时，将事件注册至 WorkerEventLoopGroup 中的一个 NioEventLoop 上。每新建一个 Channel，
- NioEventLoop 完成数据读取后，会调用绑定的 ChannelPipeline 进行事件传播
ChannelPipeline 也是**线程安全**的，数据会被传递到 ChannelPipeline 的第一个 ChannelHandler 中。数据处理完成后，将加工完成的数据再传递给下一个 ChannelHandler，
NioEventLoop 无锁串行化的设计不仅使系统吞吐量达到最大化，而且降低了用户开发业务逻辑的难度，不需要花太多精力关心线程安全问题。虽然单线程执行避免了线程切换，但是它的缺陷就是不能执行时间过长的 I/O 操作，一旦某个 I/O 事件发生阻塞，那么后续的所有 I/O 事件都无法执行，甚至造成事件积压。在使用 Netty 进行程序开发时，我们一定要对 ChannelHandler 的实现逻辑有充分的风险意识。
NioEventLoop 线程的可靠性至关重要，一旦 NioEventLoop 发生阻塞或者陷入空轮询，就会导致整个系统不可用。在 JDK 中， Epoll 的实现是存在漏洞的，即使 Selector 轮询的事件列表为空，
抛开其它细枝末节， ， ：
```java
long time = System.nanoTime();
if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
selectCnt = 1;
} else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 & &
selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
selector = selectRebuildSelector(selectCnt);
selectCnt = 1;
break;
}
```
Netty提供了一种检测机制判断线程是否可能陷入空轮询， ：
- 每次执行 select 操作之前记录当前时间 currentTimeNanos
- **time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos**，如果事件轮询的持续时间大于等于 timeoutMillis， ， ，
- Netty 引入了计数变量 selectCnt。在正常情况下， （ ）
Netty 采用这种方法巧妙地规避了 JDK Bug。异常的 Selector 中所有的 SelectionKey 会重新注册到新建的 Selector 上，重建完成之后异常的 Selector 就可以废弃了。
#### 任务处理机制
NioEventLoop 不仅负责处理 I/O 事件，还要兼顾执行任务队列中的任务。任务队列遵循 FIFO 规则，
- **普通任务**：通过 NioEventLoop 的 execute() 方法向任务队列 taskQueue 中添加任务。例如 Netty 在写数据时会封装 WriteAndFlushTask 提交给 taskQueue。taskQueue 的实现类是多生产者单消费者队列 MpscChunkedArrayQueue， ，
- **定时任务**：通过调用 NioEventLoop 的 schedule() 方法向定时任务队列 scheduledTaskQueue 添加一个定时任务，用于周期性执行该任务。例如，心跳消息发送等。定时任务队列 scheduledTaskQueue 采用优先队列 PriorityQueue 实现
- **尾部队列**：
下面结合任务处理 runAllTasks 的源码结构，分析下 NioEventLoop 处理任务的逻辑，源码实现如下：
```java
protected boolean runAllTasks(long timeoutNanos) {
// 1. 合并定时任务到普通任务队列
fetchFromScheduledTaskQueue();
// 2. 从普通任务队列中取出任务
Runnable task = pollTask();
if (task == null) {
afterRunningAllTasks();
return false;
}
// 3. 计算任务处理的超时时间
final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
long runTasks = 0;
long lastExecutionTime;
for (;;) {
// 4. 安全执行任务
safeExecute(task);
runTasks ++;
// 5. 每执行 64 个任务检查一下是否超时
if ((runTasks & 0x3F) == 0) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
if (lastExecutionTime >= deadline) {
break;
}
}
task = pollTask();
if (task == null) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
break;
}
}
// 6. 收尾工作
afterRunningAllTasks();
this.lastExecutionTime = lastExecutionTime;
return true;
}
```
在代码中以注释的方式标注了具体的实现步骤，可以分为 6 个步骤：
- **fetchFromScheduledTaskQueue函数**：将定时任务从 scheduledTaskQueue 中取出，聚合放入普通任务队列 taskQueue 中，只有定时任务的截止时间小于当前时间才可以被合并
- **从普通任务队列taskQueue中取出任务**
- **计算任务执行的最大超时时间**
- **safeExecute函数**：安全执行任务，实际直接调用的 Runnable 的 run() 方法
- **每执行 64 个任务进行超时时间的检查**：如果执行时间大于最大超时时间，则立即停止执行任务，避免影响下一轮的 I/O 事件的处理
- **最后获取尾部队列中的任务执行**
#### EventLoop最佳实践
在日常开发中用好 EventLoop 至关重要，这里结合实际工作中的经验给出一些 EventLoop 的最佳实践方案：
- 网络连接建立过程中三次握手、安全认证的过程会消耗不少时间。这里建议采用 Boss 和 Worker 两个 EventLoopGroup，
- 由于 Reactor 线程模式适合处理耗时短的任务场景，对于耗时较长的 ChannelHandler 可以考虑维护一个业务线程池，将编解码后的数据封装成 Task 进行异步处理，避免 ChannelHandler 阻塞而造成 EventLoop 不可用
- 如果业务逻辑执行时间较短，建议直接在 ChannelHandler 中执行。例如编解码操作，这样可以避免过度设计而造成架构的复杂性
- 不宜设计过多的 ChannelHandler。对于系统性能和可维护性都会存在问题， ，
### ChannelPipeline
Pipeline 的字面意思是管道、流水线。它在 Netty 中起到的作用，和一个工厂的流水线类似。原始的网络字节流经过 Pipeline， ， ，
#### ChannelPipeline内部结构
ChannelPipeline 可以看作是 ChannelHandler 的容器载体，它是由一组 ChannelHandler 实例组成的，内部通过双向链表将不同的 ChannelHandler 链接在一起，如下图所示。当有 I/O 读写事件触发时，
根据网络数据的流向，
#### ChannelHandler接口设计
整个ChannelHandler是围绕I/O事件的生命周期所设计的， ：
**① ChannelInboundHandler的事件回调方法与触发时机**
| 事件回调方法 | 触发时机 |
| ------------------------- | -------------------------------------------------- |
| channelRegistered | Channel 被注册到 EventLoop |
| channelUnregistered | Channel 从 EventLoop 中取消注册 |
| channelActive | Channel 处于就绪状态，可以被读写 |
| channelInactive | Channel 处于非就绪状态Channel 可以从远端读取到数据 |
| channelRead | Channel 可以从远端读取到数据 |
| channelReadComplete | Channel 读取数据完成 |
| userEventTriggered | 用户事件触发时 |
| channelWritabilityChanged | Channel 的写状态发生变化 |
**② ChannelOutboundHandler的事件回调方法与触发时机**
ChannelOutboundHandler 的事件回调方法非常清晰，直接通过 ChannelOutboundHandler 的接口列表可以看到每种操作所对应的回调方法，如下图所示。这里每个回调方法都是在相应操作执行之前触发，在此就不多做赘述了。此外 ChannelOutboundHandler 中绝大部分接口都包含ChannelPromise 参数，以便于在操作完成时能够及时获得通知。
#### ChannelPipeline事件传播机制
上述ChannelPipeline可分为入站 ChannelInboundHandler 和出站 ChannelOutboundHandler 两种处理器，与此对应传输的事件类型可以分为**Inbound 事件**和**Outbound 事件**。
- **Inbound事件**： ， （ ）
- **Outbound事件**： ， （ ）
代码示例体验 ChannelPipeline 的事件传播机制：
```java
serverBootstrap.childHandler(new ChannelInitializer< SocketChannel > () {
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline()
.addLast(new SampleInBoundHandler("SampleInBoundHandlerA", false))
.addLast(new SampleInBoundHandler("SampleInBoundHandlerB", false))
.addLast(new SampleInBoundHandler("SampleInBoundHandlerC", true));
ch.pipeline()
.addLast(new SampleOutBoundHandler("SampleOutBoundHandlerA"))
.addLast(new SampleOutBoundHandler("SampleOutBoundHandlerB"))
.addLast(new SampleOutBoundHandler("SampleOutBoundHandlerC"));
}
}
```
执行结果：
#### ChannelPipeline异常传播机制
ChannelPipeline 事件传播的实现采用了经典的责任链模式， ？
建议用户自定义的异常处理器代码示例如下：
```java
public class ExceptionHandler extends ChannelDuplexHandler {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
if (cause instanceof RuntimeException) {
System.out.println("Handle Business Exception Success.");
}
}
}
```
加入统一的异常处理器后，可以看到异常已经被优雅地拦截并处理掉了。这也是 Netty 推荐的最佳异常处理实践。
### 定时器TimerTask
### 时间轮HashedWheelTimer
时间轮其实就是一种环形的数据结构，可以想象成时钟，分成很多格子，一个格子代表一段时间。并用一个链表保存在该格子上的计划任务，同时一个指针随着时间一格一格转动，并执行相应格子中的所有到期任务。任务通过时间取模决定放入那个格子。
在网络通信中管理上万的连接， ， ， ，
Netty 的时间轮 `HashedWheelTimer` 给出了一个**粗略的定时器实现**，之所以称之为粗略的实现是**因为该时间轮并没有严格的准时执行定时任务**，而是在每隔一个时间间隔之后的时间节点执行，并执行当前时间节点之前到期的定时任务。
当然具体的定时任务的时间执行精度可以通过调节 HashedWheelTimer 构造方法的时间间隔的大小来进行调节，在大多数网络应用的情况下，由于 IO 延迟的存在，并**不会严格要求具体的时间执行精度**，所以默认的 100ms 时间间隔可以满足大多数的情况，不需要再花精力去调节该时间精度。
**HashedWheelTimer的特点**
- 从源码分析可以看出，其实 HashedWheelTimer 的时间精度并不高，误差能够在 100ms 左右，同时如果任务队列中的等待任务数量过多，可能会产生更大的误差
- 但是 HashedWheelTimer 能够处理非常大量的定时任务，且每次定位到要处理任务的候选集合链表只需要 O(1) 的时间，而 Timer 等则需要调整堆，是 O(logN) 的时间复杂度
- HashedWheelTimer 本质上是`模拟了时间的轮盘`，将大量的任务拆分成了一个个的小任务列表，能够有效`节省 CPU 和线程资源`
**源码解读**
```java
public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit,
int ticksPerWheel, boolean leakDetection, long maxPendingTimeouts) {
......
}
```
- `threadFactory` ：自定义线程工厂，用于创建线程对象
- `tickDuration` ：间隔多久走到下一槽（相当于时钟走一格）
- `unit` ：
- `ticksPerWheel` ：一圈有多个槽
- `leakDetection` ：是否开启内存泄漏检测
- `maxPendingTimeouts` ：
**优缺点**
- **优点**
- 可以添加、删除、取消定时任务
- 能高效的处理大批定时任务
- **缺点**
- 对内存要求较高，占用较高的内存
- 时间精度要求不高
**定时任务方案**
目前主流的一些定时任务方案：
- Timer
- ScheduledExecutorService
- ThreadPoolTaskScheduler（ ）
- Netty的schedule（ ）
- Netty的HashedWheelTimer（ ）
- Kafka的TimingWheel（ ）
使用案例：
```java
// 构造一个 Timer 实例
Timer timer = new HashedWheelTimer();
// 提交一个任务，让它在 5s 后执行
Timeout timeout1 = timer.newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) {
System.out.println("5s 后执行该任务");
}
}, 5, TimeUnit.SECONDS);
// 再提交一个任务，让它在 10s 后执行
Timeout timeout2 = timer.newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) {
System.out.println("10s 后执行该任务");
}
}, 10, TimeUnit.SECONDS);
// 取消掉那个 5s 后执行的任务
if (!timeout1.isExpired()) {
timeout1.cancel();
}
// 原来那个 5s 后执行的任务，已经取消了。这里我们反悔了，我们要让这个任务在 3s 后执行
// 我们说过 timeout 持有上、下层的实例，所以下面的 timer 也可以写成 timeout1.timer()
timer.newTimeout(timeout1.task(), 3, TimeUnit.SECONDS);
```
### 无锁队列mpsc queue
### FastThreadLocal
### ByteBuf
### 编解码协议
netty-codec模块主要负责编解码工作， ，
### 拆包粘包
- 拆包/粘包的解决方案
- **消息长度固定**：每个数据报文都需要一个固定的长度。当接收方累计读取到固定长度的报文后，就认为已经获得一个完整的消息。当发送方的数据小于固定长度时，则需要空位补齐。消息定长法使用非常简单，但是缺点也非常明显，无法很好设定固定长度的值，如果长度太大会造成字节浪费，长度太小又会影响消息传输，所以在一般情况下消息定长法不会被采用。
- **特定分隔符**
既然接收方无法区分消息的边界，那么我们可以在每次发送报文的尾部加上特定分隔符，接收方就可以根据特殊分隔符进行消息拆分。
- **消息长度 + 消息内容**
消息长度 + 消息内容是项目开发中最常用的一种协议，如上展示了该协议的基本格式。消息头中存放消息的总长度，例如使用 4 字节的 int 值记录消息的长度，消息体实际的二进制的字节数据。接收方在解析数据时，首先读取消息头的长度字段 Len，
## 核心流程
### 服务端启动流程
Netty 服务端的启动过程大致分为三个步骤：
- 配置线程池
- Channel 初始化
- 端口绑定
#### 配置线程池
**单线程模式**
Reactor 单线程模型所有 I/O 操作都由一个线程完成，所以只需要启动一个 EventLoopGroup 即可。
```java
EventLoopGroup group = new NioEventLoopGroup(1);
ServerBootstrap b = new ServerBootstrap();
b.group(group);
```
**多线程模式**
Reactor 单线程模型有非常严重的性能瓶颈，因此 Reactor 多线程模型出现了。在 Netty 中使用 Reactor 多线程模型与单线程模型非常相似，区别是 NioEventLoopGroup 可以不需要任何参数，它默认会启动 2 倍 CPU 核数的线程。当然，你也可以自己手动设置固定的线程数。
```java
EventLoopGroup group = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(group);
```
**主从多线程模式**
在大多数场景下，我们采用的都是主从多线程 Reactor 模型。Boss 是主 Reactor， ， ，
```java
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup);
```
#### Channel 初始化
**设置Channel类型**
```java
// 客户端Channel
b.channel(NioSocketChannel.class);
b.channel(OioSocketChannel.class);
// 服务端Channel
b.channel(NioServerSocketChannel.class);
b.channel(OioServerSocketChannel.class);
b.channel(EpollServerSocketChannel.class);
// UDP
b.channel(NioDatagramChannel.class);
b.channel(OioDatagramChannel.class);
```
**注册ChannelHandler**
ServerBootstrap 的 childHandler() 方法需要注册一个 ChannelHandler。ChannelInitializer是实现了 ChannelHandler接口的匿名类，
```java
b.childHandler(new ChannelInitializer< SocketChannel > () {
@Override
// HTTP 编解码处理器
// HTTPContent 压缩处理器
// HTTP 消息聚合处理器
// 自定义业务逻辑处理器
});
```
**设置Channel参数**
ServerBootstrap 设置 Channel 属性有option和childOption两个方法，
```java
b.option(ChannelOption.SO_KEEPALIVE, true);
```
常用参数如下：
| 参数名 | 描述信息 |
| ---------------------- | ------------------------------------------------------------ |
| SO_KEEPALIVE | 设置为 true 代表启用了 TCP SO_KEEPALIVE 属性，
| SO_BACKLOG | 已完成三次握手的请求队列最大长度，同一时刻服务端可能会处理多个连接，在高并发海量连接的场景下，该参数应适当调大 |
| TCP_NODELAY | Netty 默认是 true， ， ，
| SO_SNDBUF | TCP 数据发送缓冲区大小 |
| SO_RCVBUF | TCP数据接收缓冲区大小，
| SO_LINGER | 设置延迟关闭的时间，等待缓冲区中的数据发送完成 |
| CONNECT_TIMEOUT_MILLIS | 建立连接的超时时间 |
#### 端口绑定
bind() 方法会真正触发启动，
```java
ChannelFuture f = b.bind(8080).sync();
```
## Netty流程
从功能上，流程可以分为服务启动、建立连接、读取数据、业务处理、发送数据、关闭连接以及关闭服务。整体流程如下所示(图中没有包含关闭的部分)：
## 线程模型
Netty通过Reactor模型基于多路复用器接收并处理用户请求， ， ， ， ， ， ， ，
### 单线程Reactor线程模型
下图演示了单线程reactor线程模型， ， ， ， ， ， ， ，
特别的， ， ， ， ，
### 多线程Reactor线程模型
所谓多线程reactor线程模型， ，
### 混合型Reactor线程模型
混合型reactor线程模型， ：
- 将任务处理切分成多个阶段进行，每个阶段处理完自己的部分之后，转发到下一个阶段进行处理。不同的阶段之间的执行是异步的，可以认为每个阶段都有一个独立的线程池。
- 不同的类型的任务，有着不同的处理流程，划分时需要划分成不同的阶段。如下图蓝色是一种任务、绿色是另一种任务，两种任务有着不同的执行流程
### Netty-Reactor线程模型
图中大致包含了5个步骤， ， ， ：
- 设置服务端ServerBootStrap启动参数
- 通过ServerBootStrap的bind方法启动服务端， ，
- Client发起连接CONNECT请求， ， ，
- ACCEPT事件触发后， ，
- childGroup中的NioEventLoop不断检测自己管理的NioSocketChannel是否有读写事件准备好， ，
## ByteBuf
### 工作流程
ByteBuf维护两个不同的索引： `写索引(writerIndex)` 。如下图所示:
- `ByteBuf` 维护了 `readerIndex` 和 `writerIndex` 索引
- 当 `readerIndex > writerIndex` 时，则抛出 `IndexOutOfBoundsException`
- `ByteBuf` 容量 = `writerIndex`
- `ByteBuf` 可读容量 = `writerIndex` - `readerIndex`
- `readXXX()` 和 `writeXXX()` 方法将会推进其对应的索引。自动推进
- `getXXX()` 和 `setXXX()` 方法将对 `writerIndex` 和 `readerIndex` 无影响
### 使用模式
ByteBuf本质是一个由不同的索引分别控制读访问和写访问的字节数组。ByteBuf共有三种模式： `复合缓冲区模式(Composite Buffer)` 。
#### 堆缓冲区模式(Heap Buffer)
堆缓冲区模式又称为`支撑数组(backing array)`。将数据存放在JVM的堆空间，
- **优点**： ，
- **缺点**： ，
```java
public static void heapBuffer() {
// 创建Java堆缓冲区
ByteBuf heapBuf = Unpooled.buffer();
if (heapBuf.hasArray()) { // 是数组支撑
byte[] array = heapBuf.array();
int offset = heapBuf.arrayOffset() + heapBuf.readerIndex();
int length = heapBuf.readableBytes();
handleArray(array, offset, length);
}
}
```
#### 直接缓冲区模式(Direct Buffer)
Direct Buffer属于堆外分配的直接内存， ， ， ，
- **优点**: 使用Socket传递数据时性能很好，
- **缺点**: 相对于堆缓冲区而言，
```java
public static void directBuffer() {
ByteBuf directBuf = Unpooled.directBuffer();
if (!directBuf.hasArray()) {
int length = directBuf.readableBytes();
byte[] array = new byte[length];
directBuf.getBytes(directBuf.readerIndex(), array);
handleArray(array, 0, length);
}
}
```
#### 复合缓冲区模式(Composite Buffer)
Composite Buffer是Netty特有的缓冲区。本质上类似于提供一个或多个ByteBuf的组合视图，
- Composite Buffer是一个组合视图。它提供一种访问方式让使用者自由组合多个ByteBuf，
- Composite Buffer不支持访问其支撑数组。因此如果要访问， ，
下图是将两个ByteBuf：
```java
public static void byteBufComposite() {
// 复合缓冲区，只是提供一个视图
CompositeByteBuf messageBuf = Unpooled.compositeBuffer();
ByteBuf headerBuf = Unpooled.buffer(); // can be backing or direct
ByteBuf bodyBuf = Unpooled.directBuffer(); // can be backing or direct
messageBuf.addComponents(headerBuf, bodyBuf);
messageBuf.removeComponent(0); // remove the header
for (ByteBuf buf : messageBuf) {
System.out.println(buf.toString());
}
}
```
### 字节级操作
#### 随机访问索引
ByteBuf的索引与普通的Java字节数组一样。第一个字节的索引是0， ：
- readXXX()和writeXXX()方法将会推进其对应的索引readerIndex和writerIndex。自动推进
- getXXX()和setXXX()方法用于访问数据，
```java
public static void byteBufRelativeAccess() {
ByteBuf buffer = Unpooled.buffer(); // get reference form somewhere
for (int i = 0; i < buffer.capacity ( ) ; i + + ) {
byte b = buffer.getByte(i); // 不改变readerIndex值
System.out.println((char) b);
}
}
```
#### 顺序访问索引
Netty的ByteBuf同时具有读索引和写索引， ， ： ** 可写字节区域** 。
#### 可丢弃字节区域
可丢弃字节区域是指：[0，
- discardReadBytes()效果 ----- 将可读字节区域(CONTENT)[readerIndex, writerIndex)往前移动readerIndex位，
- discardReadBytes()方法会移动可读字节区域内容(CONTENT)。如果频繁调用，会有多次数据复制开销，对性能有一定的影响
#### 可读字节区域
可读字节区域是指:[readerIndex, writerIndex)之间的区域。任何名称以read和skip开头的操作方法，
#### 可写字节区域
可写字节区域是指:[writerIndex, capacity)之间的区域。任何名称以write开头的操作方法都将改变writerIndex的值。
#### 索引管理
- markReaderIndex()+resetReaderIndex() ----- markReaderIndex()是先备份当前的readerIndex， ，
- readerIndex(int) ----- 设置readerIndex为固定的值
- writerIndex(int) ----- 设置writerIndex为固定的值
- clear() ----- 效果是: readerIndex=0, writerIndex(0)。不会清除内存
- 调用clear()比调用discardReadBytes()轻量的多。仅仅重置readerIndex和writerIndex的值， ，
#### 查找操作(indexOf)
查找ByteBuf指定的值。类似于，
- 最简单的方法 —— indexOf()
- 利用ByteProcessor作为参数来查找某个指定的值
```java
public static void byteProcessor() {
ByteBuf buffer = Unpooled.buffer(); //get reference form somewhere
// 使用indexOf()方法来查找
buffer.indexOf(buffer.readerIndex(), buffer.writerIndex(), (byte)8);
// 使用ByteProcessor查找给定的值
int index = buffer.forEachByte(ByteProcessor.FIND_CR);
}
```
#### 派生缓冲——视图
派生缓冲区为ByteBuf提供了一个访问的视图。视图仅仅提供一种访问操作， ， ，
- duplicate()
- slice()
- slice(int, int)
- Unpooled.unmodifiableBuffer(...)
- Unpooled.wrappedBuffer(...)
- order(ByteOrder)
- readSlice(int)
**理解**
- 上面的6中方法， ， ，
- 请注意： ， ，
- 如果需要拷贝现有缓冲区的真实副本，
- 使用派生缓冲区，避免了复制内存的开销，有效提高程序的性能
```java
public static void byteBufSlice() {
Charset utf8 = Charset.forName("UTF-8");
ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);
ByteBuf sliced = buf.slice(0, 15);
System.out.println(sliced.toString(utf8));
buf.setByte(0, (byte)'J');
assert buf.getByte(0) == sliced.getByte(0); // return true
}
public static void byteBufCopy() {
Charset utf8 = Charset.forName("UTF-8");
ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);
ByteBuf copy = buf.copy(0, 15);
System.out.println(copy.toString(utf8));
buf.setByte(0, (byte)'J');
assert buf.getByte(0) != copy.getByte(0); // return true
}
```
#### 读/写操作
如上文所提到的，有两种类别的读/写操作:
- get()和set()操作 ----- 从给定的索引开始，并且保持索引不变
- read()和write()操作 ----- 从给定的索引开始，并且根据已经访问过的字节数对索引进行访问
- 下图给出get()操作API，
#### 更多操作
下面的两个方法操作字面意思较难理解，给出解释:
- **hasArray()**： ， ： ，
- **array()**： ， ，
### ByteBuf分配
创建和管理ByteBuf实例的多种方式： **ByteBufUtil类** 。
#### 按序分配：
Netty通过接口ByteBufAllocator实现了(ByteBuf的)池化。Netty提供池化和非池化的ButeBufAllocator:
- `ctx.channel().alloc().buffer()` ：
- `ByteBufAllocator.DEFAULT.buffer()` ：
- `ByteBufAllocator.DEFAULT` ：有两种类型: UnpooledByteBufAllocator.DEFAULT(非池化)和PooledByteBufAllocator.DEFAULT(池化)。对于Java程序， ，
- 可以通过BootStrap中的Config为每个Channel提供独立的ByteBufAllocator实例
解释:
- 上图中的buffer()方法，
- ByteBufAllocator.DEFAULT ----- 可能是池化，也可能是非池化。默认是池化(PooledByteBufAllocator.DEFAULT)
#### Unpooled缓冲区——非池化
Unpooled提供静态的辅助方法来创建未池化的ByteBuf。
注意:
- 上图的buffer()方法，
- wrappedBuffer() ----- 创建一个视图，
创建ByteBuf代码:
```java
public void createByteBuf(ChannelHandlerContext ctx) {
// 1. 通过Channel创建ByteBuf
ByteBuf buf1 = ctx.channel().alloc().buffer();
// 2. 通过ByteBufAllocator.DEFAULT创建
ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer();
// 3. 通过Unpooled创建
ByteBuf buf3 = Unpooled.buffer();
}
```
#### ByteBufUtil类
ByteBufUtil类提供了用于操作ByteBuf的静态的辅助方法: hexdump()和equals
- hexdump()：
- equals()：
### 引用计数
Netty4.0版本中为ButeBuf和ButeBufHolder引入了引用计数技术。请区别引用计数和可达性分析算法(jvm垃圾回收)
- 谁负责释放：一般来说，是由最后访问(引用计数)对象的那一方来负责将它释放
- buffer.release()：
- buffer.retain()：
- buffer.refCnt()：返回当前对象引用计数值
- buffer.touch()：记录当前对象的访问位置，主要用于调试
- 引用计数并非仅对于直接缓冲区(direct Buffer)。ByteBuf的三种模式: 堆缓冲区(heap Buffer)、直接缓冲区(dirrect Buffer)和复合缓冲区(Composite Buffer)都使用了引用计数，某些时候需要程序员手动维护引用数值
```java
public static void releaseReferenceCountedObject(){
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
// 引用计数加1
buffer.retain();
// 输出引用计数
buffer.refCnt();
// 引用计数减1
buffer.release();
}
```
## Zero-Copy
**Netty** 的`Zero-copy` 体现在如下几个个方面:
- **通过CompositeByteBuf实现零拷贝**： ， ，
- **通过wrap操作实现零拷贝**：通过`wrap`操作， ，
- 通过slice操作实现零拷贝： ， ，
- **通过FileRegion实现零拷贝**：通过 `FileRegion` 包装的`FileChannel.tranferTo` 实现文件传输，可以直接将文件缓冲区的数据发送到目标 Channel，
### 零拷贝操作
#### 通过CompositeByteBuf实现零拷贝
```java
ByteBuf header = null;
ByteBuf body = null;
// 传统合并header和body：
ByteBuf allBuf = Unpooled.buffer(header.readableBytes() + body.readableBytes());
allBuf.writeBytes(header);
allBuf.writeBytes(body);
// 合并header和body：
CompositeByteBuf compositeByteBuf = Unpooled.compositeBuffer();
compositeByteBuf.addComponents(true, header, body);
// 底层封装了 CompositeByteBuf 操作
ByteBuf allByteBuf = Unpooled.wrappedBuffer(header, body);
```
#### 通过wrap操作实现零拷贝
```java
byte[] bytes = null;
// 传统方式：
ByteBuf byteBuf = Unpooled.buffer();
byteBuf.writeBytes(bytes);
// wrap方式：
// 即最后我们生成的生成的ByteBuf对象是和bytes数组共用了同一个存储空间, 对bytes的修改也会反映到ByteBuf对象中
ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes);
```
#### 通过slice操作实现零拷贝
slice 操作和 wrap 操作刚好相反,`Unpooled.wrappedBuffer` 可以将多个 ByteBuf 合并为一个, 而 slice 操作可以将一个 **ByteBuf ** `切片` 为多个共享一个存储区域的 ByteBuf 对象. ByteBuf 提供了两个 slice 操作方法:
```java
public ByteBuf slice();
public ByteBuf slice(int index, int length);
```
不带参数的`slice`方法等同于`buf.slice(buf.readerIndex(), buf.readableBytes())` 调用, 即返回 buf 中可读部分的切片. 而 `slice(int index, int length)` 方法相对就比较灵活了, 我们可以设置不同的参数来获取到 buf 的不同区域的切片。
用 `slice` 方法产生 header 和 body 的过程是没有拷贝操作的, header 和 body 对象在内部其实是共享了 byteBuf 存储空间的不同部分而已，即：
#### 通过FileRegion实现零拷贝
Netty 中使用 FileRegion 实现文件传输的零拷贝, 不过在底层 FileRegion 是依赖于 **Java NIO** `FileChannel.transfer` 的零拷贝功能。当有了 FileRegion 后, 我们就可以直接通过它将文件的内容直接写入 **Channel** 中, 而不需要像传统的做法: 拷贝文件内容到临时 **buffer** , 然后再将 **buffer** 写入 **Channel.** 通过这样的零拷贝操作, 无疑对传输大文件很有帮助。
### 传统IO的流程
- **「第一步」**：将文件通过 ** 「DMA」** 技术从磁盘中拷贝到内核缓冲区
- **「第二步」**：将文件从内核缓冲区拷贝到用户进程缓冲区域中
- **「第三步」**：将文件从用户进程缓冲区中拷贝到 socket 缓冲区中
- **「第四步」**： ** 「DMA」** 技术拷贝到网卡
### 零拷贝整体流程图
## TCP粘包拆包
### 粘包拆包图解
假设客户端分别发送了两个数据包D1和D2给服务端， ， ：
- 服务端分两次读取到两个独立的数据包， ，
- 服务端一次接收到了两个数据包， ，
- 服务端分两次读取到数据包， ， ，
- 服务端分两次读取到数据包， ，
- 当TCP缓存再小一点的话，
### 产生原因
产生粘包和拆包问题的主要原因是， ， ， ：
- 如果一次请求发送的数据量比较小， ， ，
- 如果一次请求发送的数据量比较大， ， ， ，
### 解决方案
#### 固定长度
对于使用固定长度的粘包和拆包场景，可以使用：
- `FixedLengthFrameDecoder` ：每次读取固定长度的消息，如果当前读取到的消息不足指定长度，那么就会等待下一个消息到达后进行补足。其使用也比较简单，只需要在构造函数中指定每个消息的长度即可。
```java
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 这里将FixedLengthFrameDecoder添加到pipeline中，
ch.pipeline().addLast(new FixedLengthFrameDecoder(20));
// 将前一步解码得到的数据转码为字符串
ch.pipeline().addLast(new StringDecoder());
// 这里FixedLengthFrameEncoder是我们自定义的，
ch.pipeline().addLast(new FixedLengthFrameEncoder(20));
// 最终的数据处理
ch.pipeline().addLast(new EchoServerHandler());
}
```
#### 指定分隔符
对于通过分隔符进行粘包和拆包问题的处理， ：
- `LineBasedFrameDecoder` ：通过换行符，即`\n`或者` \r\n`对数据进行处理
- `DelimiterBasedFrameDecoder` ：通过用户指定的分隔符对数据进行粘包和拆包处理
```java
@Override
protected void initChannel(SocketChannel ch) throws Exception {
String delimiter = "_$";
// 将delimiter设置到DelimiterBasedFrameDecoder中， ，
// 被按照_$进行分隔， ， ，
// 读取到分隔符，则舍弃当前数据段，因为其很有可能是由于码流紊乱造成的
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,
Unpooled.wrappedBuffer(delimiter.getBytes())));
// 将分隔之后的字节数据转换为字符串数据
ch.pipeline().addLast(new StringDecoder());
// 这是我们自定义的一个编码器，主要作用是在返回的响应数据最后添加分隔符
ch.pipeline().addLast(new DelimiterBasedFrameEncoder(delimiter));
// 最终处理数据并且返回响应的handler
ch.pipeline().addLast(new EchoServerHandler());
}
```
#### 数据包长度字段
处理粘拆包的主要思想是在生成的数据包中添加一个长度字段，用于记录当前数据包的长度。
- `LengthFieldBasedFrameDecoder` ：按照参数指定的包长度偏移量数据对接收到的数据进行解码，从而得到目标消息体数据。解码过程如下图所示：
- `LengthFieldPrepender` ：在响应的数据前面添加指定的字节数据，这个字节数据中保存了当前消息体的整体字节数据长度。编码过程如下图所示：
```java
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 这里将LengthFieldBasedFrameDecoder添加到pipeline的首位，
// 进行长度字段解码，这里也会对数据进行粘包和拆包处理
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2));
// LengthFieldPrepender是一个编码器，
ch.pipeline().addLast(new LengthFieldPrepender(2));
// 对经过粘包和拆包处理之后的数据进行json反序列化，
ch.pipeline().addLast(new JsonDecoder());
// 对响应数据进行编码，
ch.pipeline().addLast(new JsonEncoder());
// 处理客户端的请求的数据，并且进行响应
ch.pipeline().addLast(new EchoServerHandler());
}
```
#### 自定义粘包拆包器
可以通过实现`MessageToByteEncoder`和`ByteToMessageDecoder`来实现自定义粘包和拆包处理的目的。
- `MessageToByteEncoder` ：
- `ByteToMessageDecoder` ：
## 高性能
- **IO线程模型**：同步非阻塞，用最少的资源做更多的事
- **内存零拷贝**：尽量减少不必要的内存拷贝，实现了更高效率的传输
- **内存池设计**：申请的内存可以重用，主要指直接内存。内部实现是用一颗二叉查找树管理内存分配情况
- **串形化处理读写**：避免使用锁带来的性能开销
- **高性能序列化协议**：支持 protobuf 等高性能序列化协议
## 操作系统调优
### 文件描述符
- 设置系统最大文件句柄数
```bash
# 查看
cat /proc/sys/fs/file-max
# 修改
在/etc/sysctl.conf插入fs.file-max=1000000
# 配置生效
sysctl -p
```
- 设置单进程打开的最大句柄数
默认单进程打开的最大句柄数是 `1024` ，通过 `ulimit -a` 可以查看相关参数，示例如下：
```powershell
[root@test ~]# ulimit -a
core file size (blocks, -c) 0
data seg size (kbytes, -d) unlimited
scheduling priority (-e) 0
file size (blocks, -f) unlimited
pending signals (-i) 256324
max locked memory (kbytes, -l) 64
max memory size (kbytes, -m) unlimited
open files (-n) 1024
......
```
当并发接入的TCP连接数超过上限时， ， `vi /etc/security/limits.conf` 命令添加如下配置参数：
```powershell
* soft nofile 1000000
* hard nofile 1000000
```
修改之后保存，注销当前用户，重新登录，通过 `ulimit -a` 查看修改的状态是否生效。
**注意**：尽管我们可以将单个进程打开的最大句柄数修改的非常大，但是当句柄数达到一定数量级之后，处理效率将出现明显下降，因此，需要根据服务器的硬件配置和处理能力进行合理设置。
### TCP/IP相关参数
需要重点调优的TCP IP参数如下：
- net.ipv4.tcp_rmem：
### 多网卡队列和软中断
- **TCP缓冲区**
根据推送消息的大小，合理设置以下两个参数，对于海量长连接，通常 32K 是个不错的选择：
- **SO_SNDBUF**：发送缓冲区大小
- **SO_RCVBUF**：接收缓冲区大小
- **软中断**
使用命令 `cat /proc/interrupts` 查看网卡硬件中断的运行情况， ， ， ， ， ： ， ， ， ，
## Netty性能调优
### 设置合理的线程数
**boss线程池优化**
对于Netty服务端， ， ， ， ， ， ， ， ， ， ， ，
**work线程池优化(I/O工作线程池)**
对于I/O工作线程池的优化， （ ） ， ， ， ：
- 通过执行 `ps -ef|grep java` 找到服务端进程pid
- 执行`top -Hp pid`查询该进程下所有线程的运行情况， ，
- 使用`printf'%x\n' pid`将pid转换成16进制格式
- 通过`jstack -f pid`命令获取线程堆栈， ， ， ， ：
**分析**
- 如果连续采集几次进行对比， ， ，
- 如果发现I/O线程的热点停留在读或写操作， ， ：
- 接口API指定：
- 系统参数指定：通过-Dio.netty.eventLoopThreads来指定NioEventLoopGroup线程池（ ）
### 心跳检测优化
心跳检测的目的就是确认当前链路是否可用，对方是否活着并且能够正常接收和发送消息。从技术层面看，要解决链路的可靠性问题，必须周期性地对链路进行有效性检测。目前最流行和通用的做法就是心跳检测。
**海量设备接入的服务端心跳优化策略**
- **要能够及时检测失效的连接，并将其剔除**。防止无效的连接句柄积压，
- **设置合理的心跳周期**。防止心跳定时任务积压， （ ， ） ，
- **使用Nety提供的链路空闲检测机制**。不要自己创建定时任务线程池，加重系统的负担，以及增加潜在的并发安全问题
**心跳检测机制分为三个层面**
- **TCP层的心跳检测**： ，
- **协议层的心跳检测**： ，
- **应用层的心跳检测**：它主要由各业务产品通过约定方式定时给对方发送心跳消息实现
**心跳检测机制分类**
- **Ping-Pong型心跳**： ， ，
- **Ping-Ping型心跳**： ， ，
**心跳检测机制策略**
- **心跳超时**： ，
- **心跳失败**： ， ，
**链路空闲检测机制**
- **读空闲**：
- **写空闲**：
- **读写空闲**：
**案例分析**
由于移动无线网络的特点， ， ， ， ， （ ） ，
在 Netty 中，可以通过在 ChannelPipeline 中增加 IdleStateHandler 的方式实现心跳检测，在构造函数中指定链路空闲时间，然后实现空闲回调接口，实现心跳的发送和检测。拦截链路空闲事件并处理心跳：
```java
public class MyHandler extends ChannelHandlerAdapter {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
// 心跳处理
}
}
}
```
**心跳优化结论**
- 对于百万级的服务器，一般不建议很长的心跳周期和超时时长
- 心跳检测周期通常不要超过60s，
- 建议通过IdleStateHandler实现心跳， ，
- 发生心跳超时或心跳失败时，都需要关闭链路，由客户端发起重连操作，保证链路能够恢复正常
- 链路空闲事件被触发后并没有关闭链路， ， ，
- 链路空闲检测类库主要包括：
### 接收和发送缓冲区调优
对于长链接， ，
**场景**： ， ， ， ， ， ， ， ×
Netty提供的ByteBuf支持容量动态调整， ：
- **FixedRecvByteBufAllocator**： ， ， ，
- **AdaptiveRecvByteBufAllocator**： ， ， ， ， ，
相对于FixedRecvByteBufAllocator， ，
```java
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
```
**注意**：无论是接收缓冲区还是发送缓冲区，缓冲区的大小建议设置为消息的平均大小，不要设置成最大消息的上限，这会导致额外的内存浪费。
### 合理使用内存池
每个NioEventLoop线程处理N个链路。**链路处理流程**：
如果使用内存池， ， ， ， ， ， （ ）
Netty默认不使用内存池， ：
```java
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
```
使用内存池之后，内存的申请和释放必须成对出现，即 retain() 和 release() 要成对出现，否则会导致内存泄露。值得注意的是，如果使用内存池，完成 ByteBuf 的解码工作之后必须显式的调用 ReferenceCountUtil.release(msg) 对接收缓冲区 ByteBuf 进行内存释放，否则它会被认为仍然在使用中，这样会导致内存泄露。
### 防止I/O线程被意外阻塞
通常情况不能在Netty的I/O线程上做执行时间不可控的操作， ， ，
生产环境中， ， ， ， ， ， ， ， ，
常用的log4j虽然支持异步写日志(AsyncAppender)，但当日志队列满之后，它会同步阻塞业务线程，直到日志队列有空闲位置可用。
类似问题具有极强的隐蔽性， ， ， ，
### I/O线程与业务线程分离
- 如果服务端不做复杂业务逻辑操作， ， ， ， ，
- 如果有复杂的业务逻辑操作，
- 对于I/O线程， ， ，
- 对于后端的业务线程池， ， ， ，
### 服务端并发连接数流控
无论服务端的性能优化到多少， ， ，
## 案例分析
### 疑似内存泄漏
**环境**：
**描述**： ， ， ， ，
**分析**： ， ， ， ， ， ： ， ，
**解决**： ，
### 当心CLOSE_WAIT
由于网络不稳定经常会导致客户端断连，如果服务端没有能够及时关闭 socket， ， ， ， ，
close_wait 是被动关闭连接是形成的，根据 TCP 状态机，服务器端收到客户端发送的 FIN， ， ，
导致 close_wait 过多的可能原因如下：
- **程序处理Bug**：导致接收到对方的 fin 之后没有及时关闭 socket， ， ，
- **关闭socket不及时**：例如 I/O 线程被意外阻塞，或者 I/O 线程执行的用户自定义 Task 比例过高，导致 I/O 操作处理不及时，链路不能被及时释放
**解决方案**
- **不要在 Netty 的 I/O 线程（ ） （ ）
- **在I/O线程上执行自定义Task要当心**
- **IdleStateHandler、ReadTimeoutHandler和WriteTimeoutHandler使用要当**
# RabbitMQ
## 模式介绍
在 RabbitMQ 官网上提供了 6 中工作模式：简单模式、工作队列模式、发布/订阅模式、路由模式、主题模式 和 RPC 模式。本篇只对前 5 种工作方式进行介绍。
### 简单模式与工作队列模式
之所以将这两种模式合并在一起介绍，是因为它们工作原理非常简单，由 3 个对象组成：生产者、队列、消费者。
[](http://images.extlight.com/rabbitmq-work-01.png)
生产者负责生产消息，将消息发送到队列中，消费者监听队列，队列有消息就进行消费。
[](http://images.extlight.com/rabbitmq-work-02.png)
当有多个消费者时，消费者平均消费队列中的消息。代码演示：
生产者：
```java
//1.获取连接
Connection connection = ConnectionUtil.getConnection();
//2.创建通道
Channel channel = connection.createChannel();
//3.申明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//4.发送消息
channel.basicPublish("", QUEUE_NAME, null, "hello simple".getBytes());
System.out.println("发送成功");
//5.释放连接
channel.close();
connection.close();
```
消费者：
```java
// 1.获取连接
Connection connection = ConnectionUtil.getConnection();
// 2.创建通道
Channel channel = connection.createChannel();
// 3.申明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 4.监听消息
channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("接收:" + message);
}
});
```
### 发布/订阅、路由与主题模式
这 3 种模式都使用到交换机。生产者不直接与队列交互，而是将消息发送到交换机中，再由交换机将消息放入到已绑定该交换机的队列中给消费者消费。常用的交换机类型有 3 种： ：
[](http://images.extlight.com/rabbitmq-work-03-1.png)
**fanout**： ，
**其中，发布/订阅模式使用的是 fanout 类型的交换机。**
[](http://images.extlight.com/rabbitmq-work-04-1.png)
**direct**：处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。如果一个队列绑定到该交换机上要求路由键 “dog”， ， ，
**其中，路由模式使用的是 direct 类型的交换机。**
[](http://images.extlight.com/rabbitmq-work-05-1.png)
**topic**：将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号 “#” 匹配一个或多个词，符号“\*”匹配不多不少一个词。因此“audit.#” 能够匹配到“audit.irs.corporate”，
**其中，主题模式使用的是 topic 类型的交换机。**代码演示：
生产者：
```java
// 1.获取连接
Connection connection = ConnectionUtil.getConnection();
// 2.创建通道
Channel channel = connection.createChannel();
// 3.申明交换机
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 4.发送消息
for (int i = 0; i < 100 ; i + + ) {
channel.basicPublish(EXCHANGE_NAME, "", null, ("hello ps" + i + "").getBytes());
}
System.out.println("发送成功");
// 5.释放连接
channel.close();
connection.close();
```
多个消费者：
```java
// 1.获取连接
Connection connection = ConnectionUtil.getConnection();
// 2.创建通道
Channel channel = connection.createChannel();
// 3.申明交换机
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 4.队列绑定交换机
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
// 5.消费消息
channel.basicQos(1);
channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("recv1:" + message);
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
```
# Dubbo
Apache Dubbo 是一款高性能、轻量级的开源 Java 服务框架。Apache Dubbo 提供了六大核心能力：**面向接口代理的高性能RPC调用， ， ， ， ，
- **面向接口代理的高性能RPC调用**
提供高性能的基于代理的远程调用能力，服务以接口为粒度，为开发者屏蔽远程调用底层细节。
- **智能负载均衡**
内置多种负载均衡策略，智能感知下游节点健康状况，显著减少调用延迟，提高系统吞吐量。
- **服务自动注册与发现**
支持多种注册中心服务，服务实例上下线实时感知。
- **高度可扩展能力**
遵循微内核+插件的设计原则， ，
- **运行期流量调度**
内置条件、脚本等路由策略，通过配置不同的路由规则，轻松实现灰度发布，同机房优先等功能。
- **可视化的服务治理与运维**
提供丰富服务治理、运维工具：随时查询服务元数据、服务健康状态及调用统计，实时下发路由策略、调整配置参数。
## RPC
Remote Procedure Call Protocol 既 ** 远程过程调用**，一种能让我们**像调用本地服务一样调用远程服务**，可以让调用者对网络通信这些细节**无感知**，比如服务消费方在执行 helloWorldService.sayHello("sowhat") 时，实质上调用的是远端的服务。这种方式其实就是**RPC**， **thrift** 、Google 的**grpc**、Twitter的**finagle**等。
## 框架设计
### Dubbo 简介
Dubbo 是阿里巴巴研发开源工具， ， ， ， ：
- 面向接口代理的高性能RPC调用
- 智能容错和负载均衡
- 服务自动注册和发现
- 高度可扩展能力
- 运行期流量调度
- 可视化的服务治理与运维
**调用过程**：
- 服务提供者 **Provider** 启动然后向 **Registry** 注册自己所能提供的服务。
- 服务消费者 Consumer 向**Registry**订阅所需服务，**Consumer** 解析**Registry**提供的元信息，从服务中通过负载均衡选择 **Provider** 调用。
- 服务提供方 **Provider** 元数据变更的话**Registry**会把变更推送给**Consumer**，以此保证**Consumer**获得最新可用信息。
**注意点**：
- **Provider** 跟 **Consumer** 在内存中记录调用次数跟时间，定时发送统计数据到**Monitor**，发送的时候是**短**连接。
- **Monitor** 跟 **Registry** 是可选的，可直接在配置文件中写好，**Provider** 跟 **Consumer** 进行直连。
- **Monitor** 跟 **Registry** 挂了也没事， **Consumer** 本地缓存了 **Provider** 信息。
- **Consumer** 直接调用 **Provider** 不会经过 **Registry** 。**Provider**、**Consumer**这俩到 **Registry** 之间是长连接。
### Dubbo框架分层
如上图，总的而言 Dubbo 分为三层。
- **Busines**层：由用户自己来提供接口和实现还有一些配置信息。
- **RPC**层： ，
- **Remoting**层：对网络传输协议和数据转换的封装。
如果每一层再细分下去，一共有十层：
1. 接口服务层（ ） ： ，
2. 配置层（ ） ： ，
3. 服务代理层（ ） ： ， ， ，
4. 服务注册层（ ） ： ，
5. 路由层（ ） ： ， ，
6. 监控层（ ） ：
7. 远程调用层（ ） ：
8. 信息交换层（ ） ： ，
9. 网络传输层（ ） ：
10. 数据序列化层（ ） ： ，
他们之间的调用关系直接看下面官网图即可：
## SPI机制
**Dubbo** 采用 ** 微内核设计** + **SPI** 扩展技术来搭好核心框架，同时满足用户定制化需求。这里重点说下**SPI**。
### SPI 含义
**SPI** 全称为 **Service Provider Interface** ，是一种**服务发现机制**。它约定在**ClassPath**路径下的**META-INF/services**文件夹查找文件，自动加载文件里所定义的类。
### Java SPI缺点
- 不能按需加载， ， ， ，
- 获取某个实现类的方式不够灵活，只能通过 Iterator 形式获取，不能根据某个参数来获取对应的实现类。
- 不支持AOP与依赖注入， ，
### Dubbo SPI
JDK自带的不好用Dubbo 就自己实现了一个 SPI， ** 可以通过名字实例化指定的实现类，并且实现了 IOC 、AOP 与 自适应扩展 SPI** 。
```properties
key = com.sowhat.value
```
Dubbo 对配置文件目录的约定，不同于 Java SPI ，
- META-INF/services/ ：该目录下 SPI 配置文件是为了用来兼容 Java SPI
- META-INF/dubbo/ ：该目录存放用户自定义的 SPI 配置文件
- META-INF/dubbo/internal/ ：该目录存 Dubbo 内部使用的 SPI 配置文件
### Dubbo SPI源码追踪
**ExtensionLoader.getExtension** 方法的整个思路是 查找缓存是否存在， ， ， ， ， ：
### injectExtension IOC
查找 set 方法，根据参数找到依赖对象则注入。
### WrapperClass AOP
包装类， ， ， ，
### Activate
Active 有三个属性， ， ，
### Adaptive自适应扩展
**需求**：根据配置来进行 SPI 扩展的加载后不想在启动的时候让扩展被加载，想根据请求时候的参数来动态选择对应的扩展。**实现**： ，
```java
public interface WheelMaker {
Wheel makeWheel(URL url);
}
// WheelMaker 接口的自适应实现类
public class AdaptiveWheelMaker implements WheelMaker {
public Wheel makeWheel(URL url) {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
// 1. 调用 url 的 getXXX 方法获取参数值
String wheelMakerName = url.getParameter("Wheel.maker");
if (wheelMakerName == null) {
throw new IllegalArgumentException("wheelMakerName == null");
}
// 2. 调用 ExtensionLoader 的 getExtensionLoader 获取加载器
// 3. 调用 ExtensionLoader 的 getExtension 根据从url获取的参数作为类名称加载实现类
WheelMaker wheelMaker = ExtensionLoader.getExtensionLoader(WheelMaker.class).getExtension(wheelMakerName);
// 4. 调用实现类的具体方法实现调用。
return wheelMaker.makeWheel(URL url);
}
}
```
查看Adaptive注解源码可知该注解可用在**类**或**方法**上，
#### Adaptive注解在类上
Adaptive 注解在类上时， ， ， ，
#### Adaptive注解在方法上
Adaptive注解在方法上时， ， ， ：
- 加载标注有 @Adaptive 注解的接口，如果不存在，则不支持 Adaptive 机制
- 为目标接口按照一定的模板生成子类代码，并且编译生成的代码，然后通过反射生成该类的对象
- 结合生成的对象实例， ， ， ，
```java
@SPI ("apple")
public interface FruitGranter {
Fruit grant();
@Adaptive
String watering(URL url);
}
---
// 苹果种植者
public class AppleGranter implements FruitGranter {
@Override
public Fruit grant() {
return new Apple();
}
@Override
public String watering(URL url) {
System.out.println("watering apple");
return "watering finished";
}
}
---
// 香蕉种植者
public class BananaGranter implements FruitGranter {
@Override
public Fruit grant() {
return new Banana();
}
@Override
public String watering(URL url) {
System.out.println("watering banana");
return "watering success";
}
}
```
调用方法实现：
```java
public class ExtensionLoaderTest {
@Test
public void testGetExtensionLoader() {
// 首先创建一个模拟用的URL对象
URL url = URL.valueOf("dubbo://192.168.0.1:1412?fruit.granter=apple");
// 通过ExtensionLoader获取一个FruitGranter对象
FruitGranter granter = ExtensionLoader.getExtensionLoader(FruitGranter.class)
.getAdaptiveExtension();
// 使用该FruitGranter调用其"自适应标注的"方法，获取调用结果
String result = granter.watering(url);
System.out.println(result);
}
}
```
通过如上方式生成一个内部类。
## 服务暴露流程
### 服务暴露总览
**Dubbo**框架是以**URL**为总线的模式，运行过程中所有的状态数据信息都可以通过**URL**来获取，比如当前系统采用什么序列化，采用什么通信，采用什么负载均衡等信息，都是通过**URL**的参数来呈现的，所以在框架运行过程中，运行到某个阶段需要相应的数据，都可以通过对应的**Key**从**URL**的参数列表中获取。**URL** 具体的参数如下：
- protocol： ， ：
- username/password：
- host/port：
- path：
- parameters：
```properties
protocol://username:password@host:port/path?k=v
```
服务暴露从代码流程看分为三部分：
- 检查配置，最终组装成 **URL** 。
- 暴露服务到到本地服务跟远程服务。
- 服务注册至注册中心。
服务暴露从对象构建转换看分为两步：
- 将服务封装成**Invoker**。
- 将**Invoker**通过协议转换为**Exporter**。
### 服务暴露源码追踪
- 容器启动， ，
- export 跟 doExport 来进行拼接构建URL， ， ，
- 调用具体 Protocol 将把包装后的 invoker 转换成 exporter，
- 然后启动服务器server， ，
- 通过 RegistryProtocol 将URL注册到注册中心，
## 服务引用流程
Dubbo中一个可执行体就是一个invoker， ，
**服务的引入时机有两种**
- 饿汉式
通过实现 Spring 的 InitializingBean 接口中的 afterPropertiesSet 方法，容器通过调用 **ReferenceBean** 的 afterPropertiesSet 方法时引入服务。
- 懒汉式(默认)
懒汉式是只有当服务被注入到其他类中时启动引入流程。
**服务引用的三种方式**
- **本地引入**：服务暴露时本地暴露，避免网络调用开销
- **直接连接引入远程服务**：不启动注册中心，直接写死远程**Provider**地址 进行直连
- **通过注册中心引入远程服务**：通过注册中心抉择如何进行负载均衡调用远程服务
**服务引用流程**
- 检查配置构建map ， ，
- 想注册中心注册自己， ， ，
- 当有多个 URL 时，先遍历构建出 invoker 然后再由 **StaticDirectory** 封装一下，然后通过 cluster 进行合并，只暴露出一个 invoker 。
- 然后再构建代理，封装 invoker 返回服务引用，之后 Comsumer 调用的就是这个代理类。
**调用方式**
- oneway：
- Async异步调用： ，
- Sync同步调用： ， ，
## 调用整体流程
**调用之前你可能需要考虑这些事**
- consumer 跟 provider 约定好通讯协议， ， ， ，
- 约定序列化模式，大致分为两大类，一种是字符型(XML或json 人可看懂 但传输效率低)，一种是二进制流(数据紧凑，机器友好)。默认使用 hessian2作为序列化协议。
- consumer 调用 provider 时提供对应接口、方法名、参数类型、参数值、版本号。
- provider列表对外提供服务涉及到负载均衡选择一个provider提供服务。
- consumer 跟 provider 定时向monitor 发送信息。
**调用大致流程**
- 客户端发起请求来调用接口，
- ClusterInvoker获得注册中心中服务列表，
- 序列化跟反序列化网络传输数据。通过NettyServer调用网络服务。
- 服务端业务线程池接受解析数据，
- 调用真正的Impl得到结果然后返回。
**调用方式**
- oneway： ，
- sync同步调用： ， ，
- Async 异步调用： ， ，
## 集群容错负载均衡
Dubbo 引入了**Cluster**、**Directory**、**Router**、**LoadBalance**、**Invoker**模块来保证Dubbo系统的稳健性， ：
- 服务发现时会将多个多个远程调用放入**Directory**，然后通过**Cluster**封装成一个**Invoker**，该**invoker**提供容错功能
- 消费者代用的时候从**Directory**中通过负载均衡获得一个可用**invoker**，最后发起调用
- 你可以认为**Dubbo**中的**Cluster**对上面进行了大的封装，自带各种鲁棒性功能
### 集群容错
集群容错是在消费者端通过**Cluster**子类实现的， ， ，
| Cluster | Cluster Invoker | 作用 |
| :--------------- | :---------------------- | :---------------------------- |
| FailoverCluster | FailoverClusterInvoker | 失败自动切换功能，**默认** |
| FailfastCluster | FailfastClusterInvoker | 一次调用，失败异常 |
| FailsafeCluster | FailsafeClusterInvoker | 调用出错则日志记录 |
| FailbackCluster | FailbackClusterInvoker | 失败返空，
| ForkingCluster | ForkingClusterInvoker | 一个任务并发调用，
| BroadcastCluster | BroadcastClusterInvoker | 逐个调用invoker，
| AvailableCluster | AvailableClusterInvoker | 哪个能用就用那个 |
| MergeableCluster | MergeableClusterInvoker | 按组合并返回结果 |
### 智能容错之负载均衡
Dubbo中一般有4种负载均衡策略。
- **RandomLoadBalance**：加权随机，它的算法思想简单。假设有一组服务器 servers = [A, B, C]，对应权重为 weights = [5, 3, 2]， ， ， ，
- **LeastActiveLoadBalance**：最少活跃数负载均衡，选择现在活跃调用数最少的提供者进行调用，活跃的调用数少说明它现在很轻松，而且活跃数都是从 0 加起来的，来一个请求活跃数+1， ，
- **RoundRobinLoadBalance**：加权轮询负载均衡，比如现在有两台服务器 A、B， ， ， ，
- **ConsistentHashLoadBalance**：一致性 Hash 负载均衡，将服务器的 IP 等信息生成一个 hash 值，
### 智能容错之服务目录
关于 服务目录Directory 你可以理解为是相同服务Invoker的集合，
- 从注册中心获得invoker列表
- 监控着注册中心invoker的变化，
- 刷新invokers列表到服务目录
### 智能容错之服务路由
服务路由其实就是路由规则，它规定了服务消费者可以调用哪些服务提供者。条件路由规则由两个条件组成，分别用于对服务消费者和提供者进行匹配。比如有这样一条规则：
```properties
host = 10.20.153.14 => host = 10.20.153.12
```
该条规则表示 IP 为 10.20.153.14 的服务消费者只可调用 IP 为 10.20.153.12 机器上的服务，不可调用其他机器上的服务。条件路由规则的格式如下：
```properties
[服务消费者匹配条件] => [服务提供者匹配条件]
```
如果服务消费者匹配条件为空，表示不对服务消费者进行限制。如果服务提供者匹配条件为空，表示对某些服务消费者禁用服务。
## 设计RPC
一个RPC框架大致需要以下功能：
1. 服务的注册与发现
2. 用动态代理
3. 负载均衡（ ）
4. 通信协议
5. 序列化与反序列化
6. 网络通信(Netty)
7. Monitor
# Nacos
## 基本架构
### 服务 (Service)
服务是指一个或一组软件功能（ ） ， （ ）
### 服务注册中心 (Service Registry)
服务注册中心，它是服务，其实例及元数据的数据库。服务实例在启动时注册到服务注册表，并在关闭时注销。服务和路由器的客户端查询服务注册表以查找服务的可用实例。服务注册中心可能会调用服务实例的健康检查 API 来验证它是否能够处理请求。
### 服务元数据 (Service Metadata)
服务元数据是指包括服务端点(endpoints)、服务标签、服务版本号、服务实例权重、路由规则、安全策略等描述服务的数据。
### 服务提供方 (Service Provider)
是指提供可复用和可调用服务的应用方。
### 服务消费方 (Service Consumer)
是指会发起对某个服务调用的应用方。
### 配置 (Configuration)
在系统开发过程中通常会将一些需要变更的参数、变量等从代码中分离出来独立管理，以独立的配置文件的形式存在。目的是让静态的系统工件或者交付物（如 WAR，
### 配置管理 (Configuration Management)
在数据中心中，系统中所有配置的编辑、存储、分发、变更管理、历史版本管理、变更审计等所有与配置相关的活动统称为配置管理。
### 名字服务 (Naming Service)
提供分布式系统中所有对象(Object)、实体(Entity)的“名字”到关联的元数据之间的映射管理服务，例如 ServiceName -> Endpoints Info, Distributed Lock Name -> Lock Owner/Status Info, DNS Domain Name -> IP List, 服务发现和 DNS 就是名字服务的2大场景。
### 配置服务 (Configuration Service)
在服务或者应用运行过程中，提供动态配置或者元数据以及配置管理的服务提供者。
## 逻辑架构
- 服务管理： ， ， ，
- 配置管理： ， ， ， ， ，
- 元数据管理：
- 插件机制： ，
- 事件机制： ，
- 日志模块：管理日志分类，日志级别，日志可移植性（尤其避免冲突），日志格式，异常码+帮助文档
- 回调机制： ，
- 寻址模式： ， ， ，
- 推送通道：
- 容量管理：管理每个租户，分组下的容量，防止存储被写爆，影响服务可用性
- 流量管理：按照租户，分组等多个维度对请求频率，长链接个数，报文大小，请求流控进行控制
- 缓存机制： ， ，
- 启动模式： ， ， ， ， ，
- 一致性协议：解决不同数据，不同一致性要求情况下，不同一致性机制
- 存储模块：解决数据持久化、非持久化存储，解决数据分片问题
- Nameserver： ，
- CMDB： ， ， ， ，
- Metrics： ，
- Trace： ， ， ， ，
- 接入管理：相当于阿里云开通服务，分配身份、容量、权限过程
- 用户管理： ， ，
- 权限管理：解决身份识别，访问控制，角色管理等问题
- 审计系统：扩展接口方便与不同公司审计系统打通
- 通知系统： ， ， ，
- OpenAPI： ， ，
- Console： ，
- SDK：
- Agent： ，
- CLI： ，
## 功能特性
Nacos 的关键特性包括：
- **服务发现和服务健康监测**
- **动态配置服务**
- **动态 DNS 服务**
- **服务及其元数据管理**
### 数据模型
Nacos 数据模型 Key 由三元组唯一确定, Namespace默认是空串， （ ） ，
### Nacos-SDK类视图
### 配置领域模型
围绕配置，主要有两个关联的实体，一个是配置变更历史，一个是服务标签（用于打标分类，方便索引），由 ID 关联。
## 安装部署
### 下载源码或者安装包
**从 Github 上下载源码方式**
```bash
git clone https://github.com/alibaba/nacos.git
cd nacos/
mvn -Prelease-nacos -Dmaven.test.skip=true clean install -U
ls -al distribution/target/
// change the $version to your actual path
cd distribution/target/nacos-server-$version/nacos/bin
```
**下载编译后压缩包方式**
您可以从 [最新稳定版本 ](https://github.com/alibaba/nacos/releases ) 下载 `nacos-server-$version.zip` 包。
```bash
unzip nacos-server-$version.zip 或者 tar -xvf nacos-server-$version.tar.gz
cd nacos/bin
```
### 启动服务器
**Linux/Unix/Mac**
启动命令(standalone代表着单机模式运行，
```shell
sh startup.sh -m standalone
```
如果您使用的是ubuntu系统， ， ：
```shell
bash startup.sh -m standalone
```
**Windows**
启动命令(standalone代表着单机模式运行，
```shell
startup.cmd -m standalone
```
### 服务测试
**服务注册**
```shell
curl -X POST 'http://127.0.0.1:8848/nacos/v1/ns/instance?serviceName=nacos.naming.serviceName& ip=20.18.7.10& port=8080'
```
**服务发现**
```shell
curl -X GET 'http://127.0.0.1:8848/nacos/v1/ns/instance/list?serviceName=nacos.naming.serviceName'
```
**发布配置**
```shell
curl -X POST "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=nacos.cfg.dataId& group=test& content=HelloWorld"
```
**获取配置**
```shell
curl -X GET "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=nacos.cfg.dataId& group=test"
```
### 关闭服务器
**Linux/Unix/Mac**
```shell
sh shutdown.sh
```
**Windows**
```shell
shutdown.cmd
```
或者双击 `shutdown.cmd` 运行文件。
## 开源案例
### Spring Boot
#### 启动配置管理
启动了 Nacos server 后，您就可以参考以下示例代码，为您的 Spring Boot 应用启动 Nacos 配置管理服务了。
**第一步**：添加依赖
```xml
< dependency >
< groupId > com.alibaba.boot< / groupId >
< artifactId > nacos-config-spring-boot-starter< / artifactId >
< version > ${latest.version}< / version >
< / dependency >
```
**注意**：版本 [0.2.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.boot/nacos-config-spring-boot-starter ) 对应的是 Spring Boot 2.x 版本，版本 [0.1.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.boot/nacos-config-spring-boot-starter ) 对应的是 Spring Boot 1.x 版本。
**第二步**：在 `application.properties` 中配置 Nacos server 的地址：
```properties
nacos.config.server-addr=127.0.0.1:8848
```
**第三步**：使用 `@NacosPropertySource` 加载 `dataId` 为 `example` 的配置源，并开启自动更新：
```java
@SpringBootApplication
@NacosPropertySource (dataId = "example", autoRefreshed = true)
public class NacosConfigApplication {
public static void main(String[] args) {
SpringApplication.run(NacosConfigApplication.class, args);
}
}
```
**第四步**：通过 Nacos 的 `@NacosValue` 注解设置属性值。
```java
@Controller
@RequestMapping ("config")
public class ConfigController {
@NacosValue (value = "${useLocalCache:false}", autoRefreshed = true)
private boolean useLocalCache;
@RequestMapping (value = "/get", method = GET)
@ResponseBody
public boolean get() {
return useLocalCache;
}
}
```
**第五步**：启动 `NacosConfigApplication` ，调用 `curl http://localhost:8080/config/get` ，返回内容是 `false` 。
**第六步**：通过调用 [Nacos Open API ](https://nacos.io/zh-cn/docs/open-api.html ) 向 Nacos server 发布配置：
```shell
curl -X POST "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=example& group=DEFAULT_GROUP& content=useLocalCache=true"
```
**第七步**：再次访问 `http://localhost:8080/config/get` ，此时返回内容为`true`，说明程序中的`useLocalCache`值已经被动态更新了。
#### 启动服务发现
本节演示如何在您的 Spring Boot 项目中启动 Nacos 的服务发现功能。
**第一步**：添加依赖
```xml
< dependency >
< groupId > com.alibaba.boot< / groupId >
< artifactId > nacos-discovery-spring-boot-starter< / artifactId >
< version > ${latest.version}< / version >
< / dependency >
```
**注意**：版本 [0.2.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.boot/nacos-discovery-spring-boot-starter ) 对应的是 Spring Boot 2.x 版本，版本 [0.1.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.boot/nacos-discovery-spring-boot-starter ) 对应的是 Spring Boot 1.x 版本。
**第二步**：在 `application.properties` 中配置 Nacos server 的地址：
```properties
nacos.discovery.server-addr=127.0.0.1:8848
```
**第三步**：使用 `@NacosInjected` 注入 Nacos 的 `NamingService` 实例：
```java
@Controller
@RequestMapping ("discovery")
public class DiscoveryController {
@NacosInjected
private NamingService namingService;
@RequestMapping (value = "/get", method = GET)
@ResponseBody
public List< Instance > get(@RequestParam String serviceName) throws NacosException {
return namingService.getAllInstances(serviceName);
}
}
@SpringBootApplication
public class NacosDiscoveryApplication {
public static void main(String[] args) {
SpringApplication.run(NacosDiscoveryApplication.class, args);
}
}
```
**第四步**：启动 `NacosDiscoveryApplication` ，调用 `curl http://localhost:8080/discovery/get?serviceName=example` ，此时返回为空 JSON 数组`[]`。
**第五步**：通过调用 [Nacos Open API ](https://nacos.io/zh-cn/docs/open-api.html ) 向 Nacos server 注册一个名称为 `example` 服务
```shell
curl -X PUT 'http://127.0.0.1:8848/nacos/v1/ns/instance?serviceName=example& ip=127.0.0.1& port=8080'
```
**第六步**：再次访问 `curl http://localhost:8080/discovery/get?serviceName=example` ，此时返回内容为：
```json
[
{
"instanceId": "127.0.0.1-8080-DEFAULT-example",
"ip": "127.0.0.1",
"port": 8080,
"weight": 1.0,
"healthy": true,
"cluster": {
"serviceName": null,
"name": "",
"healthChecker": {
"type": "TCP"
},
"defaultPort": 80,
"defaultCheckPort": 80,
"useIPPort4Check": true,
"metadata": {}
},
"service": null,
"metadata": {}
}
]
```
### Spring Cloud
#### 启动配置管理
启动了 Nacos server 后，您就可以参考以下示例代码，为您的 Spring Cloud 应用启动 Nacos 配置管理服务了。
**第一步**：添加依赖
```xml
< dependency >
< groupId > com.alibaba.cloud< / groupId >
< artifactId > spring-cloud-starter-alibaba-nacos-config< / artifactId >
< version > ${latest.version}< / version >
< / dependency >
```
**注意**：版本 [2.1.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.cloud/spring-cloud-starter-alibaba-nacos-config ) 对应的是 Spring Boot 2.1.x 版本。版本 [2.0.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.cloud/spring-cloud-starter-alibaba-nacos-config ) 对应的是 Spring Boot 2.0.x 版本，版本 [1.5.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.cloud/spring-cloud-starter-alibaba-nacos-config ) 对应的是 Spring Boot 1.5.x 版本。
**第二步**：在 `bootstrap.properties` 中配置 Nacos server 的地址和应用名
```properties
spring.cloud.nacos.config.server-addr=127.0.0.1:8848
spring.application.name=example
```
说明：之所以需要配置 `spring.application.name` ，是因为它是构成 Nacos 配置管理 `dataId` 字段的一部分。
**第三步**：在 Nacos Spring Cloud 中，`dataId` 的完整格式如下：
```properties
${prefix}-${spring.profiles.active}.${file-extension}
```
- `prefix` 默认为 `spring.application.name` 的值，也可以通过配置项 `spring.cloud.nacos.config.prefix` 来配置。
- `spring.profiles.active` 即为当前环境对应的 profile， Spring Boot文档 ](https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-profiles.html#boot-features-profiles )。 ** 注意：当 `spring.profiles.active` 为空时，对应的连接符 `-` 也将不存在， `${prefix}.${file-extension}` **
- `file-exetension` 为配置内容的数据格式，可以通过配置项 `spring.cloud.nacos.config.file-extension` 来配置。目前只支持 `properties` 和 `yaml` 类型。
**第四步**：通过 Spring Cloud 原生注解 `@RefreshScope` 实现配置自动更新：
```java
@RestController
@RequestMapping ("/config")
@RefreshScope
public class ConfigController {
@Value ("${useLocalCache:false}")
private boolean useLocalCache;
@RequestMapping ("/get")
public boolean get() {
return useLocalCache;
}
}
```
**第五步**：首先通过调用 [Nacos Open API ](https://nacos.io/zh-cn/docs/open-api.html ) 向 Nacos Server 发布配置：
```shell
curl -X POST "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=example.properties& group=DEFAULT_GROUP& content=useLocalCache=true"
```
**第六步**：运行 `NacosConfigApplication` ，调用 `curl http://localhost:8080/config/get` ，返回内容是 `true` 。
**第七步**：再次调用 [Nacos Open API ](https://nacos.io/zh-cn/docs/open-api.html ) 向 Nacos server 发布配置：
```shell
curl -X POST "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=example.properties& group=DEFAULT_GROUP& content=useLocalCache=false"
```
**第八步**：再次访问 `http://localhost:8080/config/get` ，此时返回内容为`false`，说明程序中的`useLocalCache`值已经被动态更新了。
#### 启动服务发现
本节通过实现一个简单的 `echo service` 演示如何在您的 Spring Cloud 项目中启用 Nacos 的服务发现功能，如下图示:
**第一步**：添加依赖：
```xml
< dependency >
< groupId > com.alibaba.cloud< / groupId >
< artifactId > spring-cloud-starter-alibaba-nacos-discovery< / artifactId >
< version > ${latest.version}< / version >
< / dependency >
```
**注意**：版本 [2.1.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.cloud/spring-cloud-starter-alibaba-nacos-discovery ) 对应的是 Spring Boot 2.1.x 版本。版本 [2.0.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.cloud/spring-cloud-starter-alibaba-nacos-discovery ) 对应的是 Spring Boot 2.0.x 版本，版本 [1.5.x.RELEASE ](https://mvnrepository.com/artifact/com.alibaba.cloud/spring-cloud-starter-alibaba-nacos-discovery ) 对应的是 Spring Boot 1.5.x 版本。
**第二步**：配置服务提供者，从而服务提供者可以通过 Nacos 的服务注册发现功能将其服务注册到 Nacos server 上。
i. 在 `application.properties` 中配置 Nacos server 的地址：
```properties
server.port=8070
spring.application.name=service-provider
spring.cloud.nacos.discovery.server-addr=127.0.0.1:8848
```
ii. 通过 Spring Cloud 原生注解 `@EnableDiscoveryClient` 开启服务注册发现功能：
```java
@SpringBootApplication
@EnableDiscoveryClient
public class NacosProviderApplication {
public static void main(String[] args) {
SpringApplication.run(NacosProviderApplication.class, args);
}
@RestController
class EchoController {
@RequestMapping (value = "/echo/{string}", method = RequestMethod.GET)
public String echo(@PathVariable String string) {
return "Hello Nacos Discovery " + string;
}
}
}
```
**第三步**：配置服务消费者，从而服务消费者可通过 Nacos 的服务注册发现功能从 Nacos server 上获取到它要调用的服务。
i. 在 `application.properties` 中配置 Nacos server 的地址：
```properties
server.port=8080
spring.application.name=service-consumer
spring.cloud.nacos.discovery.server-addr=127.0.0.1:8848
```
ii. 通过 Spring Cloud 原生注解 `@EnableDiscoveryClient` 开启服务注册发现功能。给 [RestTemplate ](https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-resttemplate.html ) 实例添加 `@LoadBalanced` 注解，开启 `@LoadBalanced` 与 [Ribbon ](https://cloud.spring.io/spring-cloud-netflix/multi/multi_spring-cloud-ribbon.html ) 的集成：
```java
@SpringBootApplication
@EnableDiscoveryClient
public class NacosConsumerApplication {
@LoadBalanced
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
public static void main(String[] args) {
SpringApplication.run(NacosConsumerApplication.class, args);
}
@RestController
public class TestController {
private final RestTemplate restTemplate;
@Autowired
public TestController(RestTemplate restTemplate) {this.restTemplate = restTemplate;}
@RequestMapping (value = "/echo/{str}", method = RequestMethod.GET)
public String echo(@PathVariable String str) {
return restTemplate.getForObject("http://service-provider/echo/" + str, String.class);
}
}
}
```
**第四步**：启动 `ProviderApplication` 和 `ConsumerApplication` ，调用 `http://localhost:8080/echo/2018` ，返回内容为 `Hello Nacos Discovery 2018` 。
# Sentinel
Sentinel（ ） ，
- **丰富的应用场景**：
- **完备的实时监控**：
- **广泛的开源生态**：
- **完善的 SPI 扩展点**：
## 主要特性
Sentinel 的主要特性：
## 开源生态
Sentinel 的开源生态：
Sentinel 分为两个部分:
- 核心库（
- 控制台（ ）
## Quick Start
### 手动接入Sentinel以及控制台
下面例子将展示应用如何三步接入 Sentinel。同时，
**STEP 1. 在应用中引入Sentinel Jar包**
如果应用使用 pom 工程，则在 `pom.xml` 文件中加入以下代码即可：
```xml
< dependency >
< groupId > com.alibaba.csp< / groupId >
< artifactId > sentinel-core< / artifactId >
< version > 1.8.1< / version >
< / dependency >
```
注意: Sentinel仅支持JDK 1.8或者以上版本。如果未使用依赖管理工具，请到 [Maven Center Repository ](https://mvnrepository.com/artifact/com.alibaba.csp/sentinel-core ) 直接下载JAR 包。
**STEP 2. 定义资源**
接下来，我们把需要控制流量的代码用 Sentinel API `SphU.entry("HelloWorld")` 和 `entry.exit()` 包围起来即可。在下面的例子中，我们将 `System.out.println("hello world");` 这端代码作为资源，用 API 包围起来（埋点）。参考代码如下:
```java
public static void main(String[] args) {
initFlowRules();
while (true) {
Entry entry = null;
try {
entry = SphU.entry("HelloWorld");
/*您的业务逻辑 - 开始*/
System.out.println("hello world");
/*您的业务逻辑 - 结束*/
} catch (BlockException e1) {
/*流控逻辑处理 - 开始*/
System.out.println("block!");
/*流控逻辑处理 - 结束*/
} finally {
if (entry != null) {
entry.exit();
}
}
}
}
```
完成以上两步后，代码端的改造就完成了。当然，我们也提供了 [注解支持模块 ](https://github.com/alibaba/Sentinel/wiki/注解支持 )，可以以低侵入性的方式定义资源。
**STEP 3. 定义规则**
接下来，通过规则来指定允许该资源通过的请求次数，如下面的代码定义了资源 `HelloWorld` 每秒最多只能通过 20 个请求。
```java
private static void initFlowRules(){
List< FlowRule > rules = new ArrayList< >();
FlowRule rule = new FlowRule();
rule.setResource("HelloWorld");
rule.setGrade(RuleConstant.FLOW_GRADE_QPS);
// Set limit QPS to 20.
rule.setCount(20);
rules.add(rule);
FlowRuleManager.loadRules(rules);
}
```
完成上面 3 步， 使用文档 ](https://github.com/alibaba/Sentinel/wiki/如何使用 )。
**STEP 4. 检查效果**
Demo 运行之后，我们可以在日志 `~/logs/csp/${appName}-metrics.log.xxx` 里看到下面的输出:
```shell
|--timestamp-|------date time----|-resource-|p |block|s |e|rt
1529998904000|2018-06-26 15:41:44|HelloWorld|20|0 |20|0|0
1529998905000|2018-06-26 15:41:45|HelloWorld|20|5579 |20|0|728
1529998906000|2018-06-26 15:41:46|HelloWorld|20|15698|20|0|0
1529998907000|2018-06-26 15:41:47|HelloWorld|20|19262|20|0|0
1529998908000|2018-06-26 15:41:48|HelloWorld|20|19502|20|0|0
1529998909000|2018-06-26 15:41:49|HelloWorld|20|18386|20|0|0
```
其中 `p` 代表通过的请求, `block` 代表被阻止的请求, `s` 代表成功执行完成的请求个数, `e` 代表用户自定义的异常, `rt` 代表平均响应时长。可以看到，这个程序每秒稳定输出 "hello world" 20 次，和规则中预先设定的阈值是一样的。
**STEP 5. 启动 Sentinel 控制台**
您可以参考 [Sentinel 控制台文档 ](https://github.com/alibaba/Sentinel/wiki/控制台 ) 启动控制台，可以实时监控各个资源的运行情况，并且可以实时地修改限流规则。
# Influxdb
InfluxDB 是用Go语言编写的一个开源分布式时序、事件和指标数据库，
## 重要特性
- **极简架构：**单机版的InfluxDB只需要安装一个binary， ，
- **极强的写入能力：** 底层采用自研的TSM存储引擎， ，
- **高效查询：**对Tags会进行索引，
- **InfluxQL**： ， ，
- **Continuous Queries**: 通过CQ能够支持auto-rollup和pre-aggregation，
## 存储引擎
InfluxDB 采用自研的TSM (Time-Structured Merge Tree) 作为存储引擎， 其核心思想是通过牺牲掉一些功能来对性能达到极致优化， ， ，
**时序数据库的需求：**
- 数十亿个单独的数据点
- 高写入吞吐量
- 高读取吞吐量
- 大型删除（数据过期）
- 主要是插入/追加工作负载，很少更新
### LSM
**LSM 的局限性**
在官方文档上有写， 为了解决高写入吞吐量的问题， Influxdb 一开始选择了LevelDB 作为其存储引擎。 然而， ，
**LSM （日志结构合并树）为 LevelDB的引擎原理**
- levelDB 不支持热备份。 对数据库进行安全备份必须关闭后才能复制。LevelDB的RocksDB和HyperLevelDB变体可以解决此问题
- 时序数据库需要提供一种自动管理数据保存的方式。 即删除过期数据， 而在levelDB 中，删除的代价过高。（通过添加墓碑的方式， 段结构合并的时候才会真正物理性的删除）
### TSM
按不同的时间范围划分为不同的分区（ ） ， ， ， ， ，
- 在最开始的时候， influxdb 采用的方案每个shard都是一个独立的数据库实例， ， ， ， ，
- 由于遇到大量的客户反馈文件句柄过多的问题，
- 之后Influxdb 最终决定仿照LSM 的思想自研TSM ， ， ，
## 系统架构
- **DataBase：
- **Retention Policy（ ） ：
- **Shard Group： ， ， ，
# Spring
Spring是一个轻量级的IoC和AOP容器框架。是为Java应用程序提供基础性服务的一套框架， ，
**Spring的优点？
- spring属于低侵入式设计，
- spring的DI机制将对象之间的依赖关系交由框架处理，
- Spring提供了AOP技术， ， ，
- spring对于主流的应用框架提供了集成支持
** 使用Spring框架的好处是什么？
- **轻量：**Spring 是轻量的，
- **控制反转：**Spring通过控制反转实现了松散耦合， ，
- **面向切面的编程(AOP)： ，
- **容器：**Spring 包含并管理应用中对象的生命周期和配置
- **MVC框架**： ，
- **事务管理：**Spring 提供一个持续的事务管理接口， （ ）
- **异常处理：**Spring 提供方便的API把具体技术相关的异常（ ， ）
## 相关概念
### Spring
Spring是一个开源容器框架， ， ， ， ， ，
### SpringMVC
Spring MVC属于SpringFrameWork的后续产品， ， （ ， ， ， ，
SpringMVC是属于SpringWeb里面的一个功能模块（ ）
### SpringBoot
Springboot是一个微服务框架， ， ， ， ，
Spring Boot基本上是Spring框架的扩展， ， ， ：
- 创建独立的spring应用
- 嵌入Tomcat, JettyUndertow 而且不需要部署他们
- 提供的“starters” poms来简化Maven配置
- 尽可能自动配置spring应用
- 提供生产指标,健壮检查和外部化配置
- 绝对没有代码生成和XML配置要求
## Spring原理
### 核心组件
### Spring常用模块
主要包括以下七个模块：
- **Spring Context**： ， （ ）
- **Spring Core**： ， ，
- **Spring AOP**：
- **Spring Web**： ， ， ， ，
- **Spring MVC**： ，
- **Spring DAO**： ， ，
- **Spring ORM**：
### Spring主要包
### Spring常用注解
## IoC
IOC就是控制反转， ， ， ， ， ，
最直观的表达就是， ， ， ， ， ，
Spring的IOC有三种注入方式 ：
## AOP
AOP， ， ， ， ， ， （ ） ， ， ， ， ；
### 静态代理
AspectJ是静态代理， ， ， ，
### 动态代理
Spring AOP使用的动态代理， ， ， ， ， ， ：
- **JDK动态代理**： ， ， ， ， （ ， ） ， ，
**InvocationHandler的invoke(Object proxy,Method method,Object[] args)** ：
- **proxy**：是最终生成的代理对象
- **method**：是被代理目标实例的某个具体方法;
- **args**：是被代理目标实例某个方法的具体入参, 在方法反射调用时使用
- **CGLIB动态代理**： ， （ ） ， ， ， ， ， ，
**静态代理与动态代理区别？**
生成AOP代理对象的时机不同， ， ，
IoC让相互协作的组件保持松散的耦合，
### 应用场景
AOP 主要应用场景有：
- Authentication 权限
- Caching 缓存
- Context passing 内容传递
- Error handling 错误处理
- Lazy loading 懒加载
- Debugging 调试
- logging, tracing, profiling and monitoring 记录跟踪 优化 校准
- Performance optimization 性能优化
- Persistence 持久化
- Resource pooling 资源池
- Synchronization 同步
- Transactions 事务
## 过滤器（ ）
主要作用是过滤字符编码、做一些业务逻辑判断， ， ， ， ， ， ：
- 用户授权Filter： ，
- 日志Filter：
- 解码Filter：
## 拦截器（ ）
主要作用是拦截用户请求， ， ，
Interceptor分两种情况， ， ， ， ，
**Filter和Interceptor的区别**
- Filter是基于函数回调（ ） ， （ ）
- Filter依赖于Servlet容器，
- Filter对几乎所有的请求起作用，
- Interceptor可以访问Action的上下文， ，
- 在action的生命周期里， ，
- Filter在过滤是只能对request和response进行操作，
### HandlerInterceptor
HandlerInterceptor是springMVC项目中的拦截器， ，
实现一个HandlerInterceptor拦截器可以直接实现HandlerInterceptor接口，
这两种方法殊途同归， ，
### MethodInterceptor
MethodInterceptor是AOP项目中的拦截器， ， ， ，
## Spring流程
### Spring容器启动流程
- 初始化Spring容器，
① 实例化BeanFactory【DefaultListableBeanFactory】工厂，
② 实例化BeanDefinitionReader注解配置读取器， （ ） ， （ ， ， ，
③ 实例化ClassPathBeanDefinitionScanner路径扫描器，
- 将配置类的BeanDefinition注册到容器中
- 调用refresh()方法刷新容器
① prepareRefresh()刷新前的预处理
② obtainFreshBeanFactory()：
③ prepareBeanFactory(beanFactory)： ，
④ postProcessBeanFactory(beanFactory)： ，
⑤ invokeBeanFactoryPostProcessors(beanFactory)： ，
⑥ registerBeanPostProcessors(beanFactory)： ， ，
⑦ initMessageSource()： ， ，
⑧ initApplicationEventMulticaster()：初始化事件派发器，在注册监听器时会用到
⑨ onRefresh()：留给子容器、子类重写这个方法，在容器刷新的时候可以自定义逻辑
⑩ registerListeners()： ，
⑪ finishBeanFactoryInitialization(beanFactory)： ， ，
⑫ finishRefresh()：
### SpringMVC流程
- 用户发送请求至前端控制器DispatcherServlet
- DispatcherServlet收到请求后， ，
- 处理器映射器根据请求url找到具体的处理器Handler， ，
- DispatcherServlet 调用 HandlerAdapter处理器适配器，
- HandlerAdapter 经过适配调用 具体处理器进行处理业务逻辑
- Handler执行完成返回ModelAndView
- HandlerAdapter将Handler执行结果ModelAndView返回给DispatcherServlet
- DispatcherServlet将ModelAndView传给ViewResolver视图解析器进行解析
- ViewResolver解析后返回具体View
- DispatcherServlet对View进行渲染视图（ ）
- DispatcherServlet响应用户
- 前端控制器 DispatcherServlet： ， ，
- 处理器映射器 HandlerMapping：
- 处理器适配器 HandlerAdapter：
- 处理器 Handler： ，
- 视图解析器 ViewResolver： ， （ ）
- 视图View： ， ， ， ，
### 非拦截器Http请求流程
用户的普通Http请求执行顺序：
### 拦截器Http请求流程
过滤器、拦截器添加后的执行顺序：
# SpringBoot
## 启动流程扩展点
### ApplicationContextInitializer
整个spring容器在刷新之前初始化`ConfigurableApplicationContext`的回调接口，简单来说，就是在容器刷新之前调用此类的`initialize`方法。用户可以在整个spring容器还没被初始化之前做一些事情。
```java
public class TestApplicationContextInitializer implements ApplicationContextInitializer {
@Override
public void initialize(ConfigurableApplicationContext applicationContext) {
System.out.println("[ApplicationContextInitializer]");
}
}
```
因为这时候spring容器还没被初始化， ， ：
- 在启动类中用`springApplication.addInitializers(new TestApplicationContextInitializer())`语句加入
- 配置文件配置`context.initializer.classes=com.example.demo.TestApplicationContextInitializer`
- Spring SPI扩展，
### BeanDefinitionRegistryPostProcessor
在读取项目中的`beanDefinition`之后执行，提供一个补充的扩展点。使用场景是你可以在这里动态注册自己的`beanDefinition`，
```java
public class TestBeanDefinitionRegistryPostProcessor implements BeanDefinitionRegistryPostProcessor {
@Override
public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
System.out.println("[BeanDefinitionRegistryPostProcessor] postProcessBeanDefinitionRegistry");
}
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
System.out.println("[BeanDefinitionRegistryPostProcessor] postProcessBeanFactory");
}
}
```
### BeanFactoryPostProcessor
是`beanFactory`的扩展接口， ， ， ，
```java
public class TestBeanFactoryPostProcessor implements BeanFactoryPostProcessor {
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
System.out.println("[BeanFactoryPostProcessor]");
}
}
```
### InstantiationAwareBeanPostProcessor
该接口继承了`BeanPostProcess`接口，区别如下：
**`BeanPostProcess`接口只在bean的初始化阶段进行扩展（ ） ， ，
该类主要的扩展点有以下5个方法， ：
- `postProcessBeforeInstantiation` ： ，
- `postProcessAfterInstantiation` ： ，
- `postProcessPropertyValues` ： ， ，
- `postProcessBeforeInitialization` ： ，
- `postProcessAfterInitialization` ： ，
使用场景：这个扩展点非常有用 ， ， ，
```java
public class TestInstantiationAwareBeanPostProcessor implements InstantiationAwareBeanPostProcessor {
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
System.out.println("[TestInstantiationAwareBeanPostProcessor] before initialization " + beanName);
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
System.out.println("[TestInstantiationAwareBeanPostProcessor] after initialization " + beanName);
return bean;
}
@Override
public Object postProcessBeforeInstantiation(Class< ?> beanClass, String beanName) throws BeansException {
System.out.println("[TestInstantiationAwareBeanPostProcessor] before instantiation " + beanName);
return null;
}
@Override
public boolean postProcessAfterInstantiation(Object bean, String beanName) throws BeansException {
System.out.println("[TestInstantiationAwareBeanPostProcessor] after instantiation " + beanName);
return true;
}
@Override
public PropertyValues postProcessPropertyValues(PropertyValues pvs, PropertyDescriptor[] pds, Object bean, String beanName) throws BeansException {
System.out.println("[TestInstantiationAwareBeanPostProcessor] postProcessPropertyValues " + beanName);
return pvs;
}
}
```
### SmartInstantiationAwareBeanPostProcessor
该扩展接口有3个触发点方法：
- `predictBeanType` ：
该触发点发生在`postProcessBeforeInstantiation`之前(在图上并没有标明，因为一般不太需要扩展这个点)， ， ， ；
- `determineCandidateConstructors` ：
该触发点发生在`postProcessBeforeInstantiation`之后， ， ，
- `getEarlyBeanReference` ：
该触发点发生在`postProcessAfterInstantiation`之后， ， ， ， ，
```java
public class TestSmartInstantiationAwareBeanPostProcessor implements SmartInstantiationAwareBeanPostProcessor {
@Override
public Class<?> predictBeanType(Class<?> beanClass, String beanName) throws BeansException {
System.out.println("[TestSmartInstantiationAwareBeanPostProcessor] predictBeanType " + beanName);
return beanClass;
}
@Override
public Constructor<?>[] determineCandidateConstructors(Class<?> beanClass, String beanName) throws BeansException {
System.out.println("[TestSmartInstantiationAwareBeanPostProcessor] determineCandidateConstructors " + beanName);
return null;
}
@Override
public Object getEarlyBeanReference(Object bean, String beanName) throws BeansException {
System.out.println("[TestSmartInstantiationAwareBeanPostProcessor] getEarlyBeanReference " + beanName);
return bean;
}
}
```
### BeanFactoryAware
这个类只有一个触发点， ， ，
使用场景为， ， ， `BeanFactory` ， ，
```java
public class TestBeanFactoryAware implements BeanFactoryAware {
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
System.out.println("[TestBeanFactoryAware] " + beanFactory.getBean(TestBeanFactoryAware.class).getClass().getSimpleName());
}
}
```
### ApplicationContextAwareProcessor
该类本身并没有扩展点， ， ， ， ， ， ， ：
- `EnvironmentAware` ：
用于获取`EnviromentAware`的一个扩展类，这个变量非常有用， 可以获得系统内的所有参数。当然个人认为这个Aware没必要去扩展，
- `EmbeddedValueResolverAware` ：
用于获取`StringValueResolver`的一个扩展类， `StringValueResolver` 用于获取基于`String`类型的properties的变量， ， ，
- `ResourceLoaderAware` ：
用于获取`ResourceLoader`的一个扩展类，`ResourceLoader`可以用于获取classpath内所有的资源对象，
- `ApplicationEventPublisherAware` ：
用于获取`ApplicationEventPublisher`的一个扩展类，`ApplicationEventPublisher`可以用来发布事件，结合`ApplicationListener`来共同使用，下文在介绍`ApplicationListener`时会详细提到。这个对象也可以通过spring注入的方式来获得。
- `MessageSourceAware` ：
用于获取`MessageSource`的一个扩展类，`MessageSource`主要用来做国际化。
- `ApplicationContextAware` ：
用来获取`ApplicationContext`的一个扩展类，`ApplicationContext`应该是很多人非常熟悉的一个类了， ， ， ， ， ，
```java
class ApplicationContextAwareProcessor implements BeanPostProcessor {
private void invokeAwareInterfaces(Object bean) {
if (bean instanceof EnvironmentAware) {
((EnvironmentAware) bean).setEnvironment(this.applicationContext.getEnvironment());
}
if (bean instanceof EmbeddedValueResolverAware) {
((EmbeddedValueResolverAware) bean).setEmbeddedValueResolver(this.embeddedValueResolver);
}
if (bean instanceof ResourceLoaderAware) {
((ResourceLoaderAware) bean).setResourceLoader(this.applicationContext);
}
if (bean instanceof ApplicationEventPublisherAware) {
((ApplicationEventPublisherAware) bean).setApplicationEventPublisher(this.applicationContext);
}
if (bean instanceof MessageSourceAware) {
((MessageSourceAware) bean).setMessageSource(this.applicationContext);
}
if (bean instanceof ApplicationContextAware) {
((ApplicationContextAware) bean).setApplicationContext(this.applicationContext);
}
}
}
```
### BeanNameAware
这个类也是Aware扩展的一种， ，
使用场景为： ， ，
```java
public class NormalBeanA implements BeanNameAware{
public NormalBeanA() {
System.out.println("NormalBean constructor");
}
@Override
public void setBeanName(String name) {
System.out.println("[BeanNameAware] " + name);
}
}
```
### @PostConstruct
这个并不算一个扩展点， ，
```java
public class NormalBeanA {
public NormalBeanA() {
System.out.println("NormalBean constructor");
}
@PostConstruct
public void init(){
System.out.println("[PostConstruct] NormalBeanA");
}
}
```
### InitializingBean
这个类， ， ， ， ，
```java
public class NormalBeanA implements InitializingBean{
@Override
public void afterPropertiesSet() throws Exception {
System.out.println("[InitializingBean] NormalBeanA");
}
}
```
### FactoryBean
一般情况下， ， ， ， ， ， ， ， ， < T > `的形式。
使用场景： ， ， ， ，
```java
public class TestFactoryBean implements FactoryBean< TestFactoryBean.TestFactoryInnerBean > {
@Override
public TestFactoryBean.TestFactoryInnerBean getObject() throws Exception {
System.out.println("[FactoryBean] getObject");
return new TestFactoryBean.TestFactoryInnerBean();
}
@Override
public Class< ?> getObjectType() {
return TestFactoryBean.TestFactoryInnerBean.class;
}
@Override
public boolean isSingleton() {
return true;
}
public static class TestFactoryInnerBean{
}
}
```
### SmartInitializingSingleton
这个接口中只有一个方法`afterSingletonsInstantiated`，其作用是是 在spring容器管理的所有单例对象（ ）
使用场景：用户可以扩展此接口在对所有单例对象初始化完毕后，做一些后置的业务处理。
```java
public class TestSmartInitializingSingleton implements SmartInitializingSingleton {
@Override
public void afterSingletonsInstantiated() {
System.out.println("[TestSmartInitializingSingleton]");
}
}
```
### CommandLineRunner
这个接口也只有一个方法：`run(String... args)`，触发时机为整个项目启动完毕后，自动执行。如果有多个`CommandLineRunner`，可以利用`@Order`来进行排序。使用场景：用户扩展此接口，进行启动项目之后一些业务的预处理。
```java
public class TestCommandLineRunner implements CommandLineRunner {
@Override
public void run(String... args) throws Exception {
System.out.println("[TestCommandLineRunner]");
}
}
```
### DisposableBean
这个扩展点也只有一个方法：`destroy()`，其触发时机为当此对象销毁时，会自动执行这个方法。比如说运行`applicationContext.registerShutdownHook`时，就会触发这个方法。
```java
public class NormalBeanA implements DisposableBean {
@Override
public void destroy() throws Exception {
System.out.println("[DisposableBean] NormalBeanA");
}
}
```
### ApplicationListener
准确的说， & springboot当中的一个扩展点， ， ， ， ， ，
接下来罗列下spring主要的内置事件：
- ContextRefreshedEvent
ApplicationContext 被初始化或刷新时，该事件被发布。这也可以在`ConfigurableApplicationContext`接口中使用 `refresh()` 方法来发生。此处的初始化是指： ， ，
- ContextStartedEvent
当使用 `ConfigurableApplicationContext` （ ） `ApplicationContext` 时，该事件被发布。你可以调查你的数据库，或者你可以在接受到这个事件后重启任何停止的应用程序。
- ContextStoppedEvent
当使用 `ConfigurableApplicationContext` 接口中的 `stop()` 停止`ApplicationContext` 时，发布这个事件。你可以在接受到这个事件后做必要的清理的工作
- ContextClosedEvent
当使用 `ConfigurableApplicationContext` 接口中的 `close()` 方法关闭 `ApplicationContext` 时，该事件被发布。一个已关闭的上下文到达生命周期末端；它不能被刷新或重启
- RequestHandledEvent
这是一个 web-specific 事件，告诉所有 bean HTTP 请求已经被服务。只能应用于使用DispatcherServlet的Web应用。在使用Spring作为前端的MVC控制器时， ，
## MVC
### HTTP请求处理流程
### 参数绑定
### 过滤器、拦截器、AOP执行顺序
## IOC
### Bean的加载过程
### Bean实例化的过程
## AOP
## Others
## FactoryBean
```java
public class UserFactoryBean implements FactoryBean< User > {
@Override
public User getObject() throws Exception {
User user = new User();
System.out.println("调用 UserFactoryBean 的 getObject 方法生成 Bean:" + user);
return user;
}
@Override
public Class< ?> getObjectType() {
// 这个 FactoryBean 返回的Bean的类型
return User.class;
}
}
```
**FactoryBean在Mybatis中的实践**
Mybatis的Mapper接口的动态代理也是通过FactoryBean注入到Spring中的。
```java
public class MapperFactoryBean< T > extends SqlSessionDaoSupportimplements FactoryBean< T > {
// mapper的接口类型
private Class< T > mapperInterface;
@Override
public T getObject() throws Exception {
// 通过SqlSession获取接口的动态搭理对象
return getSqlSession().getMapper(this.mapperInterface);
}
@Override
public Class< T > getObjectType() {
return this.mapperInterface;
}
}
```
**FactoryBean在OpenFeign中的实践**
FeignClient接口的动态代理也是通过FactoryBean注入到Spring中的。
```java
public class FeignClientFactoryBean implements
FactoryBean< Object > , InitializingBean, ApplicationContextAware {
// FeignClient接口类型
private Class< ?> type;
@Override
public Object getObject() throws Exception {
return getTarget();
}
@Override
public Class< ?> getObjectType() {
return type;
}
}
```
## @Import
@Import 的核心作用就是导入配置类，并且还可以根据配合（比如@EnableXXX）
### 实现ImportSelector接口
```java
public class UserImportSelector implements ImportSelector {
@Override
public String[] selectImports(AnnotationMetadata importingClassMetadata) {
System.out.println("调用 UserImportSelector 的 selectImports 方法获取一批类限定名");
return new String[]{"com.sanyou.spring.extension.User"};
}
}
@Import (UserImportSelector.class)
public class Application {
public static void main(String[] args) {
AnnotationConfigApplicationContext applicationContext =
new AnnotationConfigApplicationContext();
//将 Application 注册到容器中
applicationContext.register(Application.class);
applicationContext.refresh();
System.out.println("获取到的Bean为" + applicationContext.getBean(User.class));
}
}
// 运行结果如下：
// 调用 UserImportSelector 的 selectImports 方法获取一批类限定名
// 获取到的Bean为com.sanyou.spring.extension.User@282003e1
```
### 实现ImportBeanDefinitionRegistrar接口
```java
public class UserImportBeanDefinitionRegistrar implements ImportBeanDefinitionRegistrar {
@Override
public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry, BeanNameGenerator importBeanNameGenerator) {
//构建一个 BeanDefinition , Bean的类型为 User
AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.rootBeanDefinition(User.class)
// 设置 User 这个Bean的属性username的值为张小凡
.addPropertyValue("username", "张小凡")
.getBeanDefinition();
System.out.println("往Spring容器中注入User");
//把 User 这个Bean的定义注册到容器中
registry.registerBeanDefinition("user", beanDefinition);
}
}
// 导入 UserImportBeanDefinitionRegistrar
@Import (UserImportBeanDefinitionRegistrar.class)
public class Application {
public static void main(String[] args) {
AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext();
//将 Application 注册到容器中
applicationContext.register(Application.class);
applicationContext.refresh();
User user = applicationContext.getBean(User.class);
System.out.println("获取到的Bean为" + user + "， ：
}
}
// 运行结果如下：
// 往Spring容器中注入User
// 获取到的Bean为com.sanyou.spring.extension.User@6385cb26， ：
```
## BeanPostProcessor
Bean的后置处理器， ， ， ， ， ， ：
- InstantiationAwareBeanPostProcessor
- DestructionAwareBeanPostProcessor
```java
public class UserBeanPostProcessor implements BeanPostProcessor {
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof User) {
//如果当前的Bean的类型是 User ，就把这个对象 username 的属性赋值为 张小凡
((User) bean).setUsername("张小凡");
}
return bean;
}
}
public class Application {
public static void main(String[] args) {
AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext();
//将 UserBeanPostProcessor 和 User 注册到容器中
applicationContext.register(UserBeanPostProcessor.class);
applicationContext.register(User.class);
applicationContext.refresh();
User user = applicationContext.getBean(User.class);
System.out.println("获取到的Bean为" + user + "， ：
}
}
// 运行结果如下：
// 获取到的Bean为com.sanyou.spring.extension.User@21a947fe， ：
```
**Spring内置的BeanPostProcessor**
| BeanPostProcessor | 作用 |
| :------------------------------------- | :--------------------------------------------- |
| AutowiredAnnotationBeanPostProcessor | 处理@Autowired、@Value注解 |
| CommonAnnotationBeanPostProcessor | 处理@Resource、@PostConstruct、@PreDestroy注解 |
| AnnotationAwareAspectJAutoProxyCreator | 处理一些注解或者是AOP切面的动态代理 |
| ApplicationContextAwareProcessor | 处理Aware接口注入的 |
| AsyncAnnotationBeanPostProcessor | 处理@Async注解 |
| ScheduledAnnotationBeanPostProcessor | 处理@Scheduled注解 |
**BeanPostProcessor在Dubbo中的应用**
在Dubbo中可以通过@DubboReference(@Reference)来引用生产者提供的接口， ，
```java
public class ReferenceAnnotationBeanPostProcessor
extends AbstractAnnotationBeanPostProcessor
implements ApplicationContextAware, BeanFactoryPostProcessor {
// 忽略
}
```
## BeanFactoryPostProcessor
BeanFactoryPostProcessor是可以对Spring容器做处理的， ， ，
```java
public class MyBeanFactoryPostProcessor implements BeanFactoryPostProcessor {
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
// 禁止循环依赖
((DefaultListableBeanFactory) beanFactory).setAllowCircularReferences(false);
}
}
```
## Spring SPI机制
SPI全称为 (Service Provider Interface)， ， ， ， ， ， ， ，
- JDK：
- Dubbo：
- Spring：
Spring的SPI机制规定， ， ， ， ， ， ，
### 自动装配
### PropertySourceLoader
在SpringBoot环境下， ，
```java
public interface PropertySourceLoader {
// 可以支持哪种文件格式的解析
String[] getFileExtensions();
// 解析配置文件， ， < ?>结合返回回去
List< PropertySource < ? > > load(String name, Resource resource) throws IOException;
}
```
对于PropertySourceLoader的实现， ：
- **PropertiesPropertySourceLoader**：
```java
public class PropertiesPropertySourceLoader implements PropertySourceLoader {
private static final String XML_FILE_EXTENSION = ".xml";
public PropertiesPropertySourceLoader() {
}
public String[] getFileExtensions() {
return new String[]{"properties", "xml"};
}
public List< PropertySource < ? > > load(String name, Resource resource) throws IOException {
Map< String , ? > properties = this.loadProperties(resource);
return properties.isEmpty() ? Collections.emptyList() : Collections.singletonList(new OriginTrackedMapPropertySource(name, Collections.unmodifiableMap(properties), true));
}
private Map< String , ? > loadProperties(Resource resource) throws IOException {
String filename = resource.getFilename();
return (Map)(filename != null & & filename.endsWith(".xml") ? PropertiesLoaderUtils.loadProperties(resource) : (new OriginTrackedPropertiesLoader(resource)).load());
}
}
```
- **YamlPropertySourceLoader**：
```java
public class YamlPropertySourceLoader implements PropertySourceLoader {
@Override
public String[] getFileExtensions() {
return new String[] { "yml", "yaml" };
}
@Override
public List< PropertySource < ? > > load(String name, Resource resource) throws IOException {
if (!ClassUtils.isPresent("org.yaml.snakeyaml.Yaml", null)) {
throw new IllegalStateException(
"Attempted to load " + name + " but snakeyaml was not found on the classpath");
}
List< Map < String , Object > > loaded = new OriginTrackedYamlLoader(resource).load();
if (loaded.isEmpty()) {
return Collections.emptyList();
}
List< PropertySource < ? > > propertySources = new ArrayList< >(loaded.size());
for (int i = 0; i < loaded.size ( ) ; i + + ) {
String documentNumber = (loaded.size() != 1) ? " (document #" + i + ")" : "";
propertySources.add(new OriginTrackedMapPropertySource(name + documentNumber,
Collections.unmodifiableMap(loaded.get(i)), true));
}
return propertySources;
}
}
```
所以可以看出， ，
**实现可以读取json格式的配置文件**
**第一步： ，
```java
public class JsonPropertySourceLoader implements PropertySourceLoader {
@Override
public String[] getFileExtensions() {
//这个方法表明这个类支持解析以json结尾的配置文件
return new String[]{"json"};
}
@Override
public List< PropertySource < ? > > load(String name, Resource resource) throws IOException {
ReadableByteChannel readableByteChannel = resource.readableChannel();
ByteBuffer byteBuffer = ByteBuffer.allocate((int) resource.contentLength());
//将文件内容读到 ByteBuffer 中
readableByteChannel.read(byteBuffer);
//将读出来的字节转换成字符串
String content = new String(byteBuffer.array());
// 将字符串转换成 JSONObject
JSONObject jsonObject = JSON.parseObject(content);
Map< String , Object > map = new HashMap< >(jsonObject.size());
//将 json 的键值对读出来，放入到 map 中
for (String key : jsonObject.keySet()) {
map.put(key, jsonObject.getString(key));
}
return Collections.singletonList(new MapPropertySource("jsonPropertySource", map));
}
}
```
**第二步：
基于SPI机制加载PropertySourceLoader实现。因此在spring.factories文件中配置如下：
```properties
org.springframework.boot.env.PropertySourceLoader=\
cn.test.JsonPropertySourceLoader
```
**第三步：测试验证**
在application.json中配置如下：
```
{
"test.userName": "张小凡"
}
```
定义配置类：
```java
public class User {
// 注入配置文件的属性
@Value ("${test.userName:}")
private String userName;
}
```
启动项目
```java
@SpringBootApplication
public class Application {
public static void main(String[] args) {
ConfigurableApplicationContext applicationContext = SpringApplication.run(Application.class);
User user = applicationContext.getBean(User.class);
System.out.println("获取到的Bean为" + user + "， ：
}
@Bean
public User user() {
return new User();
}
}
// 运行结果如下：
// 获取到的Bean为com.sanyou.spring.extension.User@481ba2cf， ：
```
**Nacos对于PropertySourceLoader的实现**
```properties
org.springframework.boot.env.PropertySourceLoader=\
com.alibaba.cloud.nacos.parser.NacosJsonPropertySourceLoader,\
com.alibaba.cloud.nacos.parser.NacosXmlPropertySourceLoader
```
Nacos作为配置中心， ， ， ， ，
### ApplicationContextInitializer
ApplicationContextInitializer也是SpringBoot启动过程的一个扩展点。在SpringBoot启动过程， ，
```properties
org.springframework.context.ApplicationContextInitializer=\
org.springframework.boot.context.ConfigurationWarningApplicationContextInitializer,\
org.springframework.boot.context.ContextIdApplicationContextInitializer,\
org.springframework.boot.context.config.DelegatingApplicationContextInitializer,\
org.springframework.boot.rsocket.context.RSocketPortInfoApplicationContextInitializer,\
org.springframework.boot.web.context.ServerPortInfoApplicationContextInitializer
```
注意： ， ， ，
### EnvironmentPostProcessor
EnvironmentPostProcessor在SpringBoot启动过程中， ， ， ，
这个类的作用就是用来处理外部化配置文件的， ， ，
### ApplicationRunner和CommandLineRunner
ApplicationRunner和CommandLineRunner都是在SpringBoot成功启动之后会调用， ， ， ， ，
所以要想扩展这个点， ，
## Spring Event 事件
什么是Spring Event 事件， ， ，
```java
// 火灾事件
public class FireEvent extends ApplicationEvent {
public FireEvent(String source) {
super(source);
}
}
// 打119的火灾事件的监听器
public class Call119FireEventListener implements ApplicationListener< FireEvent > {
@Override
public void onApplicationEvent(FireEvent event) {
System.out.println("打119");
}
}
// 救人的火灾事件的监听器
public class SavePersonFireEventListener implements ApplicationListener< FireEvent > {
@Override
public void onApplicationEvent(FireEvent event) {
System.out.println("救人");
}
}
public class Application {
public static void main(String[] args) {
AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext();
//将 事件监听器 注册到容器中
applicationContext.register(Call119FireEventListener.class);
applicationContext.register(SavePersonFireEventListener.class);
applicationContext.refresh();
// 发布着火的事件，触发监听
applicationContext.publishEvent(new FireEvent("着火了"));
}
}
```
**Spring内置的事件**
Spring内置的事件很多，
| 事件类型 | 触发时机 |
| :-------------------- | :----------------------------------------------------------- |
| ContextRefreshedEvent | 在调用ConfigurableApplicationContext 接口中的refresh()方法时触发 |
| ContextStartedEvent | 在调用ConfigurableApplicationContext的start()方法时触发 |
| ContextStoppedEvent | 在调用ConfigurableApplicationContext的stop()方法时触发 |
| ContextClosedEvent | 当ApplicationContext被关闭时触发该事件，
### ApplicationEvent
事件的父类，所有具体的事件都得继承这个类，构造方法的参数是这个事件携带的参数，监听器就可以通过这个参数来进行一些业务操作。
### ApplicationListener
事件监听的接口， ， ， ， ， ， ，
### ApplicationEventPublisher
事件发布器， ，
ApplicationContext实现了ApplicationEventPublisher接口，
## 命名空间
**第一步：
```xml
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<!-- xmlns 和 targetNamespace 需要定义， ， -->
< xsd:schema xmlns = "http://sanyou.com/schema/sanyou"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
targetNamespace="http://sanyou.com/schema/sanyou">
< xsd:import namespace = "http://www.w3.org/XML/1998/namespace" / >
< xsd:complexType name = "Bean" >
< xsd:attribute name = "class" type = "xsd:string" use = "required" / >
< / xsd:complexType >
<!-- sanyou 便签的子标签， ， ， -->
< xsd:element name = "mybean" type = "Bean" / >
< / xsd:schema >
```
这个xsd文件来指明sanyou这个命名空间下有哪些标签和属性。这里我只指定了一个标签 mybean， ， ， ， ，
**第二步：解析这个命名空间**
解析命名空间很简单， ，
```java
public class SanYouNameSpaceHandler extends NamespaceHandlerSupport {
@Override
public void init() {
//注册解析 mybean 标签的解析器
registerBeanDefinitionParser("mybean", new SanYouBeanDefinitionParser());
}
private static class SanYouBeanDefinitionParser extends AbstractSingleBeanDefinitionParser {
@Override
protected boolean shouldGenerateId() {
return true;
}
@Override
protected String getBeanClassName(Element element) {
return element.getAttribute("class");
}
}
}
```
SanYouNameSpaceHandler的作用就是将sanyou命名空间中的mybean这个标签读出来， ， ， ，
**第三步：
先创建spring.handlers和spring.schemas文件。spring.handlers文件内容：
```properties
http\://sanyou.com/schema/sanyou=com.sanyou.spring.extension.namespace.SanYouNameSpaceHandler
```
通过spring.handlers配置文件， ：
```properties
http\://sanyou.com/schema/sanyou.xsd=META-INF/sanyou.xsd
```
spring.schemas配置xsd文件的路径。文件都有了，
**第四步：测试**
先构建一个applicationContext.xml文件，
```xml
<?xml version="1.0" encoding="UTF-8"?>
< beans xmlns:xsi = "http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://www.springframework.org/schema/beans"
xmlns:sanyou="http://sanyou.com/schema/sanyou"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://sanyou.com/schema/sanyou
http://sanyou.com/schema/sanyou.xsd
">
<!-- 使用 sanyou 标签，配置一个 User Bean -->
< sanyou:mybean class = "com.sanyou.spring.extension.User" / >
< / beans >
```
再写个测试类：
```java
public class Application {
public static void main(String[] args) {
ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");
applicationContext.refresh();
User user = applicationContext.getBean(User.class);
System.out.println(user);
}
}
// 运行结果
// com.sanyou.spring.extension.User@27fe3806
```
# SpringCloud
## Eurake
Netflix Eureka 是由 Netflix 开源的一款基于 REST 的服务发现组件，包括 Eureka Server 及 Eureka Client。
- **Eurake客户端**：
- **Eureka服务端**： ， ， ，
### Eurka工作流程
Eureka的工作流程如下：
1. **Eureka Server 启动成功，等待服务端注册** 。在启动过程中如果配置了集群，集群之间定时通过 Replicate 同步注册表，每个 Eureka Server 都存在独立完整的服务注册表信息
2. Eureka Client 启动时根据配置的 Eureka Server 地址**去注册中心注册服务**
3. Eureka Client 会**每30s向 Eureka Server 发送一次心跳请求**，证明客户端服务正常
4. 当Eureka Server **90s内没有收到Eureka Client的心跳， ，会注销该实例
5. ** 单位时间内Eureka Server统计到有大量的Eureka Client没有上送心跳，
6. ** 当Eureka Client心跳请求恢复正常之后，
7. **Eureka Client定时全量或者增量从注册中心获取服务注册表，
8. 服务调用时， ， ，
9. Eureka Client获取到目标服务器信息，
10. Eureka Client程序关闭时向Eureka Server发送取消请求，
### Eureka Server
Eureka Server注册中心服务端主要对外提供了三个功能：
- **服务注册**
服务提供者启动时，会通过 Eureka Client 向 Eureka Server 注册信息， ，
- **提供注册表**
服务消费者在调用服务时，如果 Eureka Client 没有缓存注册表的话，会从 Eureka Server 获取最新的注册表。
- **同步状态**
Eureka Client 通过注册、心跳机制和 Eureka Server 同步当前客户端的状态。
**自我保护机制**
默认情况下，如果 Eureka Server 在一定的 90s 内没有接收到某个微服务实例的心跳， ， ， ， ， ， ， ， ？ ，
Eureka Server进入自我保护机制， ：
- **Eureka不再从注册列表中移除因为长时间没收到心跳而应该过期的服务**
- **Eureka仍然能够接受新服务的注册和查询请求，
- **当网络稳定时，当前实例新的注册信息会被同步到其它节点中**
Eureka自我保护机制是为了防止误杀服务而提供的一个机制：
- **当个别客户端出现心跳失联时，则认为是客户端的问题，剔除掉客户端**
- **当Eureka捕获到大量的心跳失败时， ，
- **当客户端心跳恢复时，
如果在保护期内刚好这个服务提供者非正常下线了，此时服务消费者就会拿到一个无效的服务实例，即会调用失败。对于这个问题需要服务消费者端要有一些容错机制，如重试，断路器等。通过在 Eureka Server 配置如下参数，开启或者关闭保护机制，生产环境建议打开：
```properties
eureka.server.enable-self-preservation=true
```
### Eureka Client
Eureka Client注册中心客户端是一个 Java 客户端，用于简化与 Eureka Server 的交互。Eureka Client 会拉取、更新和缓存 Eureka Server 中的信息。因此当所有的 Eureka Server 节点都宕掉，服务消费者依然可以使用缓存中的信息找到服务提供者，但是当服务有更改的时候会出现信息不一致。
- **Register —— 服务注册**
服务的提供者，将自身注册到注册中心，服务提供者也是一个 Eureka Client。当 Eureka Client 向 Eureka Server 注册时，它提供自身的元数据，比如 IP 地址、端口，运行状况指示符 URL，
- **Renew —— 服务续约**
Eureka Client 会每隔 30 秒发送一次心跳来续约。 通过续约来告知 Eureka Server 该 Eureka Client 运行正常，没有出现问题。 默认情况下，如果 Eureka Server 在 90 秒内没有收到 Eureka Client 的续约，
```properties
# 服务续约任务的调用间隔时间，
eureka.instance.lease-renewal-interval-in-seconds=30
# 服务失效的时间，
eureka.instance.lease-expiration-duration-in-seconds=90
```
- **Eviction —— 服务剔除**
当Eureka Client和Eureka Server不再有心跳时， ，
- **Cancel —— 服务下线**
Eureka Client 在程序关闭时向 Eureka Server 发送取消请求。 发送请求后，该客户端实例信息将从 Eureka Server 的实例注册表中删除。该下线请求不会自动完成，它需要调用以下内容：
```java
DiscoveryManager.getInstance().shutdownComponent();
```
- **GetRegisty —— 获取注册列表信息**
Eureka Client 从服务器获取注册表信息， ， （ ） ，
如果由于某种原因导致注册列表信息不能及时匹配， ， ，
```properties
# 启用服务消费者从注册中心拉取服务列表的功能
eureka.client.fetch-registry=true
# 设置服务消费者从注册中心拉取服务列表的间隔
eureka.client.registry-fetch-interval-seconds=30
```
- **Remote Call —— 远程调用**
当Eureka Client从注册中心获取到服务提供者信息后， ； ，
### Eureka缓存机制
**Eureka Server数据存储**
Eureka Server的数据存储层是双层的 ConcurrentHashMap：
```java
private final ConcurrentHashMap< String , Map < String , Lease < InstanceInfo > >> registry= new ConcurrentHashMap< String , Map < String , Lease < InstanceInfo > >>();
```
- **key**：为服务名，即 `spring.application.name` ，也就是客户端实例注册的应用名
- **subKey**：为 `instanceId` ，
- **Lease< InstanceInfo > **： ，
**Eureka Server缓存机制**
Eureka Server为了提供响应效率， ， ，
- **第一层缓存： ，
依赖定时从 readWriteCacheMap 同步数据，默认时间为 30 秒。
readOnlyCacheMap：
- **第二层缓存： ，
readWriteCacheMap： ， ， ， ， ，
Eureka Client 获取全量或者增量的数据时，会先从一级缓存中获取；如果一级缓存中不存在，再从二级缓存中获取；如果二级缓存也不存在，这时候先将存储层的数据同步到缓存中，再从缓存中获取。通过 Eureka Server 的二层缓存机制，可以非常有效地提升 Eureka Server 的响应时间，通过数据存储层和缓存层的数据切割，根据使用场景来提供不同的数据支持。
**其它缓存设计**
除过 Eureka Server 端存在缓存外， ， ，
1. 首先，
2. Eureka Client 对已经获取到的注册信息也做了 30s 缓存
3. 负载均衡组件 Ribbon 也有 30s 缓存
这三个缓存加起来，就有可能导致服务注册最长延迟 90s ，这个需要我们在特殊业务场景中注意其产生的影响。
### 常用配置
Eureka Server常用配置：
```properties
#服务端开启自我保护模式，前面章节有介绍
eureka.server.enable-self-preservation=true
#扫描失效服务的间隔时间（ ， ）
eureka.server.eviction-interval-timer-in-ms= 60000
#间隔多长时间，清除过期的 delta 数据
eureka.server.delta-retention-timer-interval-in-ms=0
#请求频率限制器
eureka.server.rate-limiter-burst-size=10
#是否开启请求频率限制器
eureka.server.rate-limiter-enabled=false
#请求频率的平均值
eureka.server.rate-limiter-full-fetch-average-rate=100
#是否对标准的client进行频率请求限制。如果是false，
eureka.server.rate-limiter-throttle-standard-clients=false
#注册服务、拉去服务列表数据的请求频率的平均值
eureka.server.rate-limiter-registry-fetch-average-rate=500
#设置信任的client list
eureka.server.rate-limiter-privileged-clients=
#在设置的时间范围类，
eureka.server.renewal-percent-threshold=0.85
#多长时间更新续约的阈值
eureka.server.renewal-threshold-update-interval-ms=0
#对于缓存的注册数据，多长时间过期
eureka.server.response-cache-auto-expiration-in-seconds=180
#多长时间更新一次缓存中的服务注册数据
eureka.server.response-cache-update-interval-ms=0
#缓存增量数据的时间，以便在检索的时候不丢失信息
eureka.server.retention-time-in-m-s-in-delta-queue=0
#当时间戳不一致的时候，是否进行同步
eureka.server.sync-when-timestamp-differs=true
#是否采用只读缓存策略，只读策略对于缓存的数据不会过期。
eureka.server.use-read-only-response-cache=true
################server node 与 node 之间关联的配置#####################33
#发送复制数据是否在request中，
eureka.server.enable-replicated-request-compression=false
#指示群集节点之间的复制是否应批处理以提高网络效率。
eureka.server.batch-replication=false
#允许备份到备份池的最大复制事件数量。而这个备份池负责除状态更新的其他事件。可以根据内存大小，超时和复制流量，来设置此值得大小
eureka.server.max-elements-in-peer-replication-pool=10000
#允许备份到状态备份池的最大复制事件数量
eureka.server.max-elements-in-status-replication-pool=10000
#多个服务中心相互同步信息线程的最大空闲时间
eureka.server.max-idle-thread-age-in-minutes-for-peer-replication=15
#状态同步线程的最大空闲时间
eureka.server.max-idle-thread-in-minutes-age-for-status-replication=15
#服务注册中心各个instance相互复制数据的最大线程数量
eureka.server.max-threads-for-peer-replication=20
#服务注册中心各个instance相互复制状态数据的最大线程数量
eureka.server.max-threads-for-status-replication=1
#instance之间复制数据的通信时长
eureka.server.max-time-for-replication=30000
#正常的对等服务instance最小数量。-1表示服务中心为单节点。
eureka.server.min-available-instances-for-peer-replication=-1
#instance之间相互复制开启的最小线程数量
eureka.server.min-threads-for-peer-replication=5
#instance之间用于状态复制，
eureka.server.min-threads-for-status-replication=1
#instance之间复制数据时可以重试的次数
eureka.server.number-of-replication-retries=5
#eureka节点间间隔多长时间更新一次数据。默认10分钟。
eureka.server.peer-eureka-nodes-update-interval-ms=600000
#eureka服务状态的相互更新的时间间隔。
eureka.server.peer-eureka-status-refresh-time-interval-ms=0
#eureka对等节点间连接超时时间
eureka.server.peer-node-connect-timeout-ms=200
#eureka对等节点连接后的空闲时间
eureka.server.peer-node-connection-idle-timeout-seconds=30
#节点间的读数据连接超时时间
eureka.server.peer-node-read-timeout-ms=200
#eureka server 节点间连接的总共最大数量
eureka.server.peer-node-total-connections=1000
#eureka server 节点间连接的单机最大数量
eureka.server.peer-node-total-connections-per-host=10
#在服务节点启动时，
eureka.server.registry-sync-retries=
#在服务节点启动时，
eureka.server.registry-sync-retry-wait-ms=
#当eureka server启动的时候， ，
eureka.server.wait-time-in-ms-when-sync-empty=0
```
Eureka Client 常用配置：
```properties
#该客户端是否可用
eureka.client.enabled=true
#实例是否在eureka服务器上注册自己的信息以供其他服务发现，
eureka.client.register-with-eureka=false
#此客户端是否获取eureka服务器注册表上的注册信息，
eureka.client.fetch-registry=false
#是否过滤掉，
eureka.client.filter-only-up-instances=true
#与Eureka注册服务中心的通信zone和url地址
eureka.client.serviceUrl.defaultZone=http://${eureka.instance.hostname}:${server.port}/eureka/
#client连接Eureka服务端后的空闲等待时间，
eureka.client.eureka-connection-idle-timeout-seconds=30
#client连接eureka服务端的连接超时时间，
eureka.client.eureka-server-connect-timeout-seconds=5
#client对服务端的读超时时长
eureka.client.eureka-server-read-timeout-seconds=8
#client连接all eureka服务端的总连接数，
eureka.client.eureka-server-total-connections=200
#client连接eureka服务端的单机连接数量，
eureka.client.eureka-server-total-connections-per-host=50
#执行程序指数回退刷新的相关属性， ，
eureka.client.cache-refresh-executor-exponential-back-off-bound=10
#执行程序缓存刷新线程池的大小，
eureka.client.cache-refresh-executor-thread-pool-size=2
#心跳执行程序回退相关的属性， ，
eureka.client.heartbeat-executor-exponential-back-off-bound=10
#心跳执行程序线程池的大小,默认为5
eureka.client.heartbeat-executor-thread-pool-size=5
# 询问Eureka服务url信息变化的频率（ ） ，
eureka.client.eureka-service-url-poll-interval-seconds=300
#最初复制实例信息到eureka服务器所需的时间（ ） ，
eureka.client.initial-instance-info-replication-interval-seconds=40
#间隔多长时间再次复制实例信息到eureka服务器，
eureka.client.instance-info-replication-interval-seconds=30
#从eureka服务器注册表中获取注册信息的时间间隔（ ） ，
eureka.client.registry-fetch-interval-seconds=30
# 获取实例所在的地区。默认为us-east-1
eureka.client.region=us-east-1
#实例是否使用同一zone里的eureka服务器， ， ，
eureka.client.prefer-same-zone-eureka=true
# 获取实例所在的地区下可用性的区域列表， （ ）
eureka.client.availability-zones.china=defaultZone,defaultZone1,defaultZone2
#eureka服务注册表信息里的以逗号隔开的地区名单， ，
eureka.client.fetch-remote-regions-registry=
#服务器是否能够重定向客户端请求到备份服务器。 如果设置为false， ， ，
eureka.client.allow-redirects=false
#客户端数据接收
eureka.client.client-data-accept=
#增量信息是否可以提供给客户端看，
eureka.client.disable-delta=false
#eureka服务器序列化/反序列化的信息中获取“_”符号的的替换字符串。默认为“__“
eureka.client.escape-char-replacement=__
#eureka服务器序列化/反序列化的信息中获取“$”符号的替换字符串。默认为“_-”
eureka.client.dollar-replacement="_-"
#当服务端支持压缩的情况下，
eureka.client.g-zip-content=true
#是否记录eureka服务器和客户端之间在注册表的信息方面的差异，
eureka.client.log-delta-diff=false
# 如果设置为true,客户端的状态更新将会点播更新到远程服务器上，
eureka.client.on-demand-update-status-change=true
#此客户端只对一个单一的VIP注册表的信息感兴趣。默认为null
eureka.client.registry-refresh-single-vip-address=
#client是否在初始化阶段强行注册到服务中心，
eureka.client.should-enforce-registration-at-init=false
#client在shutdown的时候是否显示的注销服务从服务中心，
eureka.client.should-unregister-on-shutdown=true
```
Eureka Instance 常用配置：
```properties
#服务注册中心实例的主机名
eureka.instance.hostname=localhost
#注册在Eureka服务中的应用组名
eureka.instance.app-group-name=
#注册在的Eureka服务中的应用名称
eureka.instance.appname=
#该实例注册到服务中心的唯一ID
eureka.instance.instance-id=
#该实例的IP地址
eureka.instance.ip-address=
#该实例，
eureka.instance.prefer-ip-address=false
```
### Eureka集群原理
- **Eureka Server集群相互之间通过 Replicate(复制) 来同步数据**
相互之间不区分主节点和从节点，所有的节点都是平等的。在这种架构中，节点通过彼此互相注册来提高可用性，每个节点需要添加一个或多个有效的 serviceUrl 指向其它节点。
- **如果某台Eureka Server宕机，
当宕机的服务器重新恢复后，
- **Eureka Server的同步遵循一个原则： ，
如果存在多个节点，只需要将节点之间两两连接起来形成通路，那么其它注册中心都可以共享信息。每个 Eureka Server 同时也是 Eureka Client，
- **Eureka Server集群之间的状态是采用异步方式同步的**
所以不保证节点间的状态一定是一致的，不过基本能保证最终状态是一致的。
**Eureka 分区**
Eureka 提供了 Region 和 Zone 两个概念来进行分区，这两个概念均来自于亚马逊的 AWS:
- **region**：可以理解为地理上的不同区域，比如亚洲地区，中国区或者深圳等等。没有具体大小的限制。根据项目具体的情况，可以自行合理划分 region
- **zone**：可以简单理解为 region 内的具体机房，比如说 region 划分为深圳，然后深圳有两个机房，就可以在此 region 之下划分出 zone1、zone2 两个 zone
上图中的 us-east-1c、us-east-1d、us-east-1e 就代表了不同的 Zone。Zone 内的 Eureka Client 优先和 Zone 内的 Eureka Server 进行心跳同步，同样调用端优先在 Zone 内的 Eureka Server 获取服务列表，当 Zone 内的 Eureka Server 挂掉之后，才会从别的 Zone 中获取信息。
**Eurka 保证 AP**
Eureka Server 各个节点都是平等的，几个节点挂掉不会影响正常节点的工作，剩余的节点依然可以提供注册和查询服务。而 Eureka Client 在向某个 Eureka 注册时，如果发现连接失败，则会自动切换至其它节点。只要有一台 Eureka Server 还在，就能保证注册服务可用(保证可用性)，只不过查到的信息可能不是最新的(不保证强一致性)。
### Eureka一致性协议
Eureka 和 Zookeeper 的最大区别： Eureka 是 AP 模型， ， ， ， ， ？
#### 消息广播
1. Eureka Server 管理了全部的服务器列表（ ）
2. 当 Eureka Server 收到客户端的注册、下线、心跳请求时，通过 PeerEurekaNode 向其余的服务器进行消息广播，如果广播失败则重试，直到任务过期后取消任务，此时这两台服务器之间数据会出现短暂的不一致。
** 注意：** 虽然消息广播失败，但只要收到客户端的心跳，仍会向所有的服务器（包括失联的服务器）广播心跳任务。
3. 如果网络恢复正常，收到了其它服务器广播的心跳任务，此时可能有三种情况：
1. 一是脑裂很快恢复，一切正常；
2. 二是该实例已经自动过期，则重新进行注册；
3. 三是数据冲突，出现不一致的情况，则需要发起同步请求，其实也就是重新注册一次，同时踢除老的实例。
总之，通过集群之间的消息广播可以实现数据的最终一致性。
**服务注册**
1. Spring Cloud Eureka 在应用启动时，会在 EurekaAutoServiceRegistration 这个类初始化的时候，主动去 Eureka Server 端注册。
2. Eureka 在启动完成之后会启动一个 40 秒执行一次的定时任务，该任务会去监测自身的 IP 信息以及自身的配置信息是否发生改变，如果发生改变，则会重新发起注册。
3. 续约返回 404 状态码时，会去重新注册
**主动下线**
Eureka 会在 spring 容器销毁的时候执行 shutDown 方法 ，该方法首先会将自身的状态改为 DOWN，
**心跳续约与自动下线**
健康检测，一般都是 TTL(Time To Live) 机制。eg: 客户端每 5s 发送心跳，服务端 15s 没收到心跳包，更新实例状态为不健康， 30s 未收到心跳包，从服务列表中删除。**Eureka Server 默认每 30s 发送心跳包，
#### 崩溃恢复
**重启**
Spring Cloud Eureka 启动时，在初始化 EurekaServerBootstrap#initEurekaServerContext 时会调用 PeerAwareInstanceRegistryImpl#syncUp 从其它 Eureka 中同步数据。
**脑裂**
- 发生脑裂后：和 Eureka Server 同区的服务可以正常访问，而不同区的服务则自动过期。
- 脑裂恢复后：接收其它 Eureka Sever 发送过来的心跳请求，此时有三种情况：一是脑裂很快恢复，一切正常；二是该实例已经自动过期，则重新进行注册；三是数据冲突，出现不一致的情况，则需要发起同步请求。
## Zuul
Zuul 是由 Netflix 孵化的一个致力于“网关 “解决方案的开源组件。
**网关的作用**
- 统一入口：未全部为服务提供一个唯一的入口，网关起到外部和内部隔离的作用，保障了后台服务的安全性
- 鉴权校验：识别每个请求的权限，拒绝不符合要求的请求
- 动态路由：动态的将请求路由到不同的后端集群中
- 减少客户端与服务端的耦合：服务可以独立发展，通过网关层来做映射
## Feign
`Feign` 使用时分成两步：一是生成 Feign 的动态代理；二是 Feign 执行。
**总结：**
- 前两步是生成动态对象：将 Method 方法的注解解析成 MethodMetadata，
- 后几步是调用过程： ， ，
### Feign动态代理
`Feign` 的默认实现是 ReflectiveFeign，
1. Contract 统一将方法解析 MethodMetadata(*)，这样就可以通过实现不同的 Contract 适配各种 REST 声明式规范。
2. buildTemplate 实际上将 Method 方法的参数转换成 Request。
3. 将 metadata 和 buildTemplate 封装成 MethodHandler。
### 基于Feign实现负载均衡
基于 Feign 的负载均衡（整合 Ribbon） ，
```java
// RibbonClient 主要逻辑
private final Client delegate;
private final LBClientFactory lbClientFactory;
public Response execute(Request request, Request.Options options) throws IOException {
try {
URI asUri = URI.create(request.url());
String clientName = asUri.getHost();
URI uriWithoutHost = cleanUrl(request.url(), clientName);
// 封装 RibbonRequest，
LBClient.RibbonRequest ribbonRequest =
new LBClient.RibbonRequest(delegate, request, uriWithoutHost);
// executeWithLoadBalancer 实现负载均衡
return lbClient(clientName).executeWithLoadBalancer(
ribbonRequest,
new FeignOptionsClientConfig(options)).toResponse();
} catch (ClientException e) {
propagateFirstIOException(e);
throw new RuntimeException(e);
}
}
```
**总结：** 实际上是把 Client 对象包装了一下，通过 executeWithLoadBalancer 进行负载均衡，这是 Ribbon 提供了 API。更多关于 Ribbon 的负载均衡就不在这进一步说明了。
```java
public final class LBClient extends AbstractLoadBalancerAwareClient
< LBClient.RibbonRequest , LBClient . RibbonResponse > {
// executeWithLoadBalancer 方法通过负载均衡算法后，最终调用 execute
@Override
public RibbonResponse execute(RibbonRequest request, IClientConfig configOverride)
throws IOException, ClientException {
Request.Options options;
if (configOverride != null) {
options = new Request.Options(
configOverride.get(CommonClientConfigKey.ConnectTimeout, connectTimeout),
configOverride.get(CommonClientConfigKey.ReadTimeout, readTimeout),
configOverride.get(CommonClientConfigKey.FollowRedirects, followRedirects));
} else {
options = new Request.Options(connectTimeout, readTimeout);
}
// http 请求
Response response = request.client().execute(request.toRequest(), options);
if (retryableStatusCodes.contains(response.status())) {
response.close();
throw new ClientException(ClientException.ErrorType.SERVER_THROTTLED);
}
return new RibbonResponse(request.getUri(), response);
}
}
```
### 基于Feign实现熔断
基于 Feign 的熔断与限流（整合 Hystrix） ， ， ， `HystrixFeign` ，其实也就是实现了 HystrixInvocationHandler。
```java
// HystrixInvocationHandler 主要逻辑
public Object invoke(final Object proxy, final Method method, final Object[] args)
throws Throwable {
HystrixCommand< Object > hystrixCommand =
new HystrixCommand< Object > (setterMethodMap.get(method)) {
@Override
protected Object run() throws Exception {
return HystrixInvocationHandler.this.dispatch.get(method).invoke(args);
}
@Override
protected Object getFallback() {
};
}
...
return hystrixCommand.execute();
}
```
### Feign参数编码整体流程
**总结：**
- 前两步是`Feign`代理生成阶段， ，
- Contract 接口将 UserService 中每个接口中的方法及其注解解析为 MethodMetadata，
- RequestTemplate#request 编码为一个 Request 后就可以调用 Client#execute 发送 Http 请求
- Client 的具体实现有 HttpURLConnection、Apache HttpComponnets、OkHttp3 、Netty 等。本文关注前三步：即 Feign 方法元信息解析及参数编码过程。
### Feign整体装配流程
**总结：** OpenFeign 装配有两个入口：
1. @EnableAutoConfiguration 自动装配（ ）
```properties
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.openfeign.ribbon.FeignRibbonClientAutoConfiguration,\
org.springframework.cloud.openfeign.FeignAutoConfiguration
```
- FeignAutoConfiguration自动装配 FeignContext 和 Targeter，
- `FeignContext` 是每个 FeignClient 的装配上下文，默认的配置是 FeignClientsConfiguration
- `Targeter` 有两种实现：一是 DefaultTargeter， ； ** 二是 HystrixTargeter， ，
- `Client` ：自动装配 ApacheHttpClient， ， ，
- FeignRibbonClientAutoConfiguration实现负载均衡，
- `HttpClientFeignLoadBalancedConfiguration` ApacheHttpClient 实现负载均衡
- `OkHttpFeignLoadBalancedConfiguration` OkHttpClient实现负载均衡
- `DefaultFeignLoadBalancedConfiguration` Client.Default实现负载均衡
2. @EnableFeignClients 自动扫描
@EnableFeignClients 注入 FeignClientsRegistrar， @FeignClient 标注的接口包装成 FeignClientFactoryBean 对象，最终通过 Feign.Builder 生成该接口的代理对象。而 Feign.Builder 的默认配置是 FeignClientsConfiguration，
**注意：** 熔断与限流是 FeignAutoConfiguration 通过注入 HystrixTargeter 完成的，
### 其它
Feign是是一个声明式的Web Service客户端。
Feign就是使用了动态代理：
- 首先，如果你对某个接口定义了@FeignClient注解，
- 接着你要是调用那个接口，本质就是会调用 Feign创建的动态代理，
- Feign的动态代理会根据你在接口上的@RequestMapping等注解，
- 最后针对这个地址，发起请求、解析响应
## Ribbon
Ribbon Netflix 公司开源的一个负载均衡的组件。
Ribbon在服务消费者端配置和使用， ， ， ， ， ， ， ：
## Hystrix
Hystrix是Netstflix 公司开源的一个项目，它提供了熔断器功能，能够阻止分布式系统中出现联动故障。
Hystrix是隔离、熔断以及降级的一个框架， ， ， ， ， ， ， ， ， ， ， ， ：
## Gateway
Spring Cloud Gateway 是 Spring 官方基于 Spring 5.0、 Spring Boot 2.0 和 Project Reactor 等技术开发的网关， Spring Cloud Gateway 旨在为微服务架构提供简单、 有效且统一的 API 路由管理方式。
## Config
Spring Cloud 中提供了分布式配置中 Spring Cloud Config ，为外部配置提供了客户端和服务器端的支持。
## Bus
使用 Spring Cloud Bus, 可以非常容易地搭建起消息总线。
## OAuth2
Sprin Cloud 构建的微服务系统中可以使用 Spring Cloud OAuth2 来保护微服务系统。
## Sleuth
Spring Cloud Sleuth是Spring Cloud 个组件，它的主要功能是在分布式系统中提供服务链路追踪的解决方案。
# MyBatis
Mybatis是一个半ORM（ ） ， ， ， ， ，
作为一个半ORM框架， ，
通过xml 文件或注解的方式将要执行的各种 statement 配置起来， ， （ ）
由于MyBatis专注于SQL本身， ， ， ，
**Mybaits优缺点**
- 优点
- 基于SQL语句编程， ， ， ， ， ； ， ，
- 与JDBC相比， ， ，
- 很好的与各种数据库兼容（ ， ）
- 能够与Spring很好的集成
- 提供映射标签， ； ，
- 缺点
- SQL语句的编写工作量较大， ，
- SQL语句依赖于数据库， ，
**#{}和${}的区别是什么？**
- `${}` 是字符串替换
- `#{}` 是预处理
使用#{}可以有效的防止SQL注入，
## MyBatis架构
核心成员说明
| 核心成员 | 功能说明 |
| ---------------- | ------------------------------------------------------------ |
| Configuration | 保存MyBatis大部分配置信息 |
| SqlSession | MyBatis主要的顶层API， ，
| Executor | MyBatis 调度器，
| StatementHandler | 封装JDBC，
| ParameterHandler | 用户传递的参数转换成JDBC Statement 所对应的数据类型 |
| ResultSetHandler | 负责将JDBC返回的ResultSet结果集对象转换成List类型的集合 |
| TypeHandler | 负责java数据类型和jdbc数据类型(也可以说是数据表列类型)之间的映射和转换 |
| MappedStatement | MappedStatement维护一条< select \|update \|delete \|insert > 节点的封装 |
| SqlSource | 负责根据用户传递的parameterObject， ， ，
| BoundSql | 表示动态生成的SQL语句以及相应的参数信息 |
## MyBatis流程
## Dao接口工作原理
Mapper 接口的工作原理是JDK动态代理， ， ， ， ，
Mapper接口里的方法， ，
## MyBatis缓存
MyBatis 中有一级缓存和二级缓存，默认情况下一级缓存是开启的，而且是不能关闭的。一级缓存是指 SqlSession 级别的缓存，当在同一个 SqlSession 中进行相同的 SQL 语句查询时，第二次以后的查询不会从数据库查询，而是直接从缓存中获取，一级缓存最多缓存 1024 条 SQL。二级缓存是指可以跨 SqlSession 的缓存。是 mapper 级别的缓存， ，
### 一级缓存
Mybatis的一级缓存原理(sqlsession级别)。第一次发出一个查询 sql，
- key：
- value：
同一个 sqlsession 再次发出相同的 sql，
### 二级缓存
二级缓存的范围是 mapper 级别（ ） ，
- key：
**具体使用需要配置：**
- Mybatis 全局配置中启用二级缓存配置
- 在对应的 Mapper.xml 中配置 cache 节点
- 在对应的 select 查询节点中添加 useCache=true
## MyBatis主要组件
### SqlSessionFactoryBuilder
从名称长可以看出来使用的建造者设计模式（ ） ， ：
- 解析mybatis的xml配置文件， （ < configuration > 标签）
- 根据创建的Configuration对象， （ ）
### SqlSessionFactory
从名称上可以看出使用的工厂模式（ ） ， （ ）
- 调用openSession方法， ， （ ）
- openSession方法有多个重载，
### SqlSession
如果了解web开发， ， ， ， （ ） ， （ ） ， （ ， ， ， ， ）
SqlSession只是定义了执行sql的一些方法， ， ， （ ） ，
### Executor
Executor（ ） ， ， ：
CacheExecutor其实是一个Executor代理类， ， （ ， ， ，
Executor是对JDBC的封装。当我们使用JDBC来执行sql时， ， ， （ ） ， ，
### StatementHandler
在JDBC中， ， ， ， ，
- 对sql进行预处理
- 调用statement.executeXxx()执行sql
- 将数据库返回的结果集进行对象转换（ ）
### ParameterHandler
，
### ResultSetHandler
当执行statement.execute()后， ， ， （ ） ， ，
### TypeHandler
TypeHandler主要用在两个地方：
- 参数绑定，
MyBatis中， < resultMap > 来定义结果的映射关系， ， ， （ ）
- 获取结果集中的字段值，
# Nginx
## Nginx安装
```shell
# CentOS
yum install nginx;
# Ubuntu
sudo apt-get install nginx;
# Mac
brew install nginx;
```
安装完成后，通过 `rpm \-ql nginx` 命令查看 `Nginx` 的安装信息：
```shell
# Nginx配置文件
/etc/nginx/nginx.conf # nginx 主配置文件
/etc/nginx/nginx.conf.default
# 可执行程序文件
/usr/bin/nginx-upgrade
/usr/sbin/nginx
# nginx库文件
/usr/lib/systemd/system/nginx.service # 用于配置系统守护进程
/usr/lib64/nginx/modules # Nginx模块目录
# 帮助文档
/usr/share/doc/nginx-1.16.1
/usr/share/doc/nginx-1.16.1/CHANGES
/usr/share/doc/nginx-1.16.1/README
/usr/share/doc/nginx-1.16.1/README.dynamic
/usr/share/doc/nginx-1.16.1/UPGRADE-NOTES-1.6-to-1.10
# 静态资源目录
/usr/share/nginx/html/404.html
/usr/share/nginx/html/50x.html
/usr/share/nginx/html/index.html
# 存放Nginx日志文件
/var/log/nginx
```
主要关注的文件夹有两个：
- `/etc/nginx/conf.d/` 是子配置项存放处， `/etc/nginx/nginx.conf` 主配置文件会默认把该文件夹中所有子配置项都引入
- `/usr/share/nginx/html/` 静态文件都放在这个文件夹，也可以根据你自己的习惯放在其他地方
## 运维命令
一般可以在 `/etc/nginx/nginx.conf` 中配置。
### systemctl命令
```shell
# 开机配置
# 开机自动启动
systemctl enable nginx
# 关闭开机自动启动
systemctl disable nginx
# 启动Nginx
systemctl start nginx # 启动Nginx成功后， ，
# 停止Nginx
systemctl stop nginx
# 重启Nginx
systemctl restart nginx
# 重新加载Nginx
systemctl reload nginx
# 查看 Nginx 运行状态
systemctl status nginx
# 查看Nginx进程
ps -ef | grep nginx
# 杀死Nginx进程
kill -9 pid # 根据上面查看到的Nginx进程号， ，
```
### Nginx应用命令
```shell
# 启动
nginx -s start
# 向主进程发送信号，重新加载配置文件，热重启
nginx -s reload
# 重启 Nginx
nginx -s reopen
# 快速关闭
nginx -s stop
# 等待工作进程处理完成后关闭
nginx -s quit
# 查看当前 Nginx 最终的配置
nginx -T
# 检查配置是否有问题
nginx -t
```
## 配置规则
Nginx层级结构如下：
### URI匹配
```shell
location = / {
# 完全匹配 =
# 大小写敏感 ~
# 忽略大小写 ~*
}
location ^~ /images/ {
# 前半部分匹配 ^~
# 可以使用正则，如：
# location ~* \.(gif|jpg|png)$ { }
}
location / {
# 如果以上都未匹配，会进入这里
}
```
### upstream
```nginx
语法：
...
}
上下文：
示例：
upstream back_end_server{
server 192.168.100.33:8081
}
```
`upstream` 用于定义上游服务器（指的就是后台提供的应用服务器）的相关信息。
```nginx
upstream test_server{
# weight=3 权重值，
# max_conns=1000 上游服务器的最大并发连接数
# fail_timeout=10s 服务器不可用的判定时间
# max_fails=2 服务器不可用的检查次数
# backup 备份服务器，仅当其他服务器都不可用时才会启用
# down 标记服务器长期不可用，离线维护；
server 127.0.0.1:8081 weight=3 max_conns=1000 fail_timeout=10s max_fails=2;
# 限制每个worker子进程与上游服务器空闲长连接的最大数量
keepalive 16;
# 单个长连接可以处理的最多 HTTP 请求个数
keepalive_requests 100;
# 空闲长连接的最长保持时间
keepalive_timeout 60s;
}
```
### proxy_pass
```nginx
语法：
上下文：
示例：
proxy_pass http://127.0.0.1:8081
proxy_pass http://127.0.0.1:8081/proxy
```
`URL` 参数原则：
- `URL` 必须以 `http` 或 `https` 开头
- `URL` 中可以携带变量
- `URL` 中是否带 `URI` ，会直接影响发往上游请求的 `URL`
`URL` 后缀带 `/` 和不带 `/` 的区别：
- 不带 `/` 意味着 `Nginx` 不会修改用户 `URL` ，而是直接透传给上游的应用服务器
- 带 `/` 意味着 `Nginx` 会修改用户 `URL` ，修改方法是将 `location` 后的 `URL` 从用户 `URL` 中删除
```nginx
# 用户请求 URL ：
location /bbs/{
# 请求到达上游应用服务器的 URL ：
proxy_pass http://127.0.0.1:8080;
# 请求到达上游应用服务器的 URL ：
proxy_pass http://127.0.0.1:8080/;
}
```
## 应用场景
### 反向代理
```nginx
# 1 /etc/nginx/conf.d/proxy.conf
# 1.1 模拟被代理服务
server{
listen 8080;
server_name localhost;
location /proxy/ {
root /usr/share/nginx/html/proxy;
index index.html;
}
}
# 1.2 /usr/share/nginx/html/proxy/index.html
< h1 > 121.42.11.34 proxy html < / h1 >
# 2 /etc/nginx/conf.d/proxy.conf
# 2.1 back server
upstream back_end {
server 121.42.11.34:8080 weight=2 max_conns=1000 fail_timeout=10s max_fails=3;
keepalive 32;
keepalive_requests 80;
keepalive_timeout 20s;
}
# 2.2 代理配置
server {
listen 80;
# vim /etc/hosts进入配置文件,添加如下内容：
server_name proxy.lion.club;
location /proxy {
proxy_pass http://back_end/proxy;
}
}
```
### 负载均衡
```nginx
# 1 /etc/nginx/conf.d/balance.conf
# 1.1 模拟被代理服务1
server{
listen 8020;
location / {
return 200 'return 8020 \n';
}
}
# 1.2 模拟被代理服务2
server{
listen 8030;
location / {
return 200 'return 8030 \n';
}
}
# 1.3 模拟被代理服务3
server{
listen 8040;
location / {
return 200 'return 8040 \n';
}
}
# 2 /etc/nginx/conf.d/balance.conf
# 2.1 demo server list
upstream demo_server {
server 121.42.11.34:8020;
server 121.42.11.34:8030;
server 121.42.11.34:8040;
}
# 2.2 代理配置
server {
listen 80;
server_name balance.lion.club;
location /balance/ {
proxy_pass http://demo_server;
}
}
```
#### hash算法
```nginx
upstream demo_server {
hash $request_uri;
server 121.42.11.34:8020;
server 121.42.11.34:8030;
server 121.42.11.34:8040;
}
server {
listen 80;
server_name balance.lion.club;
location /balance/ {
proxy_pass http://demo_server;
}
}
```
`hash $request_uri` 表示使用 `request_uri` 变量作为 `hash` 的 `key` 值，只要访问的 `URI` 保持不变，就会一直分发给同一台服务器。
#### ip_hash
```nginx
upstream demo_server {
ip_hash;
server 121.42.11.34:8020;
server 121.42.11.34:8030;
server 121.42.11.34:8040;
}
server {
listen 80;
server_name balance.lion.club;
location /balance/ {
proxy_pass http://demo_server;
}
}
```
根据客户端的请求 `ip` 进行判断，只要 `ip` 地址不变就永远分配到同一台主机。它可以有效解决后台服务器 `session` 保持的问题。
#### 最少连接数算法
```nginx
upstream demo_server {
zone test 10M; # zone可以设置共享内存空间的名字和大小
least_conn;
server 121.42.11.34:8020;
server 121.42.11.34:8030;
server 121.42.11.34:8040;
}
server {
listen 80;
server_name balance.lion.club;
location /balance/ {
proxy_pass http://demo_server;
}
}
```
各个 `worker` 子进程通过读取共享内存的数据，来获取后端服务器的信息。来挑选一台当前已建立连接数最少的服务器进行分配请求。
### 配置缓存
**① proxy_cache**
存储一些之前被访问过、而且可能将要被再次访问的资源，使用户可以直接从代理服务器获得，从而减少上游服务器的压力，加快整个访问速度。
```nginx
语法：
默认值：
上下文：
```
**② proxy_cache_path**
设置缓存文件的存放路径。
```nginx
语法：
默认值：
上下文：
```
参数含义：
- `path` 缓存文件的存放路径
- `level path` 的目录层级
- `keys_zone` 设置共享内存
- `inactive` 在指定时间内没有被访问， ，
**③ proxy_cache_key**
设置缓存文件的 `key` 。
```nginx
语法：
默认值：
上下文：
```
**④ proxy_cache_valid**
配置什么状态码可以被缓存，以及缓存时长。
```nginx
语法：
上下文：
配置示例：
```
**⑤ proxy_no_cache**
定义相应保存到缓存的条件，如果字符串参数的至少一个值不为空且不等于“ 0”，
```nginx
语法：
上下文：
示例：
```
**⑥ proxy_cache_bypass**
定义条件，在该条件下将不会从缓存中获取响应。
```nginx
语法：
上下文：
示例：
```
**⑦ upstream_cache_status 变量**
它存储了缓存是否命中的信息，会设置在响应头信息中，在调试中非常有用。
```nginx
MISS: 未命中缓存
HIT：
EXPIRED: 缓存过期
STALE: 命中了陈旧缓存
REVALIDDATED: Nginx验证陈旧缓存依然有效
UPDATING: 内容陈旧，但正在更新
BYPASS: X响应从原始服务器获取
```
```nginx
proxy_cache_path /etc/nginx/cache_temp levels=2:2 keys_zone=cache_zone:30m max_size=2g inactive=60m use_temp_path=off;
upstream cache_server{
server 121.42.11.34:1010;
server 121.42.11.34:1020;
}
server {
listen 80;
server_name cache.lion.club;
# 场景一：实时性要求不高，则配置缓存
location /demo {
proxy_cache cache_zone; # 设置缓存内存，上面配置中已经定义好的
proxy_cache_valid 200 5m; # 缓存状态为200的请求，
proxy_cache_key $request_uri; # 缓存文件的key为请求的URI
add_header Nginx-Cache-Status $upstream_cache_status # 把缓存状态设置为头部信息，响应给客户端
proxy_pass http://cache_server; # 代理转发
}
# 场景二：实时性要求非常高，则配置不缓存
# URI 中后缀为 .txt 或 .text 的设置变量值为 "no cache"
if ($request_uri ~ \.(txt|text)$) {
set $cache_name "no cache"
}
location /test {
proxy_no_cache $cache_name; # 判断该变量是否有值，如果有值则不进行缓存，如果没有值则进行缓存
proxy_cache cache_zone; # 设置缓存内存
proxy_cache_valid 200 5m; # 缓存状态为200的请求，
proxy_cache_key $request_uri; # 缓存文件的key为请求的URI
add_header Nginx-Cache-Status $upstream_cache_status # 把缓存状态设置为头部信息，响应给客户端
proxy_pass http://cache_server; # 代理转发
}
}
```
### HTTPS
下载证书的压缩文件，里面有个 `Nginx` 文件夹，把 `xxx.crt` 和 `xxx.key` 文件拷贝到服务器目录，再进行如下配置：
```nginx
server {
listen 443 ssl http2 default_server; # SSL 访问端口号为 443
server_name lion.club; # 填写绑定证书的域名(我这里是随便写的)
ssl_certificate /etc/nginx/https/lion.club_bundle.crt; # 证书地址
ssl_certificate_key /etc/nginx/https/lion.club.key; # 私钥地址
ssl_session_timeout 10m;
ssl_protocols TLSv1 TLSv1.1 TLSv1.2; # 支持ssl协议版本， ，
location / {
root /usr/share/nginx/html;
index index.html index.htm;
}
}
```
### 开启gzip压缩
在 `/etc/nginx/conf.d/` 文件夹中新建配置文件 `gzip.conf` ：
```nginx
# 默认off，
gzip on;
# 要采用 gzip 压缩的 MIME 文件类型，其中 text/html 被系统强制启用
gzip_types text/plain text/css application/json application/x-javascript text/xml application/xml application/xml+rss text/javascript;
# ---- 以上两个参数开启就可以支持Gzip压缩了 ----
# 默认 off， ，
gzip_static on;
# 默认 off， ，
gzip_proxied any;
# 用于在响应消息头中添加Vary： ，
gzip_vary on;
# gzip 压缩比，压缩级别是 1-9， ，
gzip_comp_level 6;
# 获取多少内存用于缓存压缩结果，
gzip_buffers 16 8k;
# 允许压缩的页面最小字节数， ，
gzip_min_length 1k;
# 默认 1.1，启用 gzip 所需的 HTTP 最低版本
gzip_http_version 1.1;
```
## 常用配置
### 侦听端口
```nginx
server {
# Standard HTTP Protocol
listen 80;
# Standard HTTPS Protocol
listen 443 ssl;
# For http2
listen 443 ssl http2;
# Listen on 80 using IPv6
listen [::]:80;
# Listen only on using IPv6
listen [::]:80 ipv6only=on;
}
```
### 访问日志
```nginx
server {
# Relative or full path to log file
access_log /path/to/file.log;
# Turn 'on' or 'off'
access_log on;
}
```
### 域名
```nginx
server {
# Listen to yourdomain.com
server_name yourdomain.com;
# Listen to multiple domains server_name yourdomain.com www.yourdomain.com;
# Listen to all domains
server_name *.yourdomain.com;
# Listen to all top-level domains
server_name yourdomain.*;
# Listen to unspecified Hostnames (Listens to IP address itself)
server_name "";
}
```
### 静态资源
```nginx
server {
listen 80;
server_name yourdomain.com;
location / {
root /path/to/website;
}
}
```
### 重定向
```nginx
server {
listen 80;
server_name www.yourdomain.com;
return 301 http://yourdomain.com$request_uri;
}
server {
listen 80;
server_name www.yourdomain.com;
location /redirect-url {
return 301 http://otherdomain.com;
}
}
```
### 反向代理
```nginx
server {
listen 80;
server_name yourdomain.com;
location / {
proxy_pass http://0.0.0.0:3000;
# where 0.0.0.0:3000 is your application server (Ex: node.js) bound on 0.0.0.0 listening on port 3000
}
}
```
### 负载均衡
```nginx
upstream node_js {
server 0.0.0.0:3000;
server 0.0.0.0:4000;
server 123.131.121.122;
}
server {
listen 80;
server_name yourdomain.com;
location / {
proxy_pass http://node_js;
}
}
```
### SSL协议
```nginx
server {
listen 443 ssl;
server_name yourdomain.com;
ssl on;
ssl_certificate /path/to/cert.pem;
ssl_certificate_key /path/to/privatekey.pem;
ssl_stapling on;
ssl_stapling_verify on;
ssl_trusted_certificate /path/to/fullchain.pem;
ssl_protocols TLSv1 TLSv1.1 TLSv1.2;
ssl_session_timeout 1h;
ssl_session_cache shared:SSL:50m;
add_header Strict-Transport-Security max-age=15768000;
}
# Permanent Redirect for HTTP to HTTPS
server {
listen 80;
server_name yourdomain.com;
return 301 https://$host$request_uri;
}
```
## 日志分析
```shell
# 统计IP访问量
awk '{print $1}' access.log | sort -n | uniq | wc -l
# 查看某一时间段的IP访问量(4-5点)
grep "07/Apr/2017:0[4-5]" access.log | awk '{print $1}' | sort | uniq -c| sort -nr | wc -l
# 查看访问最频繁的前100个IP
awk '{print $1}' access.log | sort -n |uniq -c | sort -rn | head -n 100
# 查看访问100次以上的IP
awk '{print $1}' access.log | sort -n |uniq -c |awk '{if($1 >100) print $0}'|sort -rn
# 查询某个IP的详细访问情况,按访问频率排序
grep '104.217.108.66' access.log |awk '{print $7}'|sort |uniq -c |sort -rn |head -n 100
# 页面访问统计
# 查看访问最频的页面(TOP100)
awk '{print $7}' access.log | sort |uniq -c | sort -rn | head -n 100
# 查看访问最频的页面([排除php页面】(TOP100)
grep -v ".php" access.log | awk '{print $7}' | sort |uniq -c | sort -rn | head -n 100
# 查看页面访问次数超过100次的页面
cat access.log | cut -d ' ' -f 7 | sort |uniq -c | awk '{if ($1 > 100) print $0}' | less
# 查看最近1000条记录，
tail -1000 access.log |awk '{print $7}'|sort|uniq -c|sort -nr|less
# 请求量统计
# 统计每秒的请求数,top100的时间点(精确到秒)
awk '{print $4}' access.log |cut -c 14-21|sort|uniq -c|sort -nr|head -n 100
# 统计每分钟的请求数,top100的时间点(精确到分钟)
awk '{print $4}' access.log |cut -c 14-18|sort|uniq -c|sort -nr|head -n 100
# 统计每小时的请求数,top100的时间点(精确到小时)
awk '{print $4}' access.log |cut -c 14-15|sort|uniq -c|sort -nr|head -n 100
# 性能分析
# 列出传输时间超过 3 秒的页面，
cat access.log|awk '($NF > 3){print $7}'|sort -n|uniq -c|sort -nr|head -20
# 列出php页面请求时间超过3秒的页面， ，
cat access.log|awk '($NF > 1 && $7~/\.php/){print $7}'|sort -n|uniq -c|sort -nr|head -100
# 蜘蛛抓取统计
# 统计蜘蛛抓取次数
grep 'Baiduspider' access.log |wc -l
# 统计蜘蛛抓取404的次数
grep 'Baiduspider' access.log |grep '404' | wc -l
# TCP连接统计
# 查看当前TCP连接数
netstat -tan | grep "ESTABLISHED" | grep ":80" | wc -l
# 用tcpdump嗅探80端口的访问看看谁最高
tcpdump -i eth0 -tnn dst port 80 -c 1000 | awk -F"." '{print $1"."$2"."$3"."$4}' | sort | uniq -c | sort -nr
awk '{print $1}' $logpath |sort -n|uniq|wc -l
# 系统正在统计某一个时间段IP访问量为
sed -n '/22\/Jun\/2017:1[5]/,/22\/Jun\/2017:1[6]/p' $logpath|awk '{print $1}'|sort -n|uniq|wc -l
# 访问100次以上的IP
awk '{print $1}' $logpath|sort -n|uniq -c|awk '{if($1>100) print $0}'|sort -rn
# 访问最频繁的请求(TOP50)
awk '{print $7}' $logpath |sort |uniq -c|sort -rn |head -n 50
# 统计每秒的请求数(TOP50)
awk '{print $4}' $logpath|cut -c 14-21|sort |uniq -c|sort -nr|head -n 50
# 统计每分钟的请求数(TOP50)
awk '{print $4}' $logpath|cut -c 14-18|sort|uniq -c|sort -nr|head -n 50
# 统计每小时请求数(TOP50)
awk '{print $4}' $logpath|cut -c 14-15|sort|uniq -c|sort -nr|head -n 50
# 传输时间超过1秒的请求(TOP20)
cat $logpath|awk '($NF > 1){print $7}'|sort -n|uniq -c|sort -nr|head -20
```
# LVS
LVS是Linux Virtual Server的简写， ，
## 工作模式
LVS有三种常见的负载均衡的模式： （ ） （ ）
**术语**
- **DS**： ，
- **RS**： ，
- **VIP**： ， ，
- **DIP**： ，
- **RIP**： ，
- **CIP**： ，
(假设 cip 是200.200.200.2， ）
### NAT模式(网络地址转换)
**NAT模式(NetWork Address Translation-网络地址转换)**。客户发出请求， ， ，
**原理**
- client发出请求（ ， ）
- 请求包到达lvs， （ ， ）
- 请求包到达rs， （ ， ）
- 这个回复包不能直接给client，
- 但是因为lvs是网关， ，
- 网关修改sip为VIP， （ ， ）
**优点**
- 配置简单
- 支持端口映射（看名字就知道）
- RIP一般是私有地址，
**缺点**
- LVS和所有RS必须在同一个vlan
- 进出流量都要走LVS转发
- LVS容易成为瓶颈
- 一般而言需要将VIP配置成RS的网关
**分析**
- **为什么NAT要求lvs和RS在同一个vlan？
因为回复包必须经过lvs再次修改sip为vip， ， （ ） ， ， ， ，
**NAT结构**
- LVS修改进出包的(sip, dip)的时只改了其中一个（ ）
- NAT最大的缺点是要求LVS和RS必须在同一个vlan，
### IP TUN模型(IP隧道)
**IP TUN模型(IP Tunneling-IP隧道)**。和DR模式差不多， ， ， ， ， ，
**原理**
- 请求包到达LVS后，
- 新的IP包的目的IP是某一RS的IP，
- RS收到报文后IPIP内核模块解封装，
- 发现目的IP是VIP， ，
**优点**
- 集群节点可以跨vlan
- 跟DR一样，
**缺点**
- RS上必须安装运行IPIP模块
- 多增加了一个IP头
- LVS和RS上的tunl0虚拟网卡上配置同一个VIP（ ）
**分析**
- **为什么IP TUN不要求同一个vlan？
因为IP TUN中不是修改MAC来路由， ，
- **IP TUN性能**
回包不走LVS， ，
**IP TUN结构**
- 图中红线是再次封装过的包，
- DR可能在小公司用的比较多，
### DR模式(直接路由)
**DR模式(Director Routing-直接路由)**。整个DR模式都是停留在第二层的数据链路层。直接修改MAC。实现报文的转发。
**原理**
- 请求流量(sip 200.200.200.2, dip 200.200.200.1) 先到达LVS
- 然后LVS， ，
- 然后丢给交换机，
- 选中的RS看到MAC地址是自己的、dip也是自己的，
- 回复包（ ， ）
- 经过交换机直接回复给client了（ ）
**优点**
- DR模式是性能最好的一种模式， ，
**缺点**
- 要求LVS和rs在同一个vlan
- RS需要配置vip同时特殊处理arp
- 不支持端口映射
**分析**
- **为什么要求LVS和RS在同一个vlan（ ） ？
因为DR模式依赖多个RS和LVS共用同一个VIP， ，
- **DR 模式为什么性能最好？**
因为回复包不走LVS了， ， ， ，
- **DR 模式为什么回包不需要走LVS了？
因为RS和LVS共享同一个vip， ， （ ）
**DR结构**
- 绿色是请求包进来，
### full NAT模式
**full NAT模式(full NetWork Address Translation-全部网络地址转换)**。
**原理（ ）
- client发出请求（ ）
- 请求包到达lvs， （ ， ）
- 请求包到达rs， （ ， ）
- 这个回复包的目的IP是VIP(不像NAT中是 cip)，
- lvs修改sip为vip， ， （ ， ）
**优点**
- 解决了NAT对LVS和RS要求在同一个vlan的问题，
**缺点**
- RS看不到cip（ ）
- 进出流量还是都走的lvs， （ ）
**分析**
- **为什么full NAT解决了NAT中要求的LVS和RS必须在同一个vlan的问题？
因为LVS修改进来的包的时候把(sip, dip)都修改了(这也是full的主要含义吧)， （ ） ， ， ， ，
- **为什么full NAT后RS看不见cip了？
因为cip被修改掉了， ， ， ，
**full NAT结构**
- full NAT解决了NAT的同vlan的要求，
- 但还没解决进出流量都走LVS的问题（ ）
### ENAT模式
**ENAT模式(enhence NAT)**。ENAT模式在内部也会被称为 三角模式或者DNAT/SNAT模式。
**原理**
- client发出请求（ ， ）
- 请求包到达lvs， （ ， ） ，
- 请求包根据ip路由到达rs，
- 回复包(RIP, vip)被ctk模块截获， （
- 因为回复包的目的地址是cip所以不需要经过lvs，
**优点**
- 不要求LVS和RS在同一个vlan
- 出去的流量不需要走LVS，
**缺点**
- 集团实现的自定义方案， （ ）
**分析**
- **为什么ENAT的回复包不需要走回LVS了？
因为full NAT模式下要走回去是需要LVS再次改写回复包的IP， ，
- **为什么ENAT的LVS和RS可以在不同的vlan？
跟full NAT一样。
**ENAT结构**
## 调度算法
### 静态调度算法
只根据算法进行调度 而不考虑后端服务器的实际连接情况和负载情况。
- **RR： （ ）
调度器通过”轮叫”调度算法将外部请求按顺序轮流分配到集群中的真实服务器上，它均等地对待每一台服务器，而不管服务器上实际的连接数和系统负载｡
- **WRR： （ ）
调度器通过“加权轮叫”调度算法根据真实服务器的不同处理能力来调度访问请求。这样可以保证处理能力强的服务器处理更多的访问流量。调度器可以自动问询真实服务器的负载情况,并动态地调整其权值。
- **DH： （ ）
根据请求的目标IP地址，
- **SH： （ ）
源地址散列”调度算法根据请求的源IP地址，
### 动态调度算法
- **LC： （ ）
调度器通过”最少连接”调度算法动态地将网络请求调度到已建立的链接数最少的服务器上。如果集群系统的真实服务器具有相近的系统性能，采用”最小连接”调度算法可以较好地均衡负载。
- **WLC： （ ）
在集群系统中的服务器性能差异较大的情况下，调度器采用“加权最少链接”调度算法优化负载均衡性能，具有较高权值的服务器将承受较大比例的活动连接负载｡调度器可以自动问询真实服务器的负载情况,并动态地调整其权值。
- **SED： （ ）
在WLC基础上改进， （ ） ， ， ， ， ： ，
- **NQ永不排队/最少队列调度（ ）
无需队列。如果有台 realserver的连接数＝ ， ， ， ， ， ， ， ， ，
- **LBLC： （ ）
基于局部性的最少链接”调度算法是针对目标IP地址的负载均衡， ， ，
- **LBLCR： （ ）
带复制的基于局部性最少链接”调度算法也是针对目标IP地址的负载均衡， ， ， ， ， ； ， ， ， ， ， ，
## 常见问题分析
### 脑裂
由于两台高可用服务器对之间， ， ， ， ， ， ， ， ，
**1）
- 心跳链路故障，导致无法通信
- 开启防火墙阻挡心跳消息传输
- 心跳网卡地址配置等不正确
- 其他： ， ，
备注：
- 心跳线坏了（故障或老化）
- 网卡相关驱动坏了， （ ）
- 心跳线间连接的设备故障（网卡及交换机）
- 仲裁机器出问题
**2）
- 采用串行或以太网电缆连接，同时用两条心跳线路
- 做好裂脑的监控报警，在问题发生时人为第一时间介入仲裁
- 启用磁盘锁，即正在服务的一方只在发现心跳线全部断开时，才开启磁盘锁
- fence设备（ ）
- 增加仲裁盘
- 加冗余线路
### 负载不均
**原因分析**
- lvs自身的会话保持参数设置。优化：
- lvs调度算法设置，
- 后端RS节点的会话保持参数，
- 访问量较少的情况下，不均衡的现象更加明显
- 用户发送的请求时间长短和请求资源多少以及大小
### 排错
先检查客户端到服务端——>然后检查负载均衡到RS端——>最后检查客户端到LVS端。
- 调度器上lvs调度规则及IP正确性
- RS节点上VIP和ARP抑制的检查
**生成思路**
- 对绑定的VIP做实时监控，
- 把绑定的VIP做成配置文件，
**ARP抑制的配置思路**
- 如果是单个VIP，
- 如果RS端有多个VIP绑定， ，
- RS节点上自身提供服务的检查
- 辅助排除工具有tcpdump、ping等
- 负载均衡和反向代理三角形排查理论
# Keepalived
Keepalived是一个免费开源的， & 7交换机制软件， （ ） （ ） ， （ ） ，
## 功能体系
Keepalived的所有功能是配置keepalived.conf文件来实现的。上图是Keepalived的功能体系结构， ：
- **内核空间（ ）
主要包括IPVS（ ， ） （ ）
- **用户空间（ ）
- WatchDog：
- VRRP Stack： ， ，
- Checkers： ， ， ，
- IPVS wrapper：
- Netlink Reflector：
**重要功能**
- 管理LVS负载均衡软件
- 实现LVS集群节点的健康检查中
- 作为系统网络服务的高可用性（ ）
**高可用故障切换转移原理**
Keepalived高可用服务对之间的故障切换转移，
在 Keepalived服务正常工作时， （ ） ， ， ， ， ， ， ， ，
**VRRP**
全称Virtual Router Redundancy Protocol ,中文名为虚拟路由冗余协议 ， ，
## 脑裂问题
在高可用（ ） ， ， ， ， ， ， ； ， ， （ ）
对付HA系统“裂脑”的对策， ：
- 添加冗余的心跳线， ： （ ） ，
- 启用磁盘锁。正在服务一方锁住共享磁盘， ， ， ， ， ： （ ）
- 设置仲裁机制。例如设置参考IP（ ） ， ， ， ， （ ） ， ， ， ，
### 产生原因
一般来说，裂脑的发生，有以下几种原因：
- 高可用服务器对之间心跳线链路发生故障，导致无法正常通信
- 因心跳线坏了（包括断了，老化）
- 因网卡及相关驱动坏了， （ ）
- 因心跳线间连接的设备故障（网卡及交换机）
- 因仲裁的机器出问题（采用仲裁的方案）
- 高可用服务器上开启了 iptables防火墙阻挡了心跳消息传输
- 高可用服务器上心跳网卡地址等信息配置不正确，导致发送心跳失败
- 其他服务配置不当等原因， ，
**提示：** Keepalived配置里同一 VRRP实例如果 virtual_router_id两端参数配置不一致也会导致裂脑问题发生。
### 解决方案
在实际生产环境中，我们可以从以下几个方面来防止裂脑问题的发生：
- 同时使用串行电缆和以太网电缆连接，同时用两条心跳线路，这样一条线路坏了，另一个还是好的，依然能传送心跳消息
- 当检测到裂脑时强行关闭一个心跳节点（ ， ） ，
- 做好对裂脑的监控报警（如邮件及手机短信等或值班）.在问题发生时人为第一时间介入仲裁，降低损失。例如，百度的监控报警短倍就有上行和下行的区别。报警消息发送到管理员手机上，管理员可以通过手机回复对应数字或简单的字符串操作返回给服务器.让服务器根据指令自动处理相应故障，这样解决故障的时间更短.
在实施高可用方案时，要根据业务实际需求确定是否能容忍这样的损失。对于一般的网站常规业务，这个损失是可容忍的。
## 安装部署
**第一步：
`yum install keepalived -y `
```
/etc/keepalived
/etc/keepalived/keepalived.conf #keepalived 服务主配置文件
/etc/rc.d/init.d/keepalived #服务启动脚本
/etc/sysconfig/keepalived
/usr/bin/genhash
/usr/libexec/keepalived
/usr/sbin/keepalived
```
**第二步：配置文件说明**
全局配置
```shell
global_defs { # 全局配置
notification_email { # 定义报警邮件地址
acassen@firewall.loc
failover@firewall.loc
sysadmin@firewall.loc
}
notification_email_from Alexandre.Cassen@firewall.loc # 定义发送邮件的地址
smtp_server 192.168.200.1 # 邮箱服务器
smtp_connect_timeout 30 # 定义超时时间
router_id LVS_DEVEL # 定义路由标识信息，相同局域网唯一
}
```
虚拟ip配置brrp
```shell
vrrp_instance VI_1 { # 定义实例
state MASTER # 状态参数 master/backup 只是说明
interface eth0 # 虚IP地址放置的网卡位置
virtual_router_id 51 # 同一家族要一直，
priority 100 # 优先级决定是主还是备 越大越优先
advert_int 1 # 主备通讯时间间隔
authentication {
auth_type PASS
auth_pass 1111 # 认证
}
virtual_ipaddress {
192.168.200.16 # 设备之间使用的虚拟ip地址
192.168.200.17
192.168.200.18
}
}
```
**第三步：最终配置文件**
主负载均衡服务器配置
```shell
[root@lb01 conf]# cat /etc/keepalived/keepalived.conf
! Configuration File for keepalived
global_defs {
router_id lb01
}
vrrp_instance VI_1 {
state MASTER
interface eth0
virtual_router_id 51
priority 150
advert_int 1
authentication {
auth_type PASS
auth_pass 1111
}
virtual_ipaddress {
10.0.0.3
}
}
```
备负载均衡服务器配置
```shell
[root@lb02 ~]# cat /etc/keepalived/keepalived.conf
! Configuration File for keepalived
global_defs {
router_id lb02
}
vrrp_instance VI_1 {
state BACKUP
interface eth0
virtual_router_id 51
priority 100
advert_int 1
authentication {
auth_type PASS
auth_pass 1111
}
virtual_ipaddress {
10.0.0.3
}
}
```
**第四步：
```shell
[root@lb02 ~]# /etc/init.d/keepalived start
Starting keepalived: [ OK ]
```
**第五步：在进行访问测试之前要保证后端的节点都能够单独的访问**
测试连通性. 后端节点
```shell
[root@lb01 conf]# curl -H host:www.etiantian.org 10.0.0.8
web01 www
[root@lb01 conf]# curl -H host:www.etiantian.org 10.0.0.7
web02 www
[root@lb01 conf]# curl -H host:www.etiantian.org 10.0.0.9
web03 www
[root@lb01 conf]# curl -H host:bbs.etiantian.org 10.0.0.9
web03 bbs
[root@lb01 conf]# curl -H host:bbs.etiantian.org 10.0.0.8
web01 bbs
[root@lb01 conf]# curl -H host:bbs.etiantian.org 10.0.0.7
web02 bbs
```
**第六步：
```shell
[root@lb01 conf]# ip a
1: lo: < LOOPBACK , UP , LOWER_UP > mtu 65536 qdisc noqueue state UNKNOWN
link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
inet 127.0.0.1/8 scope host lo
inet6 ::1/128 scope host
valid_lft forever preferred_lft forever
2: eth0: < BROADCAST , MULTICAST , UP , LOWER_UP > mtu 1500 qdisc pfifo_fast state UP qlen 1000
link/ether 00:0c:29:90:7f:0d brd ff:ff:ff:ff:ff:ff
inet 10.0.0.5/24 brd 10.0.0.255 scope global eth0
inet 10.0.0.3/24 scope global secondary eth0:1
inet6 fe80::20c:29ff:fe90:7f0d/64 scope link
valid_lft forever preferred_lft forever
3: eth1: < BROADCAST , MULTICAST , UP , LOWER_UP > mtu 1500 qdisc pfifo_fast state UP qlen 1000
link/ether 00:0c:29:90:7f:17 brd ff:ff:ff:ff:ff:ff
inet 172.16.1.5/24 brd 172.16.1.255 scope global eth1
inet6 fe80::20c:29ff:fe90:7f17/64 scope link
valid_lft forever preferred_lft forever
```
**第七步：【总结】配置文件修改**
Keepalived主备配置文件区别：
- router_id 信息不一致
- state 状态描述信息不一致
- priority 主备竞选优先级数值不一致
# HAProxy
HAProxy提供高可用性、负载均衡以及基于TCP和HTTP应用的代 理， ， ， ， ，
## 四层负载均衡
所谓的四层就是ISO参考模型中的第四层。四层负载均衡也称为四层交换机，
以常见的TCP应用为例， ， ， ， ， ， ， ， ，
## 七层负载均衡
七层负载均衡器也称为七层交换机， ， ， ， ， ， ， ， ， ， ， ， ， ，
这里仍以常见的TCP应用为例， ， ， ， ， ，
**四层和七层负载均衡**
对比四层负载均衡和七层负载均衡运行的整个过程， ， ， ， ， ， ，
## 负载均衡策略
- roundrobin：
- static-rr：
- leastconn：
- source： ，
- uri：
- url_param：
- rdp-cookie(name)：
**常用的负载均衡算法**
- 轮询算法：
- 根据请求源IP算法：
- 最少连接者先处理算法：
## HAProxy与LVS的区别
HAProxy负载均衡与LVS负载均衡的区别：
- 两者都是软件负载均衡产品， ，
- LVS是基于四层的IP负载均衡技术，
- LVS工作在ISO模型的第四层， ， ，
- HAProxy虽然功能强大， ，
综上所述， ， ， ，
## 安装配置
**第一步：安装依赖**
```shell
[root@test ~] # yum -y install make gcc gcc-c++ openssl-devel
```
**第二步：
```shell
[root@test ~] # wget http://haproxy.1wt.eu/download/1.3/src/haproxy-1.3.20.tar.gz
[root@test ~] # tar zcvf haproxy-1.3.20.tar.gz
[root@test ~] # cd haproxy-1.3.20
[root@test ~] # make TARGET=linux26 PREFIX=/usr/local/haproxy # 将haproxy安装到/usr/local/haproxy
[root@test ~] # make install PREFIX=/usr/local/haproxy
```
