微服务技术栈

Nacos安装与启动

Nacos服务的注册和发现

Nacos服务的分级存储

NacosRule负载均衡

Nacos环境隔离

“采用“配置集中管理”,可以很好的解决传统的“配置文件过于分散”的问题。所有的配置都集中在配置中心这一个地方管理,不需要每一个项目都自带一个,这样极大的减轻了开发成本。 采用“配置与应用分离”,可以很好的解决传统的“配置文件无法区分环境”的问题,配置并不跟着环境走,当不同环境有不同需求的时候,就到配置中心获取即可,极大的减轻了运维部署成本。 具备“实时更新”的功能,就是用来解决传统的“静态化配置”的问题。线上系统需要调整参数的时候,只需要在配置中心动态修改即可。

nacos统一配置原理

Nacos实现配置管理

基于feign的远程调用

自定义Feign的配置

feign的性能优化

feign的最佳实现 解耦合，提内聚

什么是FeignClient

ribbon负载均衡原理

ribbon负载均衡策略

配置负载均衡规则

参照服务的注册和发现中，NacosRlue负载均衡

1、分布式服务架构、微服务架构与 API 网关
    在微服务架构里，服务的粒度被进一步细分，各个业务服务可以被独立的设计、开发、测试、部署和管理。
这时，各个独立部署单元可以用不同的开发测试团队维护，可以使用不同的编程语言和技术平台进行设计，这就要求必须使用一种语言和平台无关的服务协议作为各个单元间的通讯方式。

2、API 网关的定义
    网关的角色是作为一个 API 架构，用来保护、增强和控制对于 API 服务的访问。
API 网关是一个处于应用程序或服务（提供 REST API 接口服务）之前的系统，用来管理授权、访问控制和流量限制等，这样 REST API 接口服务就被 API 网关保护起来，对所有的调用者透明。因此，隐藏在 API 网关后面的业务系统就可以专注于创建和管理服务，而不用去处理这些策略性的基础设施。

3、API网关职能
① 作为所有API接口服务请求的接入点
② 作为所有后端服务的聚合点
③ 实现安全、验证、路由、过滤、流控等策略
④ 对所有API服务和策略进行统一管理

4、API 网关的分类与功能
① 流量网关，关注稳定与安全
全局流控，日志统计，防止sql注入，防止web攻击，屏蔽工具扫描，黑白IP名单，证书/加密处理
② 业务网关，提供更好的服务
服务级别流控，服务降级与熔断，路由和负载均衡、灰度策略，服务过滤、聚合与发现，权限验证和用户等级策略，业务规则与参数校验，多级缓存策略

搭建网关服务

断言工厂

过滤器

DockerHub：
DockerHub是一个Docker镜像的托管平台。这样的平台称为Docker Registry。
国内也有类似于DockerHub的公开服务，比如 网易云镜像服务、阿里云镜像库等。

镜像（Image）：
Docker将应用程序及其所需的依赖、函数库、环境、配置等文件打包在一起，称为镜像。
容器（Container）：
镜像中的应用程序运行后形成的进程就是容器，只是Docker会给容器做隔离，对外不可见。

Docker是一个CS架构的程序，由两部分组成：

服务端(server)：Docker守护进程，负责处理Docker指令，管理镜像、容器等
客户端(client)：通过命令或RestAPI向Docker服务端发送指令。可以在本地或远程向服务端发送指

1.1.卸载

1.2.安装docker

1.3.启动docker

1.4.配置镜像加速

镜像操作命令

容器操作命令

为什么要使用数据卷

操作数据卷

挂载数据卷

镜像结构

Dockerfile

下载安装

Docker-Compose使用

什么是Docker-Compose

3.1.简化版镜像仓库
Docker官方的Docker Registry是一个基础版本的Docker镜像仓库，具备仓库管理的完整功能，但是没有图形化界面。

搭建方式比较简单，命令如下：

docker run -d \
    --restart=always \
    --name registry \
    -p 5000:5000 \
    -v registry-data:/var/lib/registry \
    registry
命令中挂载了一个数据卷registry-data到容器内的/var/lib/registry 目录，这是私有镜像库存放数据的目录。

访问http://YourIp:5000/v2/_catalog 可以查看当前私有镜像服务中包含的镜像

3.2.带有图形化界面版本
使用DockerCompose部署带有图象界面的DockerRegistry，命令如下：

version: '3.0'
services:
  registry:
    image: registry
    volumes:
      - ./registry-data:/var/lib/registry
  ui:
    image: joxit/docker-registry-ui:static
    ports:
      - 8080:80
    environment:
      - REGISTRY_TITLE=传智教育私有仓库
      - REGISTRY_URL=http://registry:5000
    depends_on:
      - registry
3.3.配置Docker信任地址
我们的私服采用的是http协议，默认不被Docker信任，所以需要做一个配置：

# 打开要修改的文件
vi /etc/docker/daemon.json
# 添加内容：
"insecure-registries":["http://192.168.150.101:8080"]
# 重加载
systemctl daemon-reload
# 重启docker
systemctl restart docker

推送镜像到私有镜像服务必须先tag，步骤如下：
① 重新tag本地镜像，名称前缀为私有仓库的地址：192.168.150.101:8080/
docker tag nginx:latest 192.168.150.101:8080/nginx:1.0
② 推送镜像
docker push 192.168.150.101:8080/nginx:1.0
③ 拉取镜像
docker pull 192.168.150.101:8080/nginx:1.0

优点：
时效性强，可以立即得到结果

缺点：
1、耦合度高
每次加入新的需求，都要修改原来的代码
2、性能下降
调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和。
3、资源浪费
调用链中的每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下会极度浪费系统资源
4、级联失败
如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障

优点：
• 吞吐量提升：无需等待订阅者处理完成，响应更快速
• 故障隔离：服务没有直接调用，不存在级联失败问题
• 调用间没有阻塞，不会造成无效的资源占用
• 耦合度极低，每个服务都可以灵活插拔，可替换
• 流量削峰：不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件

缺点：
• 依赖于Broker的可靠性、安全性、吞吐能力
• 架构复杂了，业务没有明显的流程线，不好追踪管理

1、拉取镜像
docker pull rabbitmq:3-management
2、创建运行容器
docker run \
-e RABBITMQ_DEFAULT_USER=itcast \
-e RABBITMQ_DEFAULT_PASS=123321 \
--name mq \
--hostname mq1 \
-p 15672:15672 \
-p 5672:5672 \
-d \
rabbitmq:3-management

提到RabbitMQ，就不得不提AMQP协议。AMQP协议是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。

先了解一下AMQP协议中间的几个重要概念：

Server：接收客户端的连接，实现AMQP实体服务。
Connection：连接，应用程序与Server的网络连接，TCP连接。
Channel：信道，消息读写等操作在信道中进行。客户端可以建立多个信道，每个信道代表一个会话任务。
Message：消息，应用程序和服务器之间传送的数据，消息可以非常简单，也可以很复杂。有Properties和Body组成。Properties为外包装，可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body就是消息体内容。
Virtual Host：虚拟主机，用于逻辑隔离。一个虚拟主机里面可以有若干个Exchange和Queue，同一个虚拟主机里面不能有相同名称的Exchange或Queue。
Exchange：交换器，接收消息，按照路由规则将消息路由到一个或者多个队列。如果路由不到，或者返回给生产者，或者直接丢弃。RabbitMQ常用的交换器常用类型有direct、topic、fanout、headers四种，后面详细介绍。
Binding：绑定，交换器和消息队列之间的虚拟连接，绑定中可以包含一个或者多个RoutingKey。
RoutingKey：路由键，生产者将消息发送给交换器的时候，会发送一个RoutingKey，用来指定路由规则，这样交换器就知道把消息发送到哪个队列。路由键通常为一个“.”分割的字符串，例如“com.rabbitmq”。
Queue：消息队列，用来保存消息，供消费者消费。
我们完全可以直接使用 Connection 就能完成信道的工作，为什么还要引入信道呢?

试想这样一个场景， 一个应用程序中有很多个线程需要从 RabbitMQ 中消费消息，或者生产消息，那么必然需要建立很多个 Connection，也就是许多个 TCP 连接。然而对于操作系统而言，建立和销毁 TCP 连接是非常昂贵的开销，如果遇到使用高峰，性能瓶颈也随之显现。 RabbitMQ 采用 TCP 连接复用的方式，不仅可以减少性能开销，同时也便于管理 。

RabbitMQ常用的交换器类型有direct、topic、fanout、headers四种。

Direct Exchange
该类型的交换器将所有发送到该交换器的消息被转发到RoutingKey指定的队列中，也就是说路由到BindingKey和RoutingKey完全匹配的队列中。

Topic Exchange
该类型的交换器将所有发送到Topic Exchange的消息被转发到所有RoutingKey中指定的Topic的队列上面。
Exchange将RoutingKey和某Topic进行模糊匹配，其中“”用来匹配一个词，“#”用于匹配一个或者多个词。例如“com.#”能匹配到“com.rabbitmq.oa”和“com.rabbitmq”；而"login."只能匹配到“com.rabbitmq”。

Fanout Exchange
该类型不处理路由键，会把所有发送到交换器的消息路由到所有绑定的队列中。优点是转发消息最快，性能最好。

Headers Exchange
该类型的交换器不依赖路由规则来路由消息，而是根据消息内容中的headers属性进行匹配。headers类型交换器性能差，在实际中并不常用。

一、基本消息队列的消息发送
流程：
1. 建立connection
// 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

2. 创建channel
        // 2.创建通道Channel
        Channel channel = connection.createChannel();

3. 利用channel声明队列
        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

4. 利用channel向队列发送消息
        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功：【" + message + "】");

        // 5.关闭通道和连接
        channel.close();
        connection.close();

二、基本消息队列的消息接收流程：
5. 建立connection
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();
6. 创建channel
    // 2.创建通道Channel
        Channel channel = connection.createChannel();
7. 利用channel声明队列
    // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);
8. 定义consumer的消费行为handleDelivery()
    // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息：【" + message + "】");
            }
        });
9. 利用channel将消费者与队列绑定

advanced Message Queuing Protocol：
是用于在应用程序或或之间传递业务消息的开放标准。该协议与语言平台无关，更符合微服务间独立的要求

Spring AMQP 是基于AMQP协议定义的一套API规范，提供了模板来发送接收消息。
包含两部分，其中Spring-amqp是基础抽像，spring-rabbit是底层的默认实现。

1、在父工程中引入 Spring AMQP依赖
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2、在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列
      ① 在publisher服务中编写application.yml，添加mq连接信息：
       spring:
          rabbitmq:
          host: 192.168.150.101  #主机名
          port: 5672  #端口
          virtual-host:/  #虚拟主机
          username: itcast#用户名
          password:123321  #密码
      ② 在publisher服务中新建一个测试类，编写测试方法：
           public classSpringAmqpTest {
                @Autowired
                 private RabbitTemplate;

                 public void testSimpleQueue() {
                     String queueName = "simple.queue";
                     String message = "hello, spring amqp!";
                     rabbitTemplate.convertAndSend(queueName, message);
          }
3、在consumer服务中编写消费逻辑，绑定simple.queue这个队列
      ① 在consumer服务中编写application.yml，添加mq连接信息：（同2①）
      ② 在consumer服务中新建一个类，编写消费逻辑：
           @Component
            public class SpringRabbitListener {
                @RabbitListener(queues ="simple.queue")
                 public void listenSimpleQueueMessage(String msg)throwsInterruptedException {
                       System.out.println("spring消费者接收到消息：【"+ msg +"】");
                 }
           }

修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限：
spring:
  rabbitmq:
    host: 192.168.150.101  #主机名
    port: 5672 #端口
    virtual-host: / #虚拟主机
    username: itcast #用户名
    password: 123321 #密码
    listener:
      simple:
        prefetch: 1  #每次只能获取一条消息，处理完成才能获取下一个消息

可以提高消息处理速度，避免队列消息堆积，其实是通过增加consumer 来加快消息的处理速度

FanoutExchange

DirectExchange

TopicExchange

消息转换器是做什么的

SpringAMQ实现

子主题

1.1 创建网络
因为我们需要部署kibana容器，因此需要让ES和kibana容器互联。这里先创建一个网络：
docker network create es-net
1.2 加载镜像
docker pull elasticsearch 
1.3 运行
运行docker命令，部署单点ES:
docker run -d \ #-d 代表后台运行
    --name es \
    # -e 代表配置环境变量 ES_JAVA_OPTS配置ES运行内存
    -e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
    # 配置RS的运行模式 single-node 单点运行
    -e "discovery.type=single-node" \
    # -v 数据卷挂载 
    -v es-data:/usr/share/elasticsearch/data \
    -v es-plugins:/usr/share/elasticsearch/plugins \
    --privileged \
    --network es-net \
    # 9200是暴露的http端口，用于用户访问
    -p 9200:9200 \
    # 9300 是ES各个节点间互联的端口
    -p 9300:9300 \
elasticsearch:7.12.1

kibana可以给我们提供一个elasticSearch的可视化界面，便于我们学习
2.1 加载镜像
docker pull kibana
2.2 部署
docker run -d \
--name kibana \
-e ELASTICSEARCH_HOSTS=http://es:9200 \
--network = es-net \
-p 5601:5601 \
kibana:7.12.1

es默认的分词器对中文不友好，需要安装插件才能对中文分词

3.1下载IK分词器压缩包
https://github.com/medcl/elasticsearch-analysis-ik/releases?after=v6.3.2
3.2 解压IK分词器压缩包，上传到es容器的插件数据卷中
/var/lib/docker/volumes/es-plugins/_data
3.3 重启容器
docker restart es
3.4 查看日志
docker logs -f es
3.5 测试
ik分词器包含两种模式：
ik_smart：最少切分
ik_max_word：最细切分

要拓展ik分词器的词库，只需要修改一个ik分词器目录中的config目录中的IkAnalyzer.cfg.xml文件：
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
        <comment>IK Analyzer 扩展配置</comment>
        <!--用户可以在这里配置自己的扩展字典 *** 添加扩展词典-->
        <entrykey="ext_dict">ext.dic</entry>
</properties>
然后在名为ext.dic的文件中，添加想要拓展的词语即可：
传智播客
奥力给

要禁用某些敏感词条，只需要修改一个ik分词器目录中的config目录中的IkAnalyzer.cfg.xml文件：
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
        <comment>IK Analyzer 扩展配置</comment>
        <!--用户可以在这里配置自己的扩展字典-->
        <entry key="ext_dict">ext.dic</entry>
         <!--用户可以在这里配置自己的扩展停止词字典  *** 添加停用词词典-->
        <entry key="ext_stopwords">stopword.dic</entry>
</properties>
<?xml version="1.0" encoding="UTF-8"?>

然后在名为stopword.dic的文件中，添加想要拓展的词语即可：
习大大

mapping是对索引库中文档的约束，常见的mapping属性包括：
  • type：字段数据类型，常见的简单类型有：
        • 字符串：text：（可分词的文本）、keyword（精确值，例如：品牌、国家、ip地址）
        • 数值：long、integer、short、byte、double、float、
        • 布尔：boolean
        • 日期：date
        • 对象：object
• index：是否创建索引，默认为true
• analyzer：使用哪种分词器
• properties：该字段的子字段

ES中通过Restful请求操作索引库、文档。请求内容用DSL语句来表示。
创建索引库和mapping的DSL语法如下：

PUT /索引库名称
{
  "mappings": {
    "properties": {
      "字段名":{
        "type": "text",
        "analyzer": "ik_smart"
      },
      "字段名2":{
        "type": "keyword",
        "index": "false"
      },
      "字段名3":{
        "properties": {
          "子字段":{
            "type": "keyword"
          }
        }
      },
    // ...略
    }
  }
}

例：
PUT /heima
{
  "mappings": {
    "properties": {
      "info":{
        "type": "text",
        "analyzer": "ik_smart"
      },
      "email":{
        "type": "keyword",
        "index": "false"
      },
      "name":{
        "properties": {
          "firstName": {
            "type": "keyword"
          }
        }
      },
    // ...略
    }
  }
}

查看索引库语法：GET /索引库名
示例：GET /heima
删除索引库的语法：DELETE /索引库名
示例：DELETE /heima

修改索引库
索引库和mapping一旦创建无法修改，但是可以添加新的字段，语法如下：
示例：
PUT /索引库名/_mapping
{
  "properties": {
    "新字段名":{
      "type": "integer"
    }
  }
}
例：
PUT /heima/_mapping
{
  "properties": {
    "age":{
      "type": "integer"
    }
  }
}

新增文档的DSL语法如下：
POST /索引库名/_doc/文档id
{
    "字段1": "值1",
    "字段2": "值2",
    "字段3": {
        "子属性1": "值3",
        "子属性2": "值4"   
    },
    // ...
}

示例
POST /heima/_doc/1
{
    "info": "黑马程序员Java讲师",
    "email": "zy@itcast.cn",
    "name": {
        "firstName": "云",
        "lastName": "赵"
     }
}

查看文档语法：GET /索引库名/_doc/文档id
示例：GET /heima/_doc/1
删除索引库的语法：DELETE /索引库名/_doc/文档id
示例：DELETE /heima/_doc/1

方式一：全量修改，会删除旧文档，添加新文档
PUT /索引库名/_doc/文档id
{
    "字段1": "值1",
    "字段2": "值2",
    // ...略
}
示例：
PUT /heima/_doc/1
{
    "info": "黑马程序员高级Java讲师",
    "email": "zy@itcast.cn",
    "name": {
        "firstName": "云",
        "lastName": "赵"
    }
}

方式二：增量修改，修改指定字段值
POST /索引库名/_update/文档id
{
    "doc": {
        "字段名":"新的值",
    }
}
示例
POST /heima/_update/1
{
  "doc": {
    "email": "ZhaoYun@itcast.cn"
  }
}

Elasticsearch提供了基于JSON的DSL
（Domain Specific  Language）来定义查询。
常见的查询类型包括：
    • 查询所有：查询出所有数据，一般测试用。例如：match_all
    • 全文检索（full text）查询：利用分词器对用户输入内容分词，然后去倒排索引库中匹配。例如：
        • match_query
        • multi_match_query
    • 精确查询：根据精确词条值查找数据，一般是查找 keyword、数值、日期、boolean等类型字段。例如：
        • ids
        • range
        • term
    • 地理（geo）查询：根据经纬度查询。例如：
        • geo_distance
        • geo_bounding_box
    • 复合（compound）查询：复合查询可以将上述各种查询条件组合起来，合并查询条件。例如：
        • bool
        • function_score

查询的基本语法如下：
GET /索引库名/_search
{
  "query": {
    "查询类型": {     
        "查询条件": "条件值"
    }
  }
}

//查询所有
GET /索引库名/_search
{
  "query": {
    "match_all": {
     }
  }
}

match查询：
全文检索查询的一种，会对用户输入内容分词，然后去倒排索引库检索，语法：
GET /indexName/_search
{
  "query": {
    "match": {
      "FIELD": "TEXT"
    }
  }
}

multi_match：
与match查询类似，只不过允许同时查询多个字段，语法：
GET /indexName/_search
{
  "query": {
    "multi_match": {
      "query": "TEXT",
      "fields": ["FIELD1", " FIELD12"]
    }
  }
}

精确查询一般是查找keyword、数值、日期、boolean等类型字段。所以不会对搜索条件分词。
常见的有：
• term：根据词条精确值查询
• range：根据值的范围查询

精确查询常见的有term查询和range查询。语法如下：
// term查询
GET /indexName/_search
{
  "query": {
    "term": {
      "FIELD": {
        "value": "VALUE"
      }
    }
  }
}
// range查询
GET /indexName/_search
{
  "query": {
    "range": {
      "FIELD": {
        "gte": 10,
        "lte": 20
      }
    }
  }
}

根据经纬度查询，官方文档。例如：
• geo_bounding_box：查询geo_point值落在某个矩形范围的所有文档
// geo_bounding_box查询
GET /indexName/_search
{
  "query": {
    "geo_bounding_box": {
      "FIELD": {
        "top_left": {
          "lat": 31.1,
          "lon": 121.5
        },
        "bottom_right": {
          "lat": 30.9,
          "lon": 121.7
        }
      }
    }
  }
}

• geo_distance：查询到指定中心点小于某个距离值的所有文档
// geo_distance查询
GET /indexName/_search
{
  "query": {
    "geo_distance": {
      "distance": "15km",
      "FIELD": "31.21,121.5"
    }
  }
}

Function Score Query

Boolean Query

分页

深度分页问题

排序

原理是这样的：
• 将搜索结果中的关键字用标签标记出来
• 在页面中给标签添加css样式
语法：
GET /hotel/_search
{
  "query": {
    "match": {
      "FIELD": "TEXT"
    }
  },
  "highlight": {
    "fields": { // 指定要高亮的字段
      "FIELD": {
        "pre_tags": "<em>",  // 用来标记高亮字段的前置标签
        "post_tags": "</em>"// 用来标记高亮字段的后置标签
      }
    }
  }
}

我们通过match_all来演示下基本的API，先看请求DSL的组织：
@Test
void testMatchAll() throws IOException {
// 1.准备
Request SearchRequest request = new SearchRequest("hotel");  //这句等同于 GET /indexName/_search
// 2.组织DSL参数
request.source().query(QueryBuilders.matchAllQuery()); //这句等同于{
                                                                                                                   "query": {
                                                                                                                        "match_all": {}
                                                                                                                   }
                                                                                                              }
// 3.发送请求，得到响应结果
SearchResponse response =client.search(request, RequestOptions.DEFAULT);
// ...
解析响应结果
}

假设存在如下调用链  A服务调用B服务，B服务调用C服务
Service1 --> Service2 --> Service3
        而此时，Service A的流量波动很大，流量经常会突然性增加！
那么在这种情况下，就算Service A能扛得住请求，Service B和Service C未必能扛得住这突发的请求。
此时，如果Service C因为抗不住请求，变得不可用，宕机。
那么Service B会因为之前请求未响应，资源无法释放（请求阻塞），新来的请求就会调用新的连接资源，
慢慢的所有连接资源耗尽，Service B就会变得不可用。紧接着，Service A也会不可用。

一个服务失败，导致整条链路的服务都失败的情形，我们称之为服务雪崩。

1、超时处理：设定超时时间，请求超过一定时间没有响应就返回错误信息，不会无休止等待

2、舱壁模式：限定每个业务能使用的线程数，避免耗尽整个tomcat的资源，因此也叫线程隔离。

3、熔断降级：由断路器统计业务执行的异常比例，如果超出阈值则会熔断该业务，拦截访问该业务的一切请求。

4、流量控制：限制业务访问的QPS，避免服务因流量的突增而故障。

认识Sentinel

Sentinel安装

我们在order-service中整合Sentinel，并且连接Sentinel的控制台，步骤如下：
1、引入sentinel依赖：
<!--sentinel--> 
<dependency>    
   <groupId>com.alibaba.cloud</groupId>
   <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
</dependency>

2、配置控制台地址：
spring:   
  cloud:      
    sentinel:       
      transport:         
        dashboard: localhost:8080

3、访问微服务的任意端点，触发sentinel监控

簇点链路

使用方法

流控模式

流控效果

热点参数限流

熔断降级的作用

降级

隔离

异常比例或异常数

慢调用

什么是熔断降级

什么是服务授权

如何实现

默认情况下，发生限流、降级、授权拦截时，都会抛出异常到调用方。如果要自定义异常时的返回结果，需要实现BlockExceptionHandler接口：
public interface BlockExceptionHandler {     
     /**      
      * 处理请求被限流、降级、授权拦截时抛出的异常：BlockException      
      */   
     void handle(HttpServletRequest request, HttpServletResponse response, BlockException e) throws Exception; 
 }

而BlockException包含很多个子类，分别对应不同的场景：
FlowException  限流异常
ParamFlowException  热点参数异常
DegradeException  降级异常
AuthorityException  授权规则异常
SystemBlockException  系统规则异常

Sentinel的控制台规则管理有三种模式

实现push模式

事务中的所有操作，要么全部成功，要么全部失败

要保证数据库内部完整性约束、声明性约束

对同一资源操作的事务不能同时发生

对数据库做的一切修改将永久保存，不管是否出现故障

一致性（Consistence）

可用性（Availability）

分区容错性（partition tolerance ）

BASE理论是对CAP的一种解决思路，包含三个思想：
Basically Available （基本可用）：分布式系统在出现故障时，允许损失部分可用性，即保证核心可用。
Soft State（软状态）：在一定时间内，允许出现中间状态，比如临时的不一致状态。
Eventually Consistent（最终一致性）：虽然无法保证强一致性，但是在软状态结束后，最终达到数据一致。

而分布式事务最大的问题是各个子事务的一致性问题，因此可以借鉴CAP定理和BASE理论：
AP模式：各子事务分别执行和提交，允许出现结果不一致，然后采用弥补措施恢复数据即可，实现最终一致。
CP模式：各个子事务执行后互相等待，同时提交，同时回滚，达成强一致。但事务等待过程中，处于弱可用状态。

Seata事务管理中有三个重要的角色：
TC (Transaction Coordinator) - 事务协调者：维护全局和分支事务的状态，协调全局事务提交或回滚。
TM (Transaction Manager) - 事务管理器：定义全局事务的范围、开始全局事务、提交或回滚全局事务。
RM (Resource Manager) - 资源管理器：管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

XA模式：强一致性分阶段事务模式，牺牲了一定的可用性，无业务侵入
TCC模式：最终一致的分阶段事务模式，有业务侵入
AT模式：最终一致的分阶段事务模式，无业务侵入，也是Seata的默认模式
SAGA模式：长事务模式，有业务侵入

Seata的部署

Seata集成微服务

当某分支事务的try阶段阻塞时，可能导致全局事务超时而触发二阶段的cancel操作。
在未执行try操作时先执行了cancel操作，这时cancel不能做回滚，就是空回滚。

对于已经空回滚的业务，如果以后继续执行try，就永远不可能confirm或cancel，
这就是业务悬挂。应当阻止执行空回滚后的try操作，避免悬挂

1、Redis内部有触发RDB的机制，可以在redis.conf文件中找到，格式如下：
# 900秒内，如果至少有1个key被修改，则执行bgsave ， 如果是save "" 则表示禁用RDB
save 900 1  
save 300 10  
save 60 10000  #代表60秒内至少执行1000次修改则触发RDB

2、RDB的其它配置也可以在redis.conf文件中设置：
# 是否压缩 ,建议不开启，压缩也会消耗cpu，磁盘的话不值钱
rdbcompression yes

# RDB文件名称
dbfilename dump.rdb  

# 文件保存的路径目录
dir ./ 

3、RDB的缺点？
RDB执行间隔时间长，两次RDB之间写入数据有丢失的风险
fork子进程、压缩、写出RDB文件都比较耗时

1、AOF默认是关闭的，需要修改redis.conf配置文件来开启AOF：
# 是否开启AOF功能，默认是no
appendonly yes
# AOF文件的名称
appendfilename "appendonly.aof"

2、AOF的命令记录的频率也可以通过redis.conf文件来配：
# 表示每执行一次写命令，立即记录到AOF文件
appendfsync always 
# 写命令执行完先放入AOF缓冲区，然后表示每隔1秒将缓冲区数据写到AOF文件，是默认方案
appendfsync everysec 
# 写命令执行完先放入AOF缓冲区，由操作系统决定何时将缓冲区内容写回磁盘
appendfsync no

3、因为是记录命令，AOF文件会比RDB文件大的多。而且AOF会记录对同一个key的多次写操作，但只有最后一次写操作才有意义。通过执行bgrewriteaof命令，可以让AOF文件执行重写功能，用最少的命令达到相同效果。

Redis也会在触发阈值时自动去重写AOF文件。阈值也可以在redis.conf中配置：
# AOF文件比上次文件 增长超过多少百分比则触发重写 
auto-aof-rewrite-percentage 100 
# AOF文件体积最小多大以上才触发重写 
 auto-aof-rewrite-min-size 64mb 

RDB 的持久化方式是对整个内存做快照，通过设置Save时间，来确定是多久更新一次，文件体积小
这会导致数据可能不完整，但宕机时恢复速度快，他在执行写操作时会占用CPU资源和内存资源，
适用于可容忍部分数据丢失，追求更快的启动速度的场景

AOF 是记录每次的执行命令，数据相对完整，文件体积大，但宕机时恢复速度慢，因为要一条一条的执行命令。
数据完整性高，系统资源占用低，但在重写时，会占用大量的CPU和内存
适用于对数据安全性较高的场景

一、下载并解压Redis
1、执行下面的命令下载redis：
wget https://download.redis.io/releases/redis-6.2.6.tar.gz

2、解压redis：
tar xzf redis-6.2.6.tar.gz

3、移动redis目录，一般都会将redis目录放置到 /usr/local/redis目录：
mv redis-6.2.6 /usr/local/redis

二、编译并安装redis
1、进入redis安装目录，执行make命令编译redis：
cd /usr/local/redis

make
等待make命令执行完成即可。

如果执行make命令报错：cc 未找到命令，原因是虚拟机系统中缺少gcc，执行下面命令安装gcc：

yum -y install gcc automake autoconf libtool make

如果执行make命令报错：致命错误:jemalloc/jemalloc.h: 没有那个文件或目录，则需要在make指定分配器为libc。
执行下面命令即可正常编译：

make MALLOC=libc

make命令执行完，redis就编译完成了。

2、执行下面命令安装redis，并指定安装目录

make install PREFIX=/usr/local/redis

至此，redis即安装成功。

三、启动redis
1、进入redis安装目录，执行下面命令启动redis服务

./bin/redis-server redis.conf

2.Redis主从集群

## 2.1.集群结构

我们搭建的主从集群结构，共包含三个节点，一个主节点，两个从节点。

这里我们会在同一台虚拟机中开启3个redis实例，模拟主从集群，信息如下：

|       IP        | PORT |  角色  |
| :-------------: | :--: | :----: |
| 192.168.150.101 | 7001 | master |
| 192.168.150.101 | 7002 | slave  |
| 192.168.150.101 | 7003 | slave  |

## 2.2.准备实例和配置

要在同一台虚拟机开启3个实例，必须准备三份不同的配置文件和目录，配置文件所在目录也就是工作目录。

1）创建目录
我们创建二个文件夹，名字分别叫6380、6381：
# 进入/tmp目录
cd /tmp
# 创建目录
mkdir 6380 6381

2）配置从节点redis.config
port 6380 #从节点端口号
pidfile /var/run/redis_6380.pid  # 把pid进程号写入pidfile配置的文件
logfile "6380.log"
dir /usr/local/redis-5.0.3/data/6380  # 指定数据存放目录
# 需要注释掉bind
# bind 127.0.0.1（bind绑定的是自己机器网卡的ip，如果有多块网卡可以配多个ip，
代表允许客户端通过机器的哪些网卡ip去访问，内网一般可以不配置bind，注释掉即可）
replicaof 192.168.0.60 6379   # 配置主从复制，从本机6379的redis实例复制数据，Redis 5.0之前使用slaveof
replica-read-only yes  # 配置从节点只读

3）拷贝配置文件到新创建的目录下6380
cp ./redis.conf  /tmp/6380

4）重复2，3步再配置一个6381从节点


2.3.启动

为了方便查看日志，我们打开3个ssh窗口，分别启动3个redis实例，启动命令：
进入redis的bin目录，输入以下命令，启动三个节点
redis-server redis.conf    （主）
redis-server /tmp/6380/redis.conf   （从）
redis-server /tmp/6381/redis.conf   （从）

启动后：
如果要一键停止，可以运行下面命令：
printf '%s\n' 7001 7002 7003 | xargs -I{} -t redis-cli -p {} shutdown

## 2.5.测试

执行下列操作以测试：

- 利用redis-cli连接6379，执行```set num 123```

- 利用redis-cli连接6380，执行```get num```，再执行```set num 666```

- 利用redis-cli连接6381，执行```get num```，再执行```set num 888```

可以发现，只有在6379这个master节点上可以执行写操作，7002和7003这两个slave节点只能执行读操作。

# 3.搭建哨兵集群

## 3.1.集群结构

这里我们搭建一个三节点形成的Sentinel集群，来监管之前的Redis主从集群。

三个sentinel实例信息如下：

| 节点 |       IP        | PORT  |
| ---- | :-------------: | :---: |
| s1   | 192.168.150.101 | 26379 |
| s2   | 192.168.150.101 | 26380 |
| s3   | 192.168.150.101 | 26381 |

## 3.2.准备实例和配置

要在同一台虚拟机开启3个实例，必须准备三份不同的配置文件和目录，配置文件所在目录也就是工作目录。

我们创建三个文件夹，名字分别叫s1、s2、s3：

```sh
# 进入/tmp目录
cd /tmp
# 创建目录
mkdir s1 s2 s3
```

然后我们在s1目录创建一个sentinel.conf文件，添加下面的内容：
#是否开启守护线程
daemonize no
# 不需要创建该文件，文件会自动生成，这个文件是在开启守护线程时生成的
pidfile "/var/run/redis-sentinel-26379.pid" 
# 不需要创建该文件，文件会自动生成（注意，不填写时，日志会在终端直接输出，填写后日志输出到文件中）
logfile "26379.log" 
# 当前sentinel实例的端口号
port 26379  
# 指定主节点信息  - `mymaster`：主节点名称 `2`：选举master时的quorum值
sentinel monitor mymaster 192.168.177.128 6379 2
# 主节点宕机后的等待时间
sentinel down-after-milliseconds mymaster 5000
sentinel failover-timeout mymaster 60000
dir "/tmp/s1"

然后将s1/sentinel.conf文件拷贝到s2、s3两个目录中（在/tmp目录执行下列命令）：

# 方式一：逐个拷贝
cp s1/sentinel.conf s2
cp s1/sentinel.conf s3
# 方式二：管道组合命令，一键拷贝
echo s2 s3 | xargs -t -n 1 cp s1/sentinel.conf

修改s2、s3两个文件夹内的配置文件，将端口分别修改为27002、27003：
sed -i -e 's/27001/27002/g' -e 's/s1/s2/g' s2/sentinel.conf
sed -i -e 's/27001/27003/g' -e 's/s1/s3/g' s3/sentinel.conf

## 3.3.启动
为了方便查看日志，我们打开3个ssh窗口，分别启动3个redis实例，启动命令：

# 第1个
redis-sentinel s1/sentinel.conf
# 第2个
redis-sentinel s2/sentinel.conf
# 第3个
redis-sentinel s3/sentinel.conf

## 3.4.测试
尝试让master节点7001宕机，
查看sentinel日志，查看7003的日志，查看7002的日志

 4.1.集群结构

分片集群需要的节点数量较多，这里我们搭建一个最小的分片集群，包含3个master节点，每个master包含一个slave节点
这里我们会在同一台虚拟机中开启6个redis实例，模拟分片集群，信息如下：
|       IP        | PORT |  角色  |
| :-------------: | :--: | :----: |
| 192.168.150.101 | 7001 | master |
| 192.168.150.101 | 7002 | master |
| 192.168.150.101 | 7003 | master |
| 192.168.150.101 | 8001 | slave  |
| 192.168.150.101 | 8002 | slave  |
| 192.168.150.101 | 8003 | slave  |

## 4.2.准备实例和配置

删除之前的7001、7002、7003这几个目录，重新创建出7001、7002、7003、8001、8002、8003目录：

# 进入/tmp目录
cd /tmp
# 创建目录
mkdir 7001 7002 7003 8001 8002 8003

在/tmp下准备一个新的redis.conf文件，内容如下：

port 6379
# 开启集群功能
cluster-enabled yes
# 集群的配置文件名称，不需要我们创建，由redis自己维护
cluster-config-file /tmp/6379/nodes.conf
# 节点心跳失败的超时时间
cluster-node-timeout 5000
# 持久化文件存放目录
dir /tmp/6379
# 绑定地址
bind 0.0.0.0
# 让redis后台运行（守护进程）
daemonize yes
# 注册的实例ip
replica-announce-ip 192.168.177.128
# 保护模式（不需要做什么用户名密码的校验了）
protected-mode no
# 数据库数量
databases 1
# 日志
logfile /tmp/6379/run.log

将这个文件拷贝到每个目录下：

# 进入/tmp目录
cd /tmp
# 执行拷贝
echo 7001 7002 7003 8001 8002 8003 | xargs -t -n 1 cp redis.conf

修改每个目录下的redis.conf，将其中的6379修改为与所在目录一致：

# 进入/tmp目录
cd /tmp
# 修改配置文件
printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t sed -i 's/6379/{}/g' {}/redis.conf

## 4.3.启动

因为已经配置了后台启动模式，所以可以直接启动服务：

# 进入/redis目录
cd /usr/local/redis/
# 一键启动所有服务
printf  '%s\n'  /tmp/7001 /tmp/7002 /tmp/7003 /tmp/8001 /tmp/8002 /tmp/8003 | xargs -I{} -t ./bin/redis-server {}/redis.conf

通过ps查看状态：
ps -ef | grep redis
发现服务都已经正常启动

如果要关闭所有进程，可以执行命令：
ps -ef | grep redis | awk '{print $2}' | xargs kill

或者（推荐这种方式）：
printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t redis-cli -p {} shutdown

## 4.4.创建集群

虽然服务启动了，但是目前每个服务之间都是独立的，没有任何关联。

我们需要执行命令来创建集群，在Redis5.0之前创建集群比较麻烦，5.0之后集群管理命令都集成到了redis-cli中。

1）Redis5.0之前

Redis5.0之前集群命令都是用redis安装包下的src/redis-trib.rb来实现的。因为redis-trib.rb是有ruby语言编写的所以需要安装ruby环境。

 # 安装依赖
 yum -y install zlib ruby rubygems
 gem install redis

然后通过命令来管理集群：

# 进入redis的src目录
cd /tmp/redis-6.2.4/src
# 创建集群
./redis-trib.rb create --replicas 1 192.168.150.101:7001 192.168.150.101:7002 192.168.150.101:7003 192.168.150.101:8001 192.168.150.101:8002 192.168.150.101:8003

2）Redis5.0以后

我们使用的是Redis6.2.4版本，集群管理以及集成到了redis-cli中，格式如下：
redis-cli --cluster create --cluster-replicas 1 192.168.150.101:7001 192.168.150.101:7002 192.168.150.101:7003 192.168.150.101:8001 192.168.150.101:8002 192.168.150.101:8003

命令说明：
- `redis-cli --cluster`或者`./redis-trib.rb`：代表集群操作命令
- `create`：代表是创建集群
- `--replicas 1`或者`--cluster-replicas 1` ：指定集群中每个master的副本个数为1，此时`节点总数 ÷ (replicas + 1)` 得到的就是master的数量。因此节点列表中的前n个就是master，其它节点都是slave节点，随机分配到不同master

然后会问，是否同意这么分配节点，输入yes，则集群开始创建

通过命令可以查看集群状态：
redis-cli -p 7001 cluster nodes

## 4.5.测试

尝试连接7001节点，存储一个数据：
# 连接
redis-cli -c -p 7001
# 存储数据
set num 123
# 读取数据
get num
# 再次存储
set a 1

Replication Id：简称replid，是数据集的标记，id一致则说明是同一数据集。每一个master都有唯一的replid，slave则会继承master节点的replid
offset：偏移量，随着记录在repl_baklog中的数据增多而逐渐增大。slave完成同步时也会记录当前同步的offset。如果slave的offset小于master的offset，说明slave数据落后于master，需要更新。
因此slave做数据同步，必须向master声明自己的replication id 和offset，master才可以判断到底需要同步哪些数据

slave节点请求增量同步
master节点判断replid，发现不一致，拒绝增量同步
master将完整内存数据生成RDB，发送RDB到slave
slave清空本地数据，加载master的RDB
master将RDB期间的命令记录在repl_baklog，并持续将log中的命令发送给slave
slave执行接收到的命令，保持与master之间的同步

全量同步：当slave节点第一次连接master节点时或slave节点断开时间太久，repl_baklog中的offset已经被覆盖时
                 执行全量同步，master将完整内存数据生成RDB，发送RDB到slave。后续命令则记录在repl_baklog，
                 逐个发送给slave。
增量同步：slave节点断开又恢复，并且在repl_baklog中能找到offset时，执行增量同步，slave提交自己的offset到master，
                 master获取repl_baklog中从offset之后的命令给slave

可以从以下几个方面来优化Redis主从就集群：
在master中配置repl-diskless-sync yes启用无磁盘复制，避免全量同步时的磁盘IO。
Redis单节点上的内存占用不要太大，减少RDB导致的过多磁盘IO
适当提高repl_baklog的大小，发现slave宕机时尽快实现故障恢复，尽可能避免全量同步
限制一个master上的slave节点数量，如果实在是太多slave，则可以采用主-从-从链式结构，减少master压力

Redis提供了哨兵（Sentinel）机制来实现主从集群的自动故障恢复。哨兵的结构和作用如下：
监控：Sentinel 会不断检查您的master和slave是否按预期工作
自动故障恢复：如果master故障，Sentinel会将一个slave提升为master。当故障实例恢复后也以新的master为主
通知：Sentinel充当Redis客户端的服务发现来源，当集群发生故障转移时，会将最新信息推送给Redis的客户端

Sentinel基于心跳机制监测服务状态，每隔1秒向集群的每个实例发送ping命令：
主观下线：如果某sentinel节点发现某实例未在规定时间响应，则认为该实例主观下线。
客观下线：若超过指定数量（quorum）的sentinel都认为该实例主观下线，则该实例客观下线。quorum值最好超过Sentinel实例数量的一半。

一旦发现master故障，sentinel需要在salve中选择一个作为新的master，选择依据是这样的：
首先会判断slave节点与master节点断开时间长短，如果超过指定值（down-after-milliseconds * 10）则会排除该slave节点
然后判断slave节点的slave-priority值，越小优先级越高，如果是0则永不参与选举
如果slave-prority一样，则判断slave节点的offset值，越大说明数据越新，优先级越高
最后是判断slave节点的运行id大小，越小优先级越高。

当选中了其中一个slave为新的master后（例如slave1），故障的转移的步骤如下：
sentinel给备选的slave1节点发送slaveof no one命令，让该节点成为master
sentinel给所有其它slave发送slaveof 192.168.150.101 7002 命令，让这些slave成为新master的从节点，开始从新的master上同步数据。
最后，sentinel将故障节点标记为slave，当故障节点恢复后会自动成为新的master的slave节点

在Sentinel集群监管下的Redis主从集群，其节点会因为自动故障转移而发生变化，Redis的客户端必须感知这种变化，及时更新连接信息。Spring的RedisTemplate底层利用lettuce实现了节点的感知和自动切换。

1、在pom文件中引入redis的starter依赖：
<dependency>    
    <groupId>org.springframework.boot</groupId>     
    <artifactId>spring-boot-starter-data-redis</artifactId>  
</dependency>

2、然后在配置文件application.yml中指定sentinel相关信息：
spring:   
  redis:     
    sentinel:       
      master: mymaster # 指定master名称       
      nodes: # 指定redis-sentinel集群信息          
        - 192.168.150.101:27001        
        - 192.168.150.101:27002        
        - 192.168.150.101:27003
3、配置主从读写分离
@Bean 
public LettuceClientConfigurationBuilderCustomizer configurationBuilderCustomizer(){     
    return configBuilder -> configBuilder.readFrom(ReadFrom.REPLICA_PREFERRED); 
}
这里的ReadFrom是配置Redis的读取策略，是一个枚举，包括下面选择：
MASTER：从主节点读取
MASTER_PREFERRED：优先从master节点读取，master不可用才读取replica
REPLICA：从slave（replica）节点读取
REPLICA _PREFERRED：优先从slave（replica）节点读取，所有的slave都不可用才读取master

1、主从和哨兵可以解决高可用、高并发读的问题。但是依然有两个问题没有解决：
      海量数据存储问题
      高并发写的问题
2、使用分片集群可以解决上述问题，分片集群特征：
      集群中有多个master，每个master保存不同数据
      每个master都可以有多个slave节点
      master之间通过ping监测彼此健康状态
      客户端请求可以访问集群任意节点，最终都会被转发到正确节点

1、Redis会把每一个master节点映射到0~16383共16384个插槽（hash slot）上，查看集群信息时就能看到。
2、数据key不是与节点绑定，而是与插槽绑定。redis会根据key的有效部分计算插槽值，分两种情况：
      key中包含"{}"，且“{}”中至少包含1个字符，“{}”中的部分是有效部分
      key中不包含“{}”，整个key都是有效部分
      例如：key是num，那么就根据num计算，如果是{itcast}num，则根据itcast计算。计算方式是利用CRC16算法得到一个hash值，然后对16384取余，得         到的结果就是slot值。

添加节点到集群

从集群中删除节点

具体实现原理

步骤

1、在pom文件中引入redis的starter依赖：
<dependency>    
    <groupId>org.springframework.boot</groupId>     
    <artifactId>spring-boot-starter-data-redis</artifactId>  
</dependency>

2、然后在配置文件application.yml中指定sentinel相关信息：
spring:   
  redis:     
    cluster:       
      nodes: # 指定redis-cluster集群信息          
        - 192.168.177.128:7001        
        - 192.168.177.128:7002        
        - 192.168.177.128:7003
        - 192.168.177.128:8001
        - 192.168.177.128:8001
        - 192.168.177.128:8003
3、配置主从读写分离
@Bean 
public LettuceClientConfigurationBuilderCustomizer configurationBuilderCustomizer(){     
    return configBuilder -> configBuilder.readFrom(ReadFrom.REPLICA_PREFERRED); 
}
这里的ReadFrom是配置Redis的读取策略，是一个枚举，包括下面选择：
MASTER：从主节点读取
MASTER_PREFERRED：优先从master节点读取，master不可用才读取replica
REPLICA：从slave（replica）节点读取
REPLICA _PREFERRED：优先从slave（replica）节点读取，所有的slave都不可用才读取master