My SpringCloud notes

# 认识微服务

# 单体架构

单体架构：将业务的所有功能集中在一个项目中开发，打成一个包部署。

优点： 架构简单，部署成本低

缺点： 耦合度高（维护困难、升级困难）

# 分布式架构

分布式架构：根据业务功能对系统做拆分，每个业务功能模块作为独立项目开发，称为一个服务。

优点： 降低服务耦合，有利于服务升级和拓展

缺点： 服务调用关系错综复杂

分布式架构虽然降低了服务耦合，但是服务拆分时也有很多问题需要思考：

• 服务拆分的粒度如何界定？
• 服务之间如何调用？
• 服务的调用关系如何管理？

人们需要制定一套行之有效的标准来约束分布式架构。

# 微服务

微服务的架构特征：

• 单一职责：微服务拆分粒度更小，每一个服务都对应唯一的业务能力，做到单一职责
• 自治：团队独立、技术独立、数据独立，独立部署和交付
• 面向服务：服务提供统一标准的接口，与语言和技术无关
• 隔离性强：服务调用做好隔离、容错、降级，避免出现级联问题

微服务的上述特性其实是在给分布式架构制定一个标准，进一步降低服务之间的耦合度，提供服务的独立性和灵活性。做到高内聚，低耦合。

因此，可以认为微服务是一种经过良好架构设计的分布式架构方案 。

其中在 Java 领域最引人注目的就是 SpringCloud 提供的方案了。

# SpringCloud

SpringCloud 是目前国内使用最广泛的微服务框架。官网地址：https://spring.io/projects/spring-cloud。

SpringCloud 集成了各种微服务功能组件，并基于 SpringBoot 实现了这些组件的自动装配，从而提供了良好的开箱即用体验。

其中常见的组件包括：

另外，SpringCloud 底层是依赖于 SpringBoot 的，并且有版本的兼容关系，如下：

# 内容知识

需要学习的微服务知识内容

技术栈

自动化部署

# 技术栈对比

---

# 服务拆分

案例源码：https://gitee.com/xn2001/cloudcode/tree/master/01-cloud-demo

服务拆分注意事项

单一职责：不同微服务，不要重复开发相同业务

数据独立：不要访问其它微服务的数据库

面向服务：将自己的业务暴露为接口，供其它微服务调用，其他微服务就只需要发送请求到其他的微服务来调用那个微服务等等功能

cloud-demo：父工程，管理依赖

• order-service：订单微服务，负责订单相关业务
• user-service：用户微服务，负责用户相关业务

要求：

• 订单微服务和用户微服务都必须有各自的数据库，相互独立
• 订单服务和用户服务都对外暴露 Restful 的接口
• 订单服务如果需要查询用户信息，只能调用用户服务的 Restful 接口，不能查询用户数据库

微服务项目下，打开 idea 中的 Service，可以很方便的启动。

启动完成后，访问 http://localhost:8080/order/101

---

# 远程调用

案例源码：https://gitee.com/xn2001/cloudcode/tree/master/02-cloud-restTemplate

正如上面的服务拆分要求中所提到，

订单服务如果需要查询用户信息，只能调用用户服务的 Restful 接口，不能查询用户数据库

因此我们需要知道 Java 如何去发送 http 请求，Spring 提供了一个 RestTemplate 工具，只需要把它创建出来即可。（即注入 Bean）

需要在 order 微服务 (module) 的 springboot 启动类里面写上以下这个代码，把 RestTemplate 类的对象注册到 IOC 容器中，然后我们就可以在 order 微服务里面发送 http 请求到别的微服务去

然后再我们 order 微服务里面 controller 层调用了 service 层，service 层则有着 dao/mapper 层查询 order 微服务专属的数据库的信息 (里面除了 order 的信息外还包含了每个 order 对应的 userId, 但是没有那个 User 的信息)

我们需要在这个 service 层写以下的代码:

发送请求，自动序列化为 Java 对象。

• 首先因为我们把 RestTemplate 类的对象注册到 IOC 容器中了，我们可以直接用里面的方法

• 我们在 service 层用着调用 dao 层查询出来的结果中的 userId 信息来 (调用 RestTemplate 类里面的方法) 发送请求

• 然后这个 getForObject 方法作用就是发送请求然后把相应的结果返回给我们为一个对象，(要是我们没写那个 User.class 就传的就是个 json 对象)

• 然后我们就可以把这个请求到的 User 信息封装给我们的 order 类的实例化对象里面，然后再把这个 order 类的实例化对象返回出去给 controller,controller 再相应任何对应的请求

这样我们就在 order 微服务远程调用了 user 微服务中的 controller 层获取到了想要的信息 (注意这个 user 微服务中的 controller 层那个 RequestMapping 注解啥的要配置好才行–> 复合我们上面写的那个 http 请求)

启动完成后，访问：http://localhost:8080/order/101

在上面 order 是消费者，user 是提供者，因为 user 提供接口而 order 则调用了接口

一个微服务可以是消费者也可以是提供者，比如说 A 调用了 B,B 调用了 C, 那么 B 就是又是消费者又是提供者

---

在上面代码的 url 中，我们可以发现调用服务的地址采用硬编码，这在后续的开发中肯定是不理想的，这就需要服务注册中心（Eureka）来帮我们解决这个事情。

比如说我们的 user 微服务做了集群，有好几个 user 微服务，要是我们上面这么做只能会是给一个服务器的 user 微服务发请求，集群没意义了.

# Eureka 注册中心

案例源码：https://gitee.com/xn2001/cloudcode/tree/master/03-cloud-eureka

最广为人知的注册中心就是 Eureka，其结构如下：

不管是服务消费者还是服务提供者都是 eureka 的 client

order-service 如何得知 user-service 实例地址？

• 每一个 user-service 服务实例启动后，将自己的信息注册到 eureka-server (Eureka 服务端)，叫做服务注册
• eureka-server 保存服务名称到服务实例地址列表的映射关系
• 每一个 order-service 根据服务名称，拉取实例地址列表，这个叫服务发现或服务拉取

order-service 如何从多个 user-service 实例中选择具体的实例？

eureka-server 发给 order-service 所有注册了的 user-service 实例，然后 order-service 从实例列表中利用负载均衡算法选中一个实例地址，向该实例地址发起远程调用 (发请求)

order-service 如何得知某个 user-service 实例是否依然健康，是不是已经宕机？

• 每一个 user-service 会每隔一段时间 (默认 30 秒) 向 eureka-server 发起请求，报告自己状态，称为心跳
• 当超过一定时间没有发送心跳时，eureka-server 会认为微服务实例故障，将该实例从服务列表中剔除
• order-service 拉取服务时，就能将故障实例排除了

接下来我们动手实践的步骤包括：

# 搭建注册中心

搭建 eureka-server

引入 SpringCloud 为 eureka 提供的 starter 依赖，注意这里是用 server

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-netflix-eureka-server</artifactId>
</dependency>

编写启动类

注意要添加一个 @EnableEurekaServer 注解，开启 eureka 的注册中心功能

package com.xn2001.eureka;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.server.EnableEurekaServer;

@SpringBootApplication
@EnableEurekaServer
public class EurekaApplication {
    public static void main(String[] args) {
        SpringApplication.run(EurekaApplication.class, args);
    }
}

编写配置文件

编写一个 application.yml 文件，内容如下：

server:
  port: 10086  #服务端口
spring:
  application:
    name: eureka-server #eureka的服务名称
eureka:
  client:
    service-url:  #eureka地址信息->127.0.0.1可以换成localhost或者其他IP地址等
      defaultZone: http://127.0.0.1:10086/eureka

其中 default-zone 是因为前面配置类开启了注册中心所需要配置的 eureka 的地址信息，因为 eureka 本身也是一个微服务，这里也要将自己注册到 eureka 上 (自己注册自己)，当后面 eureka 集群时，这里就可以填写多个，使用 “,” 隔开。

启动完成后，访问 http://localhost:10086/

# 服务注册

将 user-service、order-service 都注册到 eureka

引入 SpringCloud 为 eureka 提供的 starter 依赖，注意这里是用 client

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>

在启动类上添加注解： @EnableEurekaClient

在 application.yml 文件，添加下面的配置：

spring:
  application:
      #name：orderservice
    name: userservice #注册到eureka的当前服务名称
eureka:
  client:
    service-url: #eureka服务的地址
      defaultZone: http:127.0.0.1:10086/eureka

注意这里是 eureka 的客户端 (client, 之前我们那是服务端)

注意上面的 xml 文件的依赖配置和 yaml 的服务注册到 eureka 都是在那个被注册的服务的本身的 pom.xml 和 resources 下的 application.yml 文件里面

因为提供者服务和消费者服务都需要注册到 eureka 所以我们在挨个服务都要配置上面的

注意～！！！

我们为了模仿分布式把每个微服务的端口改了

所以 user 微服务和 order 微服务的端口号不一样的

3 个项目启动后，访问 http://localhost:10086/

这里另外再补充个小技巧，我们可以通过 idea 的多实例启动，来查看 Eureka 的集群效果。

4 个项目启动后，访问 http://localhost:10086/

两个 user 微服务的实例都注册到了 eureka 服务端

# 服务拉取

在 order-service 中完成服务拉取，然后通过负载均衡挑选一个服务，实现远程调用

下面我们让 order-service 向 eureka-server 拉取 user-service 的信息，实现服务发现。

所以在 order-service 这个微服务里面:

首先给 RestTemplate 这个 Bean 添加一个 @LoadBalanced 注解，用于开启负载均衡。（后面会讲）

@Bean
@LoadBalanced
public RestTemplate restTemplate(){
    return new RestTemplate();
}

修改 OrderService 访问的 url 路径，用服务名 (就是当初我们注册那个服务给的名) 代替 ip、端口：

spring 会自动帮助我们从 eureka-server 中，根据 userservice 这个服务名称，获取实例列表后去完成负载均衡。

---

# Ribbon 负载均衡

案例源码：https://gitee.com/xn2001/cloudcode/tree/master/04-cloud-ribbon

我们添加了 @LoadBalanced 注解，即可实现负载均衡功能，这是什么原理呢？

SpringCloud 底层提供了一个名为 Ribbon 的组件，来实现负载均衡功能。

只要给那个注册 RestTemplate 类的对象到 IOC 容器的那个方法上面加上个 @LoadBalanced 注解，就证明之后用这个 RestTemplate 的方法发送请求，那个请求就会被 ribbon 拦截，然后 ribbon 就会询问 eureka 服务端得到那个传来的请求里面的那个服务名对应的所有服务实例，然后做 load balancing 选一个服务实例把请求里面的服务名换成选到的服务实例的 IP 地址和端口号，所以 order-service 可以访问到一个 user-service 服务实例

# 源码跟踪

为什么我们只输入了 service 名称就可以访问了呢？为什么不需要获取 ip 和端口，这显然有人帮我们根据 service 名称，获取到了服务实例的 ip 和端口。它就是 LoadBalancerInterceptor ，这个类会在对 RestTemplate 的请求进行拦截，然后从 Eureka 根据服务 id 获取服务列表，随后利用负载均衡算法得到真实的服务地址信息，替换服务 id。

我们进行源码跟踪：

这里的 intercept() 方法，拦截了用户的 HttpRequest 请求，然后做了几件事：

• request.getURI() ：获取请求 uri，即 http://user-service/user/8
• originalUri.getHost() ：获取 uri 路径的主机名，其实就是服务 id user-service
• this.loadBalancer.execute() ：处理服务 id，和用户请求

这里的 this.loadBalancer 是 LoadBalancerClient 类型

继续跟入 execute() 方法：

• getLoadBalancer(serviceId) ：根据服务 id 获取 ILoadBalancer ，而 ILoadBalancer 会拿着服务 id 去 eureka 中获取服务列表。
• getServer(loadBalancer) ：利用内置的负载均衡算法，从服务列表中选择一个。在图中可以看到获取了 8082 端口的服务

可以看到获取服务时，通过一个 getServer() 方法来做负载均衡:

我们继续跟入：

继续跟踪源码 chooseServer() 方法，发现这么一段代码：

我们看看这个 rule 是谁：

这里的 rule 默认值是一个 RoundRobinRule ，看类的介绍：

负载均衡默认使用了轮询算法，当然我们也可以自定义。

# 流程总结

SpringCloud Ribbon 底层采用了一个拦截器，拦截了 RestTemplate 发出的请求，对地址做了修改。

基本流程如下：

• 拦截我们的 RestTemplate 请求 http://userservice/user/1
• RibbonLoadBalancerClient 会从请求 url 中获取服务名称，也就是 user-service
• DynamicServerListLoadBalancer 根据 user-service 到 eureka 拉取服务列表
• eureka 返回列表，localhost:8081、localhost:8082
• IRule (IRule 有接口，他有很多实现类，对应着不同的 load balancing 的选择方式，上面的是轮询算法挑选的) 利用内置负载均衡规则，从列表中选择一个，例如 localhost:8081
• RibbonLoadBalancerClient 修改请求地址，用 localhost:8081 替代 userservice，得到 http://localhost:8081/user/1，发起真实请求

# 负载均衡策略

负载均衡的规则都定义在 IRule 接口中，而 IRule 有很多不同的实现类：

不同规则的含义如下：

内置负载均衡规则类	规则描述
RoundRobinRule	简单轮询服务列表来选择服务器。它是 Ribbon 默认的负载均衡规则。
AvailabilityFilteringRule	对以下两种服务器进行忽略：（1）在默认情况下，这台服务器如果 3 次连接失败，这台服务器就会被设置为 “短路” 状态。短路状态将持续 30 秒，如果再次连接失败，短路的持续时间就会几何级地增加。 （2）并发数过高的服务器。如果一个服务器的并发连接数过高，配置了 AvailabilityFilteringRule 规则的客户端也会将其忽略。并发连接数的上限，可以由客户端设置。
WeightedResponseTimeRule	为每一个服务器赋予一个权重值。服务器响应时间越长，这个服务器的权重就越小。这个规则会随机选择服务器，这个权重值会影响服务器的选择。
ZoneAvoidanceRule (默认)	以区域可用的服务器为基础进行服务器的选择。使用 Zone 对服务器进行分类，这个 Zone 可以理解为一个机房、一个机架等。而后再对 Zone 内的多个服务做轮询。
BestAvailableRule	忽略那些短路的服务器，并选择并发数较低的服务器。
RandomRule	随机选择一个可用的服务器。
RetryRule	重试机制的选择逻辑

默认的实现就是 ZoneAvoidanceRule ，是一种轮询方案。

# 自定义策略

通过定义 IRule 实现可以修改负载均衡规则，有两种方式：

1. 代码方式在 order-service 中的 OrderApplication 类中，定义一个新的 IRule：

这种方式是给所有可能会请求的提供服务来配置的负载均衡规则，而不是特定哪个微服务

2. 配置文件方式：在 order-service 的 application.yml 文件中，添加新的配置也可以修改规则：

这种方式是给某个指定的微服务来配置负载均衡规则

userservice: # 给需要调用的微服务配置负载均衡规则，orderservice服务去调用userservice服务
  ribbon:
    NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule # 负载均衡规则` 

注意：一般用默认的负载均衡规则，不做修改。

上面的就是改为 RandomRule 这个 IRule 接口的实现类，然后按照 RandomRule 里面定义的规则去做 load balancing 选择服务的规则

注意上方的不管是什么方式，都是在 order-service 微服务 (消费者) 里面配置的

# 饥饿加载

当我们启动 orderservice，第一次访问 (别的微服务) 时，时间消耗会大很多，这是因为 Ribbon 懒加载的机制。

Ribbon 默认是采用__懒加载__，即第一次访问时才会去创建 LoadBalanceClient，拉取集群地址，所以请求时间会很长。

不过之后的加载都不会那么长，因为已经被缓存到内存当中了，以后想用直接用就行.

而饥饿加载则会在项目启动时创建 LoadBalanceClient，降低第一次访问的耗时，通过下面配置开启饥饿加载：

ribbon:
  eager-load:
    enabled: true
    clients: userservice # 项目启动时直接去拉取userservice的集群，多个用","隔开,或者是yaml的数组的写法

---

# Nacos 注册中心

源码案例：https://gitee.com/xn2001/cloudcode/tree/master/05-cloud-nacos

SpringCloudAlibaba 推出了一个名为 Nacos 的注册中心，在国外也有大量的使用。

解压启动 Nacos，详细请看 Nacos 安装指南

startup.cmd -m standalone

访问：http://localhost:8848/nacos/

# 服务注册

这里上来就直接服务注册，很多东西可能有疑惑，其实 Nacos 本身就是一个 SprintBoot 项目，这点你从启动的控制台打印就可以看出来，所以就不再需要去额外搭建一个像 Eureka 的注册中心 (不用自己写一个 eureka 服务端了，直接注册我们那些微服务就行)。

引入依赖

在 cloud-demo 父工程中引入 SpringCloudAlibaba 的依赖：

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-alibaba-dependencies</artifactId>
    <version>2.2.6.RELEASE</version>
    <type>pom</type>
    <scope>import</scope>
</dependency>

然后在 user-service 和 order-service 中的 pom 文件中引入 nacos-discovery 依赖：

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>

配置 nacos 地址

在 user-service 和 order-service 的 application.yml 中添加 nacos 地址：

spring:
  cloud:
    nacos:
      server-addr: 127.0.0.1:8848

其他我们之前 ribbon 和 eureka 那些不用动 (除了注释掉了那些依赖啥的), 都一样的，会生效

项目重新启动后，可以看到三个服务都被注册进了 Nacos

浏览器访问：http://localhost:8080/order/101，正常访问，同时负载均衡也正常。

# 分级存储模型

一个服务可以有多个实例，例如我们的 user-service，可以有:

• 127.0.0.1:8081
• 127.0.0.1:8082
• 127.0.0.1:8083

假如这些实例分布于全国各地的不同机房，例如：

• 127.0.0.1:8081，在上海机房
• 127.0.0.1:8082，在上海机房
• 127.0.0.1:8083，在杭州机房

Nacos 就将同一机房内的实例，划分为一个集群。

微服务互相访问时，应该尽可能访问同集群实例，因为本地访问速度更快。当本集群内不可用时，才访问其它集群。 例如：杭州机房内的 order-service 应该优先访问同机房的 user-service。

# 配置集群

接下来我们给 user-service 配置集群

修改 user-service 的 application.yml 文件，添加集群配置：

spring:
  cloud:
    nacos:
      server-addr: localhost:8848     
      discovery:
        cluster-name: HZ # 集群名称 HZ杭州

重启两个 user-service 实例后，我们再去启动一个上海集群的实例。

-Dserver.port=8083 -Dspring.cloud.nacos.discovery.cluster-name=SH

查看 nacos 控制台：

# NacosRule

Ribbon 的默认实现 ZoneAvoidanceRule 并不能实现根据同集群优先来实现负载均衡，我们把规则改成 NacosRule 即可。我们是用 orderservice 调用 userservice，所以在 orderservice 配置规则。

@Bean
public IRule iRule(){
    //默认为轮询规则，这里自定义为随机规则
    return new NacosRule();
}

__另外，你同样可以__使用配置的形式来完成，具体参考上面的 Ribbon 栏目。

userservice:
  ribbon:
    NFLoadBalancerRuleClassName: com.alibaba.cloud.nacos.ribbon.NacosRule #负载均衡规则` 

然后，再对 orderservice 配置集群。

spring:
  cloud:
    nacos:
      server-addr: localhost:8848
      discovery:
        cluster-name: HZ # 集群名称

现在我启动了四个服务，分别是：

• orderservice - HZ
• userservice - HZ
• userservice1 - HZ
• userservice2 - SH

访问地址：http://localhost:8080/order/101

在访问中我们发现，只有同在一个 HZ 集群下的 userservice、userservice1 会被调用，并且是随机的。

我们试着把 userservice、userservice2 停掉。依旧可以访问。

在 orderservice 控制台可以看到发出了一串的警告，因为 orderservice 本身是在 HZ 集群的，这波 HZ 集群没有了 userservice，就会去别的集群找。

# 权重配置

实际部署中会出现这样的场景：

服务器设备性能有差异，部分实例所在机器性能较好，另一些较差，我们希望性能好的机器承担更多的用户请求。但默认情况下 NacosRule 是同集群内随机挑选，不会考虑机器的性能问题。

因此，Nacos 提供了权重配置来控制访问频率，0~1 之间，权重越大则访问频率越高，权重修改为 0，则该实例永远不会被访问。

在 Nacos 控制台，找到 user-service 的实例列表，点击编辑，即可修改权重。

在弹出的编辑窗口，修改权重

另外，在服务升级的时候，有一种较好的方案：我们也可以通过调整权重来进行平滑升级，例如：先把 userservice 权重调节为 0，让用户先流向 userservice2、userservice3，升级 userservice 后，再把权重从 0 调到 0.1，让一部分用户先体验，用户体验稳定后就可以往上调权重啦。

# 环境隔离

Nacos 提供了 namespace 来实现环境隔离功能。

• Nacos 中可以有多个 namespace
• namespace 下可以有 group、service 等
• 不同 namespace 之间相互隔离，例如不同 namespace 的服务互相不可见

# 创建 namespace

默认情况下，所有 service、data、group 都在同一个 namespace，名为 public (保留空间)：

我们可以点击页面新增按钮，添加一个 namespace：

然后，填写表单：

就能在页面看到一个新的 namespace：

# 配置 namespace

给微服务配置 namespace 只能通过修改配置来实现。

例如，修改 order-service 的 application.yml 文件：

spring:
  cloud:
    nacos:
      server-addr: localhost:8848
      discovery:
        cluster-name: HZ
        namespace: 492a7d5d-237b-46a1-a99a-fa8e98e4b0f9 # 命名空间ID

重启 order-service 后，访问控制台。

public

dev

此时访问 order-service，因为 namespace 不同，会导致找不到 userservice，控制台会报错：

# 临时实例

Nacos 的服务实例分为两种类型：

• 临时实例：如果实例宕机超过一定时间，会从服务列表剔除，默认的类型。
• 非临时实例：如果实例宕机，不会从服务列表剔除，也可以叫永久实例。

配置一个服务实例为永久实例：

spring:
  cloud:
    nacos:
      discovery:
        ephemeral: false # 设置为非临时实例

另外，Nacos 集群默认采用 AP 方式 (可用性)，当集群中存在非临时实例时，采用 CP 模式 (一致性)；而 Eureka 采用 AP 方式，不可切换。（这里说的是 CAP 原理，后面会写到）

# Nacos 和 Eureka 区别

• 首先先说下消费者会从注册中心 (nacos or eureka) 拉取他请求的服务的服务集群列表，这个是会存到缓存中，这样就不会每次调用都会重新从注册中心 (nacos or eureka) 拉取然后各种 load balancing 等等等，而是用内存中的然后各种 load balancing 等等等

任何注册的服务是临时实例 (默认你注册的都是临时实例) 的话，那么就跟 eureka 一样会是过多久多久的心跳检测， 如果临时实例服务挂了，那么 nacos 注册中心直接会把这个服务剔除

而非临时实例则不会发心跳检测让 nacos 注册中心知道自己还活着，而是 nacos 注册中心自己发请求到那个非临时实例问他还活着， 如果非临时实例服务挂了，那么 nacos 注册中心不会把这个服务剔除，而是把它标为不健康

设置非临时实例:

要是一个提供者挂了，那么 eureka 的注册中心不会主动推送变更消息让消费者知道他请求的提供者挂了，可能会出问题

而 nacos 的注册中心发现那个提供者服务挂了，则会立即主动推送变更消息让消费者知道他请求的提供者挂了

---

# Nacos 配置中心

案例地址：https://gitee.com/xn2001/cloudcode/tree/master/05-cloud-nacos

Nacos 除了可以做注册中心，同样可以做配置管理来使用。

当微服务部署的实例越来越多，达到数十、数百时，逐个修改微服务配置就会让人抓狂，而且很容易出错。我们需要一种统一配置管理方案，可以集中管理所有实例的配置。

Nacos 一方面可以将配置集中管理，另一方可以在配置变更时，及时通知微服务，实现配置的热更新。

# 创建配置

在 Nacos 控制面板中添加配置文件

然后在弹出的表单中，填写配置信息：

注意： 项目的核心配置，需要热更新的配置才有放到 nacos 管理的必要。基本不会变更的一些配置 (例如数据库连接) 还是保存在微服务本地比较好。

# 拉取配置

首先我们需要了解 Nacos 读取配置文件的环节是在哪一步，在没加入 Nacos 配置之前，获取配置是这样：

加入 Nacos 配置，它的读取是在 application.yml 之前的：

这时候如果把 nacos 地址放在 application.yml 中，显然是不合适的，Nacos 就无法根据地址去获取配置了。

因此，nacos 地址必须放在优先级最高的 bootstrap.yml 文件。

引入 nacos-config 依赖

首先，在 user-service 服务中，引入 nacos-config 的客户端依赖：

<!--nacos配置管理依赖-->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>

添加 bootstrap.yml

然后，在 user-service 中添加一个 bootstrap.yml 文件，内容如下：

spring:
  application:
    name: userservice # 服务名称
  profiles:
    active: dev #开发环境，这里是dev 
  cloud:
    nacos:
      server-addr: localhost:8848 # Nacos地址
      config:
        file-extension: yaml # 文件后缀名

根据 spring.cloud.nacos.server-addr 获取 nacos 地址，再根据 ${spring.application.name}-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension} 作为文件 id (就是我们 nacos 里面弄的那个配置文件的 data id)，来读取配置。

在这个例子例中，就是去读取 userservice-dev.yaml

使用代码来验证是否拉取成功

在 user-service 中的 UserController 中添加业务逻辑，读取 pattern.dateformat 配置并使用：

@Value("${pattern.dateformat}")
private String dateformat;

@GetMapping("now")
public String now(){
    //格式化时间
    return LocalDateTime.now().format(DateTimeFormatter.ofPattern(dateformat));
}

启动服务后，访问：http://localhost:8081/user/now

因为我们是给 userservice 微服务里面设置的 bootstrp.yaml 去找那个 nacos 里的配置，所以在访问 (任何方式) userservice 微服务时就会获取到那些配置

# 配置热更新

我们最终的目的，是修改 nacos 中的配置后，微服务中无需重启即可让配置生效，也就是配置热更新。

有两种方式：1. 用 @value 读取配置时，搭配 @RefreshScope ；2. 直接用 @ConfigurationProperties 读取配置

# @RefreshScope

方式一：在 @Value 注入的变量所在类上添加注解 @RefreshScope

这里我们用了那个 nacos 里的配置文件，得要加上 @RefreshScope 才会启动热更新

# @ConfigurationProperties

方式二：使用 @ConfigurationProperties 注解读取配置文件 (才发现竟然除了.properties 文件读取，还可以.yaml 文件读取)，就不需要加 @RefreshScope 注解。

在 user-service 服务中，添加一个 PatternProperties 类，读取 patterrn.dateformat 属性

@Data
@Component
@ConfigurationProperties(prefix = "pattern")
public class PatternProperties {
    public String dateformat;
}

@Autowired
private PatternProperties patternProperties;

@GetMapping("now2")
public String now2(){
    //格式化时间
    return LocalDateTime.now().format(DateTimeFormatter.ofPattern(patternProperties.dateformat));
}

这样热更新就能起效果了

只要我们 nacos 里的那个配置文件内容发生改变，不需要我们重新重启项目啊什么的，我们 nacos 会通知我们的微服务然后让他们读取这个更改后的配置文件内容，然后就生效了，不需要重新项目

# 配置共享

其实在服务启动时，nacos 会读取多个配置文件，例如：

• [spring.application.name]-[spring.profiles.active].yaml ，例如：userservice-dev.yaml
• [spring.application.name].yaml ，例如：userservice.yaml

这里的 [spring.application.name].yaml 不包含环境，因此可以被多个环境 (dev,test,…) 共享。

添加一个环境共享配置

我们在 nacos 中添加一个 userservice.yaml 文件：

在 user-service 中读取共享配置

在 user-service 服务中，修改 PatternProperties 类，读取新添加的属性：

在 user-service 服务中，修改 UserController，添加一个方法：

运行两个 UserApplication，使用不同的 profile

修改 UserApplication2 这个启动项，改变其 profile 值：

这样，UserApplication (8081) 使用的 profile 是 dev，UserApplication2 (8082) 使用的 profile 是 test

启动 UserApplication 和 UserApplication2

访问地址：http://localhost:8081/user/prop，结果：

访问地址：http://localhost:8082/user/prop，结果：

可以看出来，不管是 dev，还是 test 环境，都读取到了 envSharedValue 这个属性的值。

上面的都是同一个微服务下，那么不同微服务之间可以环境共享吗？

通过下面的两种方式来指定：

• extension-configs
• shared-configs

spring: 
  cloud:
    nacos:
      config:
        file-extension: yaml # 文件后缀名
        extends-configs: # 多微服务间共享的配置列表
          - dataId: common.yaml # 要共享的配置文件id

spring: 
  cloud:
    nacos:
      config:
        file-extension: yaml # 文件后缀名
        shared-configs: # 多微服务间共享的配置列表
          - dataId: common.yaml # 要共享的配置文件id

# 配置优先级

当 nacos、服务本地同时出现相同属性时，优先级有高低之分。

更细致的配置

# Nacos 集群

所以我们先__模拟 (只是模拟，real life 可能不一样)__三个 nacos 节点:

注意点：修改两个配置文件：

• 修改 cluster.conf
  

上面那些就是集群中每一个 nacos 节点的信息

• 修改 Nacos 的 application.properties（不是你的 application.properties）

修改完成后保存即可。

• 使用 Nginx 对 Nacos 做反向代理
  这里需要 Nginx 前置知识，可以看我以下这一篇文章：Nginx 入门：通俗理解反向代理和负载均衡，简单配置 Nginx

然后在我们微服务的配置文件中

如果你的 Nacos 配置集群死活报下图的错误：

请检查你的 MySQL 版本，需要在 5.7 及以上，而且在 8.0 以下（比较苛刻）

这样我们在 nacos 服务端的页面里面写配置文件，这些都是会存 (持久化) 到我们连接的 mysql (集群) 里的

---

# Feign 远程调用

案例地址：https://gitee.com/xn2001/cloudcode/tree/master/06-cloud-feign

我们以前利用 RestTemplate 发起远程调用的代码：

• 代码可读性差，编程体验不统一
• 参数复杂 URL 难以维护

Feign 是一个声明式的 http 客户端，官方地址：https://github.com/OpenFeign/feign

其作用就是帮助我们优雅的实现 http 请求的发送，解决上面提到的问题。

# Feign 使用

引入依赖

我们在 order-service 引入 feign 依赖：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>

添加注解

在 order-service 启动类添加注解开启 Feign

请求接口

在 order-service 中新建一个接口，内容如下

package cn.itcast.order.client;

import cn.itcast.order.pojo.User;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;

@FeignClient("userservice")
public interface UserClient {
    @GetMapping("/user/{id}")
    User findById(@PathVariable("id") Long id);
}

@FeignClient("userservice") ：其中参数填写的是 (提供者) 微服务名

@GetMapping("/user/{id}") ：其中参数填写的是请求路径

这个客户端主要是基于 SpringMVC 的注解 @GetMapping 来声明远程调用的信息

Feign 可以帮助我们发送 http 请求，无需自己使用 RestTemplate 来发送了。

就这么在消费者微服务里面写个这么一个接口就行了

之后在我们的 orderservice 微服务里面要是用的话就直接拿这个接口 UserClient 调这个方法 findById 就行了

测试

@Autowired
private UserClient userClient;

public Order queryOrderAndUserById(Long orderId) {
    // 1.查询订单
    Order order = orderMapper.findById(orderId);
    // TODO: 2021/8/20 使用feign远程调用
    User user = userClient.findById(order.getUserId());
    // 3. 将用户信息封装进订单
    order.setUser(user);
    // 4.返回
    return order;
}

我们要是要用只要声明那个接口类的属性然后自动注入 (IOC 容器会找到 openfeign 实现这个接口的实现类的对象注册到 IOC 容器然后自动注入到我们这里的这个属性)

然后我们只管调用方法就行，把那个 id 值传进去，然后 feign 帮我们按照我们写的那个接口里面那个方法的设置的请求方式，地址，参数，返回值等发送请求并把结果作为给我们的调用那个方法的结果，我们接着可以用各种操作

feign 底层已经集合了 ribbon, 可以做到 load balancing

# 自定义配置

Feign 可以支持很多的自定义配置，如下表所示：

类型	作用	说明
feign.Logger.Level	修改日志级别	包含四种不同的级别：NONE、BASIC、HEADERS、FULL
feign.codec.Decoder	响应结果的解析器	http 远程调用的结果做解析，例如解析 json 字符串为 java 对象
feign.codec.Encoder	请求参数编码	将请求参数编码，便于通过 http 请求发送
feign.Contract	支持的注解格式	默认是 SpringMVC 的注解
feign.Retryer	失败重试机制	请求失败的重试机制，默认是没有，不过会使用 Ribbon 的重试

一般情况下，默认值就能满足我们使用，如果要自定义时，只需要创建自定义的 @Bean 覆盖默认 Bean 即可。下面以日志为例来演示如何自定义配置。

基于配置文件修改 feign 的日志级别可以针对单个服务：

feign:  
  client:
    config: 
      userservice: # 针对某个微服务的配置,这里是对userservice微服务发送的请求
        loggerLevel: FULL #  日志级别` 

也可以针对所有服务：

feign:  
  client:
    config: 
      default: # 这里用default就是全局配置，如果是写服务名称，则是针对某个微服务的配置
        loggerLevel: FULL #  日志级别` 

而日志的级别分为四种：

• NONE：不记录任何日志信息，这是默认值。
• BASIC：仅记录请求的方法，URL 以及响应状态码和执行时间
• HEADERS：在 BASIC 的基础上，额外记录了请求和响应的头信息
• FULL：记录所有请求和响应的明细，包括头信息、请求体、元数据

也可以基于 Java 代码来修改日志级别，先声明一个类，然后声明一个 Logger.Level 的对象

public class DefaultFeignConfiguration {
    @Bean
    public Logger.Level feignLogLevel(){
        return Logger.Level.BASIC; // 日志级别为BASIC
    }
}

如果要全局生效，将其放到启动类的 @EnableFeignClients 这个注解中：

@EnableFeignClients(defaultConfiguration = DefaultFeignConfiguration .class)

如果是局部生效，则把它放到对应的 @FeignClient 这个注解中：

@FeignClient(value = "userservice", configuration = DefaultFeignConfiguration .class) 

# 性能优化

Feign 底层发起 http 请求，依赖于其它的框架。其底层客户端实现有：

• URLConnection：默认实现，不支持连接池
• Apache HttpClient ：支持连接池
• OKHttp：支持连接池

因此提高 Feign 性能的主要手段就是使用连接池代替默认的 URLConnection

另外，日志级别应该尽量用 basic/none，可以有效提高性能。

这里我们用 Apache 的 HttpClient 来演示连接池。

在 order-service 的 pom 文件中引入 HttpClient 依赖

<!--httpClient的依赖 -->
<dependency>
    <groupId>io.github.openfeign</groupId>
    <artifactId>feign-httpclient</artifactId>
</dependency>

配置连接池

在 order-service 的 application.yml 中添加配置

feign:
  client:
    config:
      default: # default全局的配置
        loggerLevel: BASIC # 日志级别，BASIC就是基本的请求和响应信息
  httpclient:
    enabled: true # 开启feign对HttpClient的支持
    max-connections: 200 # 最大的连接数
    max-connections-per-route: 50 # 每个路径的最大连接数

在 FeignClientFactoryBean 中的 loadBalance 方法中打断点

Debug 方式启动 order-service 服务，可以看到这里的 client，底层就是 HttpClient

日志级别一般是 none 或者 basic, 太高有损性能

# 最佳实践

# 继承方式

我们之前那种方式可以看到那个 orderservice 里面写的那个 userCilent 接口里面那个方法以及注解跟我们 userservice 里面的 controller 层的 userController 对应的方法很像 (Request 地址，方式，参数，返回值等等等)

所以我们干脆做个接口让他们都继承 / 实现

• 我们 orderservice 里面写的那个 userCilent 接口继承这个新写的接口
• 我们 userservice 里面的 controller 层的 userController 实现这个新写的接口

更省事了 (-> 有个接口定义了规范)

但是也有问题这种方式

一样的代码可以通过继承来共享：

1）定义一个 API 接口，利用定义方法，并基于 SpringMVC 注解做声明

2）Feign 客户端、Controller 都集成该接口

优点

• 简单
• 实现了代码共享

缺点

• 服务提供方、服务消费方紧耦合
• 参数列表中的注解 (SpringMVC 的) 映射并不会继承，因此 Controller 中必须再次声明方法、参数列表、注解

# 抽取方式

将 FeignClient 抽取为独立模块，并且把接口有关的 POJO (就是你消费者发送请求返回的信息存到这个 pojo (任意你写的) 对象里面), 所以干脆把 pojo 也写在 feign-api 模块、默认的 Feign 配置都放到这个模块中，提供给所有消费者使用。

例如：将 UserClient、User、Feign 的默认配置都抽取到一个 feign-api 包中，所有微服务引用该依赖包，即可直接使用。

这样就不用在每一个需要调用 userservice 服务的消费者微服务写 userClient 接口，以及 pojo 比如说 User 对象装着请求 userservice 返回的信息和 feign 的一些配置 (性能优化啥的等等), 等等等其他的，只需要在 feign-api 模块写一次就足够

但这种方式也有坏处，比如说我们 orderservice 服务只需要用一个 userClient 的方法但是我们还是需要把它全部引入进来

接下来我们就用该方法在代码中实现上面说的第二种方式 (feign-api 独立模块)

首先创建一个 module，命名为 feign-api

在 feign-api 中然后引入依赖

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>

order-service 中的 UserClient、User 都复制到 feign-api 项目中

在 order-service 中使用 feign-api

首先，删除 order-service 中的 UserClient、User

在 order-service 中引入 feign-api

<dependency>
    <groupId>com.xn2001.feign</groupId>
    <artifactId>feign-api</artifactId>
    <version>1.0</version>
</dependency>

修改注解

当定义的 FeignClient 不在 SpringBootApplication 的扫描包范围下时，这些 FeignClient 就不能使用。

修改 order-service 启动类上的 @EnableFeignClients 注解

@EnableFeignClients(basePackages = "com.xn2001.feign.clients")

---

# Gateway 网关

案例地址：https://gitee.com/xn2001/cloudcode/tree/master/07-cloud-gateway

Spring Cloud Gateway 是 Spring Cloud 的一个全新项目，该项目是基于 Spring 5.0，Spring Boot 2.0 和 Project Reactor 等响应式编程和事件流技术开发的网关，它旨在为微服务架构提供一种简单有效的统一的 API 路由管理方式。

Gateway 网关是我们服务的守门神，所有微服务的统一入口。

网关可以看作是一个特殊微服务，需要注册到 nacos，仅此而已。其余的就是跟微服务群打交道了，没 nacos 什么事了，微服务群自己又会跟 nacos 打交道，服务注册服务拉取什么的。

网关的核心功能特性：

• 请求路由
• 权限控制
• 限流

权限控制：网关作为微服务入口，需要校验用户是是否有请求资格，如果没有则进行拦截。

路由和负载均衡：一切请求都必须先经过 gateway，但网关不处理业务，而是根据某种规则，把请求转发到某个微服务，这个过程叫做路由。当然路由的目标服务有多个时，还需要做负载均衡 (这跟 ribbon 不一样，ribbon 是微服务之间的调用)。

nginx,ribbon,gateway 这三个负载均衡的对象都是不同的，比如网关的负载均衡是面向用户访问微服务，Ribbon 是面向微服务之间，Nginx 是面向 Nacos (nacos 的客户端 -> 微服务注册 (或者请求服务获取对应的服务列表等等等) 到某一个 nacos 节点) 的

限流：当请求流量过高时，在网关中按照下流的微服务能够接受的速度来放行请求，避免服务压力过大。

在 SpringCloud 中网关的实现包括两种：

• gateway
• zuul

Zuul 是基于 Servlet 实现，属于阻塞式编程。而 Spring Cloud Gateway 则是基于 Spring5 中提供的 WebFlux，属于响应式编程的实现，具备更好的性能。

# 入门使用

1. 创建 SpringBoot 工程 gateway，引入网关依赖
2. 编写启动类
3. 编写基础配置和路由规则
4. 启动网关服务进行测试

<!--网关-->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-gateway</artifactId> #自动装配网关的各种功能
</dependency>
<!--nacos服务发现依赖-->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>

网关服务本身也是个微服务，也需要把自己注册到 nacos (或者其他的) 注册中心，他自己也需要注册或者拉去他需要调用的服务，所以也需要引入 nacos 发现服务依赖

在网关微服务 (module) 里面创建 application.yml 文件，内容如下：

server:
  port: 10010 # 网关端口
spring:
  application:
    name: gateway # 服务名称
  cloud:
    nacos:
      server-addr: localhost:8848 # nacos地址 ,上面到这的配置是让网关注册到nacos以及可以联系上nacos实现服务发现 
    gateway:
      routes: # 网关路由配置
        - id: user-service # 路由id，自定义，只要唯一即可
          # uri: http://127.0.0.1:8081 # 路由的目标地址 http就是固定地址,这种方式也行但是写死了
          uri: lb://userservice # 路由的目标地址 lb就是负载均衡，后面跟服务名称,其实就是按照这个服务名去nacos找这个服务的所有实例的列表,然后负载均衡选一个
          predicates: # 路由断言，也就是判断请求是否符合路由规则的条件
            - Path=/user/** # 这个是按照路径匹配，只要以/user/开头就符合要求

当然我们还需要一个开启类开启当前网关服务，就是普通的那种 main 函数有个 SpringBootApplication 注解的那个，这里没写

我们将符合 Path 规则的一切请求，都代理到 uri 参数指定的地址。

上面的例子中，我们将 /user/** 开头的请求，代理到 lb://userservice ，其中 lb 是负载均衡 (LoadBalance)，根据服务名拉取服务列表，实现负载均衡。

重启网关，访问 http://localhost:10010 (这个是网关的端口)/user/1 时，符合 /user/** 规则，请求__转发__到我们设置的 uri：http://userservice/user/1

多个 predicates 的话，要同时满足规则，下文有例子。

# 流程图

路由配置包括：

1. 路由 id：路由的唯一标示
2. 路由目标（uri）：路由的目标地址，http 代表固定地址，lb 代表根据服务名负载均衡
3. 路由断言（predicates）：判断路由的规则
4. 路由过滤器（filters）：对请求或响应做处理

# 断言工厂

我们在配置文件中写的断言规则只是字符串，这些字符串会被 Predicate Factory (断言工厂) 读取并处理，转变为路由判断的条件。

例如 Path=/user/** 是按照路径匹配，这个规则是由

org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory 类来读取和处理的，像这样的断言工厂在 Spring Cloud Gateway 还有十几个，每一个都有自己的判断规则和条件

名称	说明	示例
After	是某个时间点后的请求	- After=2037-01-20T17:42:47.789-07:00[America/Denver]
Before	是某个时间点之前的请求	- Before=2031-04-13T15:14:47.433+08:00[Asia/Shanghai]
Between	是某两个时间点之前的请求	- Between=2037-01-20T17:42:47.789-07:00[America/Denver], 2037-01-21T17:42:47.789-07:00[America/Denver]
Cookie	请求必须包含某些 cookie	- Cookie=chocolate, ch.p
Header	请求必须包含某些 header	- Header=X-Request-Id, d+
Host	请求必须是访问某个 host（域名）	- Host= **.somehost.org , **.anotherhost.org
Method	请求方式必须是指定方式	- Method=GET,POST
Path	请求路径必须符合指定规则	- Path=/red/{segment},/blue/**
Query	请求参数必须包含指定参数	- Query=name, Jack 或者 - Query=name
RemoteAddr	请求者的 ip 必须是指定范围	- RemoteAddr=192.168.1.1/24
Weight	权重处理

官方文档：https://docs.spring.io/spring-cloud-gateway/docs/current/reference/html/#gateway-request-predicates-factories

一般的，我们只需要掌握 Path，加上官方文档的例子，就可以应对各种工作场景了。

predicates:
  - Path=/order/**
  - After=2031-04-13T15:14:47.433+08:00[Asia/Shanghai]

像这样的规则，现在是 2021 年 8 月 22 日 01:32:42，很明显 After 条件不满足，可以不会转发，路由不起作用，找不到对应的路由就会报错。

# 过滤器工厂

GatewayFilter 是网关中提供的一种过滤器，可以对进入网关的请求和微服务返回的响应做处理。

Spring 提供了 31 种不同的路由过滤器工厂。

官方文档：https://docs.spring.io/spring-cloud-gateway/docs/current/reference/html/#gatewayfilter-factories

名称	说明
AddRequestHeader	给当前请求添加一个请求头
RemoveRequestHeader	移除请求中的一个请求头
AddResponseHeader	给响应结果中添加一个响应头
RemoveResponseHeader	从响应结果中移除有一个响应头
RequestRateLimiter	限制请求的流量

下面我们以 AddRequestHeader 为例：

需求：给所有进入 userservice 的请求添加一个请求头： sign=xn2001.com is eternal (sign 是 key,xn2001.com is eternal 是 value)

只需要修改 gateway 服务的 application.yml 文件，添加路由过滤即可。

spring:
  cloud:
    gateway:
      routes: # 网关路由配置
        - id: user-service # 路由id，自定义，只要唯一即可
          # uri: http://127.0.0.1:8081 # 路由的目标地址 http就是固定地址
          uri: lb://userservice # 路由的目标地址 lb就是负载均衡，后面跟服务名称
          predicates: # 路由断言，也就是判断请求是否符合路由规则的条件
            - Path=/user/** # 这个是按照路径匹配，只要以/user/开头就符合要求
          filters:
            - AddRequestHeader=sign, xn2001.com is eternal # 添加请求头

如何验证，我们修改 userservice 中的一个接口

@GetMapping("/{id}")
public User queryById(@PathVariable("id") Long id, @RequestHeader(value = "sign", required = false) String sign) {
    log.warn(sign);
    return userService.queryById(id);
}

重启两个服务，访问：http://localhost:10010/user/1

可以看到控制台打印出了这个请求头

当然，Gateway 也是有全局过滤器的，如果要对所有的路由都生效，则可以将过滤器工厂写到 default-filters 下：

spring:
  cloud:
    gateway:
      default-filters: # 对所有请求的微服务都会生效,都会带上这里(过滤器配置的)添加的请求头
        - AddRequestHeader=sign, xn2001.com is eternal # 添加请求头

# 全局过滤器

上面介绍的过滤器工厂，网关提供了 31 种，但每一种过滤器的作用都是固定的。如果我们希望拦截请求，做自己的业务逻辑则没办法实现。

全局过滤器的作用也是处理一切进入网关的请求和微服务响应，与 GatewayFilter 的作用一样。区别在于 GlobalFilter 的逻辑可以写代码来自定义规则；而 GatewayFilter 通过配置定义，处理逻辑是固定的。

需求： 定义全局过滤器，拦截请求，判断请求的参数是否满足下面条件

• 参数中是否有 authorization
• authorization 参数值是否为 admin

如果同时满足则放行，否则拦截。

@Component
public class AuthorizeFilter implements GlobalFilter, Ordered {

    // 测试：http://localhost:10010/order/101?authorization=admin
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        // 获取第一个 authorization 参数
        String authorization = exchange.getRequest().getQueryParams().getFirst("authorization");
        if ("admin".equals(authorization)){
            // 放行
            return chain.filter(exchange);
        }
        // 设置拦截状态码信息
        exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
        // 设置拦截
        return exchange.getResponse().setComplete();
    }

    // 设置过滤器优先级，值越低优先级越高
    // 也可以在这个类上面使用 @Order 注解然后给个值
    @Override
    public int getOrder() {
        return 0;
    }
}

这里的 order 就像是 spring 的 AOP->aspectJ 的那些学过的东西

# 过滤器顺序

请求进入网关会碰到三类过滤器：DefaultFilter、当前路由的过滤器、GlobalFilter；

请求路由后，会将三者合并到一个过滤器链（集合）中，排序后依次执行每个过滤器.

排序的规则是什么呢？

• 每一个过滤器都必须指定一个 int 类型的 order 值，order 值越小，优先级越高，执行顺序越靠前。

• GlobalFilter 通过实现 Ordered 接口然后实现那个接口的方法，或者使用 @Order 注解来指定 order 值，由我们自己指定。

• 路由过滤器和 defaultFilter 的 order 由 Spring 指定，默认是按照声明顺序从 1 递增

  • 比如说你一个微服务有两个路由过滤器，第一个写的就是 1, 第二个写的就是 2。
  • 再比如说如果你写的 defaultFilter (全局，不分哪个微服务), 第一个写的就是 1, 第二个写的就是 2 (上面写的具体的微服务的路由器不影响这里的这个)

• 当过滤器的 order 值一样时，会按照 defaultFilter > 路由过滤器 > GlobalFilter 的顺序执行。

# 跨域问题

所有对我们的微服务的访问都要通过我们的网关，所以直接给网关设置跨域问题的解决等就行了

不了解跨域问题的同学可以百度了解一下；在 Gateway 网关中解决跨域问题还是比较方便的。

不过不确定到底是不是请求被拦截了还是发出去但是服务器不允许的，这个不清楚

spring:
  cloud:
    gateway:
      globalcors: # 全局的跨域处理
        add-to-simple-url-handler-mapping: true # 解决options请求被拦截问题(就是允许浏览器发请求问我们服务器允不允许跨域,而不是原来会跨域的ajax请求)
        corsConfigurations:
          '[/**]': #这里/**就是拦截所有的请求,一切进入网关的请求都做跨域处理(解决或者.)
            allowedOrigins: # 允许哪些网站的跨域请求 allowedOrigins: “*” 允许所有网站
              - "http://localhost:8090"
            allowedMethods: # 允许的跨域ajax的请求方式
              - "GET"
              - "POST"
              - "DELETE"
              - "PUT"
              - "OPTIONS"
            allowedHeaders: "*" # 允许在请求中携带的头信息
            allowCredentials: true # 是否允许携带cookie
            maxAge: 360000 # 这次跨域检测的有效期(就是浏览器在第一询问请求结束后然后我们说可以跨域,之后要是还有ajax请求,只要在我们这里设置的有效期内,浏览器就不用再发那个询问请求了)

---

# RabbitMQ

案例地址：https://gitee.com/xn2001/cloudcode/tree/master/08-rabbitmq-demo

# 同步异步通讯

微服务间通讯有同步和异步两种方式

同步通讯：就像打电话，需要实时响应。

异步通讯：就像发邮件，不需要马上回复。

两种方式各有优劣，打电话可以立即得到响应，但是你却不能跟多个人同时通话。发送邮件可以同时与多个人收发邮件，但是往往响应会有延迟。

我们之前学习的 Feign 调用就属于同步方式，虽然调用可以实时得到结果，但存在下面的问题：

同步调用的优点：

• 时效性较强，可以立即得到结果

同步调用的缺点：

• 耦合度高
• 性能和吞吐能力下降
• 有额外的资源消耗
• 有级联失败问题

异步调用则可以避免上述问题，我们以购买商品为例，用户支付后需要调用订单服务完成订单状态修改，调用物流服务，从仓库分配响应的库存并准备发货。在事件模式中，支付服务是事件发布者（publisher），在支付完成后只需要发布一个支付成功的事件（event），事件中带上订单 id。订单服务和物流服务是事件订阅者（Consumer），订阅支付成功的事件，监听到事件后完成自己业务即可。

所以跟我们之前学到的微服务之间的调用不太一样

我们这里更像是有一个生产者 (微服务) 被用户通过网关然后访问到了，然后这个微服务里面的操作需要影响到别的微服务，这个时候就可以把自己的一些信息然后各种细节等等交给 broker, 由 broker 把这个消息按照我们生产者里面设置的交换机和队列等等把它交给对应的队列，然后队列会把信息交给对应的微服务，那个微服务接收到那个消息就可以操作了

这都是异步的

这个生产者有点像我们之前的提供者 (提供接口给其他微服务使用), 然后消费者有点像我们的之前的消费者 (也是消费者) 但好像并不是！！！

这里消息队列的概念好像就是因为某种原因，微服务会把一些关于他自己的 (或者其他…) 消息交给 broker, 然后其他微服务就可以用到这个消息

为了解除事件发布者与订阅者之间的耦合，两者并不是直接通信，而是有一个中间人（Broker）。发布者发布事件到 Broker，不关心谁来订阅事件。订阅者从 Broker 订阅事件，不关心谁发来的消息。

Broker 是一个像数据总线一样的东西，所有的服务要接收数据和发送数据都发到这个总线上，这个总线就像协议一样，让服务间的通讯变得标准和可控。

• 这里的支付服务是指支付成功后的后续操作（支付成功后才会产生订单以及一系列的操作）
• 也就是说只有支付成功后请求才会被发送到这个服务，如果支付失败，在第一步调用接口时就会返回失败
• 这里要表示的就是，用户只关心支付成功没有，也就是说只关心第一步成功没有
• 至于后面的订单或者是短信通知相对来说不重要了，因为用户知道钱已经给了，如果有问题就再说
• 最简单的例子就是淘宝买东西，你一般都是付款之后就不会再看了，因为你知道支付成功了

异步调用好处：

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

异步调用缺点：

• 架构复杂了，业务没有明显的流程线，不好管理
• 需要依赖于 Broker 的可靠、安全、性能

# MQ 消息队列

MQ，中文是消息队列（MessageQueue），字面来看就是存放消息的队列，也就是事件驱动架构中的 Broker

比较常见的 MQ 实现：

• ActiveMQ
• RabbitMQ
• RocketMQ
• Kafka

几种常见 MQ 的对比：

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司 / 社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP、XMPP、SMTP、STOMP	OpenWire、STOMP、REST、XMPP、AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

以 RabbitMQ 为例，我们在 Centos7 虚拟机中使用 Docker 来安装

在线拉取镜像

docker pull rabbitmq:3-management

执行下面的命令来运行 MQ 容器

docker run \
 -e RABBITMQ_DEFAULT_USER=admin \
 -e RABBITMQ_DEFAULT_PASS=123456 \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

启动成功后访问地址：http://192.168.211.128:15672

RabbitMQ 中的一些角色

• publisher：生产者
• consumer：消费者
• exchange：交换机，负责消息路由
• queue：队列，存储消息
• virtualHost：虚拟主机，隔离不同租户的 exchange、queue、消息的隔离

MQ 的基本结构

# 入门案例

RabbitMQ 官方提供了 5 个不同的 Demo 示例，对应了不同的消息模型。

Hello World 模型

官方的 HelloWorld 是基于最基础的消息队列模型来实现的，只包括三个角色：

• publisher：消息发布者，将消息发送到队列 queue
• queue：消息队列，负责接受并缓存消息
• consumer：订阅队列，处理队列中的消息

# publisher 实现

• 建立连接
• 创建 channel
• 声明队列
• 发送消息
• 关闭连接和 channel

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.211.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("admin");
        factory.setPassword("123456");
        // 1.2.建立连接
        Connection connection = factory.newConnection();
        // 2.创建通道Channel
        Channel channel = connection.createChannel();
        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);
        // 4.发送消息
        String message = "Hello RabbitMQ！";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功：[" + message + "]");
        // 5.关闭通道和连接
        channel.close();
        connection.close();
    }
}

# consumer 实现

• 建立连接
• 创建 channel
• 声明队列
• 订阅消息

public class ConsumerTest {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.211.128");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("admin");
        factory.setPassword("123456");
        // 1.2.建立连接
        Connection connection = factory.newConnection();
        // 2.创建通道Channel
        Channel channel = connection.createChannel();
        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);
        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息：[" + message + "]");
            }
        });
        System.out.println("等待接收消息中");
    }
}

注意上面 "等待接收消息中" 会被先打印，然后才是 "接收到消息：[Hello RabbitMQ!]"

这是因为我们那个正常的 sout 是同步的，basicConsume 那个 (函数) 参数是__回调__, 异步的！！！

# SpringAMQP

SpringAMQP 是基于 RabbitMQ 封装的一套模板，并且还利用 SpringBoot 对其实现了自动装配，使用起来非常方便。

SpringAMQP 的官方地址：https://spring.io/projects/spring-amqp

SpringAMQP 提供了三个功能：

• 自动声明队列、交换机及其绑定关系
• 基于注解的监听器模式，异步接收消息
• 封装了 RabbitTemplate 工具，用于发送消息

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

# BasicQueue

首先配置 MQ 地址，在 publisher、consumer 服务中的 application.yml 中添加配置

spring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: admin # 用户名
    password: 123456 # 密码

在 consumer 服务中添加监听队列

@Component
public class RabbitMQListener {
    @RabbitListener(queues = "simple.queue") // 消费者这个注解代表在你指定的队列监听消息
    // 然后下面这个方法参数可以获得队列里面的消息,等等等->发送的是什么类型的对象接收到的就是什么类型的对象
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("消费者接收到消息：【" + msg + "】");
    }
}

在 publisher 服务中添加发送消息的测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired //直接自动注入然后用里面的方法指定交换机队列等等等发消息就可以
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "你好啊，乐心湖！";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

# WorkQueue

Work queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。

当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。

此时就可以使用 work 模型，多个消费者共同处理消息处理，速度就能大大提高了。

我们循环发送，模拟大量消息堆积现象，在 publisher 服务中的 SpringAmqpTest 类中添加一个测试方法：

/**
 * workQueue
 * 向队列中不停发送消息，模拟消息堆积。
 */
@Test
public void testWorkQueue() throws InterruptedException {
    // 队列名称
    String queueName = "simple.queue";
    // 消息
    String message = "hello, message_";
    for (int i = 0; i < 50; i++) {
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}

消息接收

要模拟多个消费者绑定同一个队列，我们在 consumer 服务的 RabbitMQListener 中添加 2 个新的方法：

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}

启动 ConsumerApplication 后，在执行 publisher 服务中刚刚编写的发送测试方法 testWorkQueue

可以看到消费者 1 很快完成了自己的 25 条消息。消费者 2 却在缓慢的处理自己的 25 条消息。

也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。这是因为 RabbitMQ 默认有一个消息预取机制，显然这不是我们想要的结果，我们需要的是能者多劳嘛，所以去限制每次只能取一条消息，可以解决这个问题 (默认的是轮询)。

在 spring 中有一个简单的配置，设置 prefetch 属性，我们修改 consumer 服务的 application.yml 文件，添加配置

默认的预取就是轮询队列中的消息给每个 (对应的) 消费者发一个 (注意这个还是只能给一个消费者发送，而不是给多个，多个的话需要发布订阅 (这个也不是发布确认！！！ 不同东西，看 rabbitmq 笔记！)), 也可以认为就是每个消费者都会挨个一个一个预取，直到队列中的消息没了，就算当前消费者自己当前消息还没处理完毕也会是预取，所以因为默认轮询最后都一样的数量

我们需要能者多劳，可以 prefetch 设置为 1, 也就是这个消费者说只能预取 1 个，只能当前消费者处理完成才能获取下一个消息，达成不公平的效果 (看 rabbitmq 专门的笔记)

在消费者里面的配置文件中:

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

Work 模型的使用：

• 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
• 通过设置 prefetch 来控制消费者预取的消息数量

# 发布 / 订阅

这个不是发布确认

发布订阅就是主要是关于交换机的，交换机可以给匹配 RoutingKey 的队列发送消息，达成如果我们发布了一个消息，我们可以让想要的某一个或者多个消费者订阅 (获取到) 到那个消息，(有点像 vue2 里面的那个 pubsub 的意思)

图中可以看到，在订阅模型中，多了一个 exchange 角色，而且过程略有变化

• Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给 exchange（交换机）

• Consumer：消费者，与以前一样，订阅队列，没有变化

• Queue：消息队列也与以前一样，接收消息、缓存消息

• Exchange：交换机，一方面，接收生产者发送的消息；另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于 Exchange 的类型。Exchange 有以下 3 种类型：

  • Fanout：广播，将消息交给所有绑定到交换机的队列
  • Direct：定向，把消息交给符合指定 routing key 的队列
  • Topic：通配符，把消息交给符合 routing pattern（路由模式） 的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

我们可能会使用备份交换机

或者发布确认来达到交换机没找到符合路由规则的队列消息可能会丢失的问题！

# Fanout

Fanout，英文翻译是扇出，在 MQ 中我们也可以称为广播。

在广播模式下，消息发送流程是这样的：

• 可以有多个队列
• 每个队列都要绑定到 Exchange（交换机）
• 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
• 交换机把消息发送给绑定过的所有队列
• 订阅队列的消费者都能拿到消息

接下里我们用 SpringAMQP 来简单实现 FanoutExchange

1. 在 consumer 服务中，利用代码声明队列、交换机，并将两者绑定
2. 在 consumer 服务中，编写两个消费者方法，分别监听 fanout.queue1 和 fanout.queue2
3. 在 publisher 中编写测试方法，向 fanout 发送消息

声明队列和交换机

Spring 提供了一个接口 Exchange，来表示所有不同类型的交换机。

在 consumer 中创建一个类，声明队列、交换机、绑定对象 Binding

@Configuration
public class FanoutConfig {

    /**
     * 声明交换机
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("xn2001.fanout");
    }

    /**
     * 声明队列
     * @return Queue
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1(FanoutExchange fanoutExchange,Queue fanoutQueue1){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
    @Bean
    public Binding bindingQueue2(FanoutExchange fanoutExchange,Queue fanoutQueue2){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }

}

通过这样 @Bean 的方式来申明确实比较麻烦，其实我们也是可以直接通过 @RabbitListener 注解来完成的，代码如下：

在 consumer 服务的 SpringRabbitListener 中添加三个方法，作为消费者

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) throws InterruptedException {
    System.out.println("接收到fanout.queue1的消息：【" + msg + "】" + LocalTime.now());
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) throws InterruptedException {
    System.err.println("接收到fanout.queue2的消息：【" + msg + "】" + LocalTime.now());
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "fanout.queue3"),
    exchange = @Exchange(value = "xn2001.fanout",type = "fanout")
))
public void listenFanoutQueue3(String msg) {
    System.out.println("接收到fanout.queue3的消息：【" + msg + "】" + LocalTime.now());
}

在 publisher 服务的 SpringAmqpTest 类中添加测试方法

/**
 * fanout
 * 向交换机发送消息
 */
@Test
public void testFanoutExchange() {
    // 交换机名称
    String exchangeName = "xn2001.fanout";
    // 消息
    String message = "hello, everybody!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

运行该方法，可以发现 fanout.queue1、fanout.queue2 都收到了交换机的消息。

总结一下：

交换机的作用是什么？

• 接收 publisher 发送的消息
• 将消息按照规则路由到与之绑定的队列
• 不能缓存消息，路由失败，消息丢失
• FanoutExchange 会将所有消息路由到每个绑定的队列

# Direct

在 Fanout 模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到 DirectExchange

在 Direct 模型下：

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey （路由 key）
• 消息的发送方向 Exchange 发送消息时，也必须指定消息的 RoutingKey 。
• Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routingkey 与消息的 Routing key 完全一致，才会接收到消息

在 consumer 的 SpringRabbitListener 中添加两个消费者，同时基于注解来声明队列和交换机

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "direct.queue1"),
    exchange = @Exchange(value = "xn2001.direct"),
    key = {"a","b"}
))
public void listenDirectQueue1(String msg){
    System.out.println("接收到direct.queue1的消息：【" + msg + "】" + LocalTime.now());
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "direct.queue2"),
    exchange = @Exchange(value = "xn2001.direct"),
    key = {"a","c"}
))
public void listenDirectQueue2(String msg){
    System.out.println("接收到direct.queue2的消息：【" + msg + "】" + LocalTime.now());
}

在 publisher 服务的 SpringAmqpTest 类中添加测试方法

/**
 * direct
 * 向交换机发送消息
 */
@Test
public void testDirectExchangeToA() {
    // 交换机名称
    String exchangeName = "xn2001.direct";
    // 消息
    String message = "hello, i am direct to a!";
    rabbitTemplate.convertAndSend(exchangeName, "a", message);
}

/**
 * direct
 * 向交换机发送消息
 */
@Test
public void testDirectExchangeToB() {
    // 交换机名称
    String exchangeName = "xn2001.direct";
    // 消息
    String message = "hello, i am direct to b!";
    rabbitTemplate.convertAndSend(exchangeName, "b", message);
}

# Topic

Topic 与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型可以让队列在绑定 Routing key 的时候使用通配符！

通配符规则：

# ：匹配一个或多个词

* ：只能匹配一个词

例如：

item.# ：能够匹配 item.spu.insert 或者 item.spu

item.* ：只能匹配 item.spu

• Queue1：绑定的是 china.# ，因此凡是以 china. 开头的 routing key 都会被匹配到。包括 china.news 和 china.weather
• Queue2：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配。包括 china.news 和 japan.news

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "topic.queue1"),
    exchange = @Exchange(value = "xn2001.topic",type = ExchangeTypes.TOPIC),
    key = {"china.#"}
))
public void listenTopicQueue1(String msg){
    System.out.println("接收到topic.queue1的消息：【" + msg + "】" + LocalTime.now());
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(value = "topic.queue2"),
    exchange = @Exchange(value = "xn2001.topic",type = ExchangeTypes.TOPIC),
    key = {"china.*"}
))
public void listenTopicQueue2(String msg){
    System.out.println("接收到topic.queue2的消息：【" + msg + "】" + LocalTime.now());
}

/**
 * topic
 * 向交换机发送消息
 */
@Test
public void testTopicExchange() {
    // 交换机名称
    String exchangeName = "xn2001.topic";
    // 消息
    String message1 = "hello, i am topic form china.news";
    String message2 = "hello, i am topic form china.news.2";
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message1);
    rabbitTemplate.convertAndSend(exchangeName, "china.news.2", message2);
}

# 消息转换器

Spring 会把你发送的消息序列化为字节发送给 MQ (所以队列存的是 java 对象序列化 (所以可以发任何对象而不只是 String which stored as bytes (字节), 对象也可以传只不过被序列化了))，接收消息的时候，还会把字节反序列化为 Java 对象。

默认情况下 Spring 采用的序列化方式是 JDK 序列化。

我们可以去试一下效果

@RabbitListener(queuesToDeclare = @Queue(value = "object.queue"))
public void listenObjectQueue(Map<String,Object> msg) throws InterruptedException {
    System.err.println("object接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}

@Test
public void testSendMap() {
    // 准备消息
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "Jack");
    msg.put("age", 21);
    // 发送消息
    rabbitTemplate.convertAndSend("object.queue", msg);
}

众所周知，JDK 序列化存在下列问题：

• 数据体积过大
• 有安全漏洞
• 可读性差

我们推荐可以使用 JSON 来序列化

在 publisher 和 consumer 两个服务中都引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

配置消息转换器。

在各自的启动类中添加一个 Bean 即可

@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter();
}

当然，接收 json 也需要配置:

# 总结一下子:

• 消息中间件就是为了一个微服务想要把某一个消息 (可以是任何内容不过一般是关于自己的一部分内容的消息) 发送到别的微服务

• rabbitmq 这个消息中间件可以为我们干这个 -> 解耦啊，解峰啊，某个微服务崩了不会导致整个系统崩啊等等等

• 正常来说就是一个微服务发送消息给 rabbitmq 消息队列，这个队列可能就一个消费者连着，可能多个连着这一个 (也就是轮询，或者按照能力来预取消息，或者某种自己定的预取值), 这样反正一个队列里面的信息只会被着多个消费者中的一个来消费

• 这个消费者消费消息的方式就是写个回调函数，只有我们消费者自己的同步任务完成了然后接收到来自队列的消息才会执行这个回调函数，当然这个回调函数也会有那个传来的消息等等等

• 为了确保我们 rabbitmq 消息队列发送消息给消费者确实是被消费者消费掉了 (各种处理完成之后), 而不是只是单纯的接收到了消息，我们需要手动确认 (而不是默认的那种一接收到就会跟我们的 rabbitmq 确认收到了), 这是因为 rabbitmq 一看到来自消费者的确认就会把消息从队列中删除，要是我们只是接收到了就发确认那么可能还没处理完消费者崩了，rabbitmq 以为我们已经 ok 了就把那个删了 (不过我不确定 springboot 整合的 rabbitmq 是不是已经帮我们做了手动确认什么的), 不过要是我们改为了手动确认且只有在我们这个消费者各种处理完消息才确认那么如果消费者在消费消息的时候因为某种原因崩了，rabbitmq 就知道消息没有真正被消费，就会把这个消息发给别的当前监听当前这个队列的消费者来处理

• 不过万一我们 rabbitmq 本身崩了怎么办？

  • 首先可以把队列和消息设置为持久化到硬盘，那么如果消息没有收到来自消费者的确认就代表这个消息没有被消费就会持久化到部署 rabbitmq 的服务器的硬盘上
  • 同样我们的队列也需要持久化 (一般默认的好像都是临时队列，就是我们一重启当前的 rabbitmq 节点之前弄的临时队列就没了), 也就是 durable 为 true, 这样就算 rabbitmq 本身崩了，设置为 durable 为 true 的队列因为持久化到了硬盘上再重启的时候还在
    • rabbitmq 笔记里有关于 rabbitmq 节点然后把一个队列镜像到其他 rabbitmq 节点等等等

  不过注意持久化到硬盘也会需要时间要是这期间崩了好像也不行…(不太确定)

• 接着我们可能想要对多个消费者发同一个消息，那么就需要交换机，交换机有 fanout,direct, 和 topic (还有 header 不过不怎么用了)

  • fanout 就是广播，给所有当前交换机绑定的队列都发消息，这样这些队列都有这些消息，对应的消费者都能获得到那个消息 (除了一个队列对应多个消费者理所当然就只能一个消费者才能消费那个消息)
  • direct 就是正常按照 RoutingKey 方式找当前交换机绑定的队列发，一个队列绑定一个交换机可以有多个 RoutingKey, 两个队列绑定同一个交换机这两个队列绑定的 RoutingKey 可以是同一个 (应该可以)
  • topic 就是按照 pattern 找当前交换机绑定的有着对应的 RoutingKey 的队列发消息

• 其实我们生产者 (发消息的) 给 rabbitmq (交换机…so on…), 我们交换机也需要给我们生产者发送一个发布确认的消息，这样就避免我们 rabbitmq 本身崩了，我们生产者还不知道。发布确认有三只用 (单个和批量属于同步，然后异步的方式，主要用的都是异步的)

  • 首先需要开启发布确认 ->channel 设置 confirm 模式 (不确定 springboot 中怎么操作的)

  • 然后如果是为异步的发布确认，那么我们就可以在生产者那边定义回调函数，比如说如果 rabbitmq 给我们返回了发布确认的消息，这个发布确认就会被调用那个消息的发布确认的回调函数 (当然我们有那个消息本身和序号等等等进行我们想要的操作)

    最主要的是如果 rabbitmq 因为某种原因发布失败也会调用我们设置的对应的回调函数 (当然我们有那个消息本身和序号等等等进行我们想要的操作)

  所以生产者这边也有异步操作，不过这个主要是针对安全性 (个人认为)

• 还有就是死信队列，比如说我们一个消息的队列因为那三种原因某一种就可以把消息 pass on to another exchanger which we call as dead letter exchanger, this is just a normal exchanger which will then pass on to a dead letter queue with matching roughtingKey we have given then to any 监听这个死信队列的消费者这个主要是为了如果监听那个一开始队列的消费者处理不了那个消息 (其实就是那三种原因某一种), 那么就会按照我们给的死信交换机等等设置给到另外一个消费者看看能不能处理

• 然后延迟队列就是死信队列一种，主要是消息 TTL 时间过期，发布那个消息时给一个 TTL 时间，然后给一个延迟队列，这个队列没有消费者消费那个消息所以就会一直带着一直到 TTL 时间，一到 TTL 时间就会交给我们设置的死信交换机 -> 死信队列–> 监听这个死信队列的消费者来进行各种判断啊消费这个消息，一般用于我们想要做个定时任务

• 接着就是我们消息发给交换机，交换机可能找不到我们给的 Routingkey 对应的路由 (队列), 一般我们什么都不做的话那个消息会直接被丢掉，我们可以设置 mandatory 参数然后在生产者这边定义回调函数，这个回调函数 (异步) 会在交换机找不到我们给的 Routingkey 对应的队列时调用，当然可以接收到那个消息等等等，这个跟我们说的发布确认不一样，这个只是针对于交换机找不到对应路由 (的队列), 而发布确认更像是交换机本身没有接收 (发布) 那个消息，可能是因为 rabbitmq 崩了等等等

• 我们还有可以给那个交换机设置一个备份交换机，如果那个交换机找不到我们给的 Routingkey 对应的队列，就会把消息 pass on to 我们给他设置的备份交换机，这个备份交换机可以拿消息用来给各种队列发…(比如说给两个队列，一个队列把消息发给用来本分消息的消费者，一个队列把消息发给做警告的消费者)

• 我们还可以给一个消息设置优先级，这样这个更高优先级的消息进入一个队列中，会先从队列中被监听的消费者取出消费

• 还有什么惰性队列就是把消息放到硬盘中，减轻内存什么的，只有消费者要消费对应的消息时才会被加载到内存中 (好像是，这个了解的不清楚)

• 幂等性 -> 可能某种原因消息被消费者多次消费，我们可以唯一 ID + 指纹码机制然后判断到底有没有已经消费过和这个消息啊方式解决，或者是 Redis 原子性方式解决

• 接着就是多个 rabbitmq 节点 -> 集群啊啥的

---

# ELasticsearch

案例地址：https://gitee.com/xn2001/cloudcode/tree/master/09-elasticsearch-hotel-demo

elasticsearch 是一款非常强大的开源搜索引擎，具备非常多强大功能，可以帮助我们从海量数据中快速找到需要的内容，可以用来实现搜索、日志统计、分析、系统监控等功能。

# 倒排索引

首先，倒排索引的概念是基于 MySQL 这样的正向索引而言的。

那么我们先讲何为正向索引。例如给下表（tb_goods）中的 id 创建索引

如果是根据 id 查询，那么直接走索引，查询速度非常快。

但如果是基于 title 做模糊查询，只能是逐行扫描数据，流程如下：

1. 用户搜索数据，条件是 title 符合 "%手机%"
2. 逐行获取数据，比如 id 为 1 的数据
3. 判断数据中的 title 是否符合用户搜索条件
4. 如果符合则放入结果集，不符合则丢弃。然后回到步骤 1

逐行扫描，也就是全表扫描，随着数据量增加，其查询效率也会越来越低。当数据量达到数百万时，就是。。。

而倒排索引中有两个非常重要的概念：

• 文档（ Document ）：用来搜索的数据，其中的每一条数据就是一个文档。例如一个网页、一个商品信息
• 词条（ Term ）：对文档数据或用户搜索数据，利用某种算法分词，得到的具备含义的词语就是词条。例如：我是中国人，就可以分为：我、是、中国人、中国、国人这样的几个词条

创建倒排索引是对正向索引的一种特殊处理，流程如下：

• 将每一个文档的数据利用算法分词，得到一个个词条
• 创建表，每行数据包括词条、词条所在文档 id、位置等信息
• 因为词条唯一性，可以给词条创建索引，例如 hash 表结构索引

如图：

倒排索引的搜索流程如下（以搜索 "华为手机" 为例）

1. 用户输入条件 "华为手机" 进行搜索
2. 对用户输入内容分词，得到词条： 华为 、 手机
3. 拿着词条在倒排索引中查找，可以得到包含词条的文档 id 有 1、2、3
4. 拿着文档 id 到正向索引中查找具体文档

虽然要先查询倒排索引，再查询正向索引，但是词条和文档 id 都建立了索引，查询速度非常快！无需全表扫描。

为什么一个叫做正向索引，一个叫做倒排索引呢？

正向索引是最传统的，根据 id 索引的方式。但根据词条查询时，必须先逐条获取每个文档，然后判断文档中是否包含所需要的词条，是根据文档找词条的过程

倒排索引则相反，是先找到用户要搜索的词条，根据得到的文档 id 获取该文档。是根据词条找文档的过程

# 文档和字段

elasticsearch 是面向 ** 文档（Document）** 存储的，可以是数据库中的一条商品数据，一个订单信息。文档数据会被序列化为 json 格式后存储在 elasticsearch

而 JSON 文档中往往包含很多的字段（Field），类似于数据库中的列。

# 索引和映射

索引（Index），就是相同类型的文档的集合。

例如：

• 所有用户文档，就可以组织在一起，称为用户的索引；
• 所有商品的文档，可以组织在一起，称为商品的索引；
• 所有订单的文档，可以组织在一起，称为订单的索引；

因此，我们可以把索引当做是数据库中的表。

数据库的表会有约束信息，用来定义表的结构、字段的名称、类型等信息。因此，索引库中就有映射（mapping），是索引中文档的字段约束信息，类似表的结构约束。

mysql 与 elasticsearch

MySQL	Elasticsearch	说明
Table	Index	索引 (index)，就是文档的集合，类似数据库的表 (table)
Row	Document	文档（Document），就是一条条的数据，类似数据库中的行（Row），文档都是 JSON 格式
Column	Field	字段（Field），就是 JSON 文档中的字段，类似数据库中的列（Column）
Schema	Mapping	Mapping（映射）是索引中文档的约束，例如字段类型约束。类似数据库的表结构（Schema）
SQL	DSL	DSL 是 elasticsearch 提供的 JSON 风格的请求语句，用来操作 elasticsearch，实现 CRUD

• Mysql：擅长事务类型操作，可以确保数据的安全和一致性
• Elasticsearch：擅长海量数据的搜索、分析、计算

因此在企业中，往往是两者结合使用：

• 对安全性要求较高的写操作，使用 MySQL 实现
• 对查询性能要求较高的搜索需求，使用 ELasticsearch 实现
• 两者再基于某种方式，实现数据的同步，保证一致性

# 安装 Elasticsearch

因为我们还需要部署 kibana 容器，需要让 es 和 kibana 容器互联。这里先创建一个网络：

docker network create es-net

安装

docker run -d \
--name es \
-e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
-e "discovery.type=single-node" \
-v es-data:/usr/share/elasticsearch/data \
-v es-plugins:/usr/share/elasticsearch/plugins \
--privileged \
--network es-net \
-p 9200:9200 \
-p 9300:9300 \
elasticsearch:7.12.1

命令解释：

• -e "cluster.name=es-docker-cluster" ：设置集群名称
• -e "http.host=0.0.0.0" ：监听的地址，可以外网访问
• -e "ES_JAVA_OPTS=-Xms512m -Xmx512m" ：内存大小
• -e "discovery.type=single-node" ：非集群模式
• -v es-data:/usr/share/elasticsearch/data ：挂载逻辑卷，绑定 es 的数据目录
• -v es-logs:/usr/share/elasticsearch/logs ：挂载逻辑卷，绑定 es 的日志目录
• -v es-plugins:/usr/share/elasticsearch/plugins ：挂载逻辑卷，绑定 es 的插件目录
• --privileged ：授予逻辑卷访问权
• --network es-net ：加入一个名为 es-net 的网络中
• -p 9200:9200 ：端口映射配置

访问地址：http://192.168.211.128:9200 即可看到 elasticsearch 的响应结果

# 安装 kibana

kibana 可以给我们提供一个 elasticsearch 的可视化界面，便于我们学习命令。

docker run -d \
--name kibana \
-e ELASTICSEARCH_HOSTS=http://es:9200 \
--network=es-net \
-p 5601:5601  \
kibana:7.12.1

• --network es-net ：加入一个名为 es-net 的网络中，与 elasticsearch 在同一个网络中
• -e ELASTICSEARCH_HOSTS=http://es:9200" ：设置 elasticsearch 的地址，因为 kibana 已经与 elasticsearch 在一个网络，因此可以用容器名直接访问 elasticsearch
• -p 5601:5601 ：端口映射配置

访问地址：http://192.168.211.128:5601，即可看到结果

控制面板：http://192.168.211.128:5601/app/dev_tools#/console

# 安装 IK 分词器

由于国内访问 GitHub 较慢，我们选择离线模式安装。

安装插件需要知道 elasticsearch 的 plugins 目录位置，而我们用了数据卷挂载，因此需要查看 elasticsearch 的数据卷目录，通过下面命令查看

docker volume inspect es-plugins

显示结果：

[
    {
        "CreatedAt": "2022-05-06T10:06:34+08:00",
        "Driver": "local",
        "Labels": null,
        "Mountpoint": "/var/lib/docker/volumes/es-plugins/_data",
        "Name": "es-plugins",
        "Options": null,
        "Scope": "local"
    }
]

说明 plugins 目录被挂载到了 /var/lib/docker/volumes/es-plugins/_data 这个目录中

重启容器

# 4、重启容器
docker restart es
 # 查看es日志
docker logs -f es

IK 分词器包含两种模式：

• ik_smart ：智能切分，粗粒度
• ik_max_word ：最细切分，细粒度

我们在上面的 Kibana 控制台测试

GET /_analyze
{
  "analyzer": "ik_max_word",
  "text": "钟老师你好菜啊"
}

# 扩展词词典

在上面的 IK 分词器我们可以随着热点词来扩展，可以自己添加，比如 ” 钟老师应该是一个热点词 “，另外你也可以配置一些停用掉的敏感词，让其不进行分词。

打开 IK 分词器 config 目录是 IKAnalyzer.cfg.xml ，添加一个文件名，我们以 ext.dic 文件名为例。

我们去创建 ext.dic ，在其中添加热点词就好了，一个词一行。

重启 elasticsearch

docker restart es

重新测试

GET /_analyze
{
  "analyzer": "ik_max_word",
  "text": "钟老师你好菜啊"
}

# 索引库操作

# Mapping 属性映射

索引库就类似数据库表，mapping 映射就类似表的结构

我们要向 es 中存储数据，必须先创建 “库” 和 “表”

mapping 是对索引库中文档的约束，常见的 mapping 属性包括：

• type：字段数据类型，常见的简单类型有：

  • 字符串：text（可分词的文本）、keyword（精确值，例如：品牌、国家、ip 地址）
  • 数值：long、integer、short、byte、double、float、
  • 布尔：boolean
  • 日期：date
  • 对象：object

• index：是否创建索引，默认为 true

• analyzer：使用哪种分词器

• properties：该字段的子字段

我们以需要存储下面的 JSON 为例来讲解

{
    "age": 21,
    "weight": 52.1,
    "isMarried": false,
    "info": "钟老师真菜",
    "email": "jialna@qq.com",
    "score": [99.1, 99.5, 98.9],
    "name": {
        "firstName": "湖",
        "lastName": "心"
    }
}

首先对应的每个字段映射（mapping）情况如下：

• age：类型为 integer；参与搜索，index 为 true；无需分词器

• weight：类型为 float；参与搜索，index 为 true；无需分词器

• isMarried：类型为 boolean；参与搜索，index 为 true；无需分词器

• info：类型为字符串，需要分词，因此是 text；参与搜索，index 为 true；分词器可以用 ik_smart

• email：类型为字符串，但是不需要分词，因此是 keyword；不参与搜索，index 为 false；无需分词器

• score：虽然是数组，但是我们只看元素的类型，类型为 float；参与搜索，index 为 true；无需分词器

• name：类型为 object，需要定义多个子属性

  • name.firstName：类型为字符串，不需要分词，keyword；参与搜索，index 为 true；无需分词器
  • name.lastName：类型为字符串，不需要分词，keyword；参与搜索，index 为 true；无需分词器

# 创建索引库和映射

上面我们了解了 Mapping 属性映射，接下来我们就去看看如何创建索引库及映射。

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

PUT /xn2001
{
  "mappings": {
    "properties": {
      "info":{
        "type": "text",
        "analyzer": "ik_smart"
      },
      "email":{
        "type": "keyword",
        "index": "false"
      },
      "name":{
        "properties": {
          "firstName": {
            "type": "keyword"
          },
          "lastName": {
            "type": "keyword"
          }
        }
      }
    }
  }
}

我们用真实的数据库表来创建一个索引库

• 字段名、字段数据类型，可以参考数据表结构的名称和类型
• 是否参与搜索要分析业务来判断，例如图片地址，就无需参与搜索
• 是否分词呢要看内容，内容如果是一个整体就无需分词
• 分词器，我们可以统一使用 ik_max_word

PUT /hotel
{
  "mappings": {
    "properties": {
      "id": {
        "type": "keyword"
      },
      "name":{
        "type": "text",
        "analyzer": "ik_max_word",
        "copy_to": "all"
      },
      "address":{
        "type": "keyword",
        "index": false
      },
      "price":{
        "type": "integer"
      },
      "score":{
        "type": "integer"
      },
      "brand":{
        "type": "keyword",
        "copy_to": "all"
      },
      "city":{
        "type": "keyword",
        "copy_to": "all"
      },
      "starName":{
        "type": "keyword"
      },
      "business":{
        "type": "keyword"
      },
      "location":{
        "type": "geo_point"
      },
      "pic":{
        "type": "keyword",
        "index": false
      },
      "all":{
        "type": "text",
        "analyzer": "ik_max_word"
      }
    }
  }
}

特殊字段说明：

• location：地理坐标，里面包含精度、纬度
• all：一个组合字段，其目的是将多字段的值利用 copy_to 合并，提供给用户搜索，这样一来就只需要搜索一个字段就可以得到结果，性能更好。

ES 中支持两种地理坐标数据类型：

• geo_point：由纬度（latitude）和经度（longitude）确定的一个点。例如：“32.8752345, 120.2981576”
• geo_shape：有多个 geo_point 组成的复杂几何图形。例如一条直线，“LINESTRING (-77.03653 38.897676, -77.009051 38.889939)”

# 修改索引库

倒排索引结构虽然不复杂，但是一旦数据结构改变（比如改变了分词器），就需要重新创建倒排索引，这简直是灾难。因此索引库一旦创建，无法修改 mapping

虽然无法修改 mapping 中已有的字段，但是却允许添加新的字段到 mapping 中，不会对倒排索引产生影响。

PUT /索引库名/_mapping
{
  "properties": {
    "新字段名":{
      "type": "integer"
    }
  }
}

# 删除索引库

DELETE /索引库名

# 查询索引库

GET /数据库名

# DSL 文档操作

# 新增文档

POST /索引库名/_doc/文档id
{
    "字段1": "值1",
    "字段2": "值2",
    "字段3": {
        "子属性1": "值3",
        "子属性2": "值4"
    }
    // ...
}

POST /xn2001/_doc/1
{
    "info": "我不会Java",
    "email": "jialna@qq.com",
    "name": {
        "firstName": "钟",
        "lastName": "弟弟"
    }
}

# 修改文档

修改文档有两种方式：

• 全量修改：直接覆盖原来的文档
• 增量修改：修改文档中的部分字段

全量修改是覆盖原来的文档，其本质是：

• 根据指定的 id 删除文档
• 新增一个相同 id 的文档

注意：如果根据 id 删除时，id 不存在，第二步的新增也会执行，也就是变成了新增操作

PUT /{索引库名}/_doc/id
{
    "字段1": "值1",
    "字段2": "值2",
    // ... 略
}

PUT /xn2001/_doc/1
{
    "info": "我也不会敲代码",
    "email": "3300123589@qq.com",
    "name": {
        "firstName": "弟弟",
        "lastName": "钟"
    }
}

增量修改是只修改指定 id 匹配的文档中的部分字段

POST /{索引库名}/_update/文档id
{
    "doc": {
         "字段名": "新的值",
    }
}

POST /heima/_update/1
{
  "doc": {
    "email": "update@qq.com"
  }
}

# 查询文档

GET /{索引库名称}/_doc/{id}

# 删除文档

DELETE /{索引库名}/_doc/{id}

# RestClient 文档操作

ES 官方提供了各种不同语言的客户端，用来操作 ES。这些客户端的本质就是组装 DSL 语句，通过 http 请求发送给 ES。官方文档地址：https://www.elastic.co/guide/en/elasticsearch/client/index.html

其中的 Java Rest Client 又包括两种：

• Java Low Level Rest Client
• Java High Level Rest Client

我们下面学习的是 Java HighLevel Rest Client 客户端 API

# 初始化 RestClient

在 elasticsearch 提供的 API 中，elasticsearch 一切交互都封装在一个名为 RestHighLevelClient 的类中，必须先完成这个对象的初始化，建立与 elasticsearch 的连接。

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
</dependency>

SpringBoot 默认的 ES 版本是 7.6.2，我们需要覆盖默认的 ES 版本

<properties>
    <java.version>1.8</java.version>
    <elasticsearch.version>7.12.1</elasticsearch.version>
</properties>

初始化 RestHighLevelClient，初始化的代码如下：

RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(
        HttpHost.create("http://192.168.211.128:9200")
));

我们创建一个测试类 HotelIndexTest，然后将初始化的代码编写在 @BeforeEach 方法

/**
 * @author 乐心湖
 * @version 1.0
 * @date 2021/9/19 17:18
 */
public class HotelIndexTest {

    private RestHighLevelClient restHighLevelClient;

    @Test
    void testInit(){
        System.out.println(this.restHighLevelClient);
    }

    @BeforeEach
    void init(){
        this.restHighLevelClient = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://192.168.211.128:9200")
        ));
    }

    @AfterEach
    void down() throws IOException {
        this.restHighLevelClient.close();
    }
}

# 创建索引库

@Test
void createHotelIndex() throws IOException {
    //指定索引库名
    CreateIndexRequest hotel = new CreateIndexRequest("hotel");
    //写入JSON数据，这里是Mapping映射
    hotel.source(HotelConstants.MAPPING_TEMPLATE, XContentType.JSON);
    //创建索引库
    restHighLevelClient.indices().create(hotel, RequestOptions.DEFAULT);
}

public class HotelConstants {
    public static String MAPPING_TEMPLATE = "{\n" +
            "  \"mappings\": {\n" +
            "    \"properties\": {\n" +
            "      \"id\": {\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"name\":{\n" +
            "        \"type\": \"text\",\n" +
            "        \"analyzer\": \"ik_max_word\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"address\":{\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"index\": false\n" +
            "      },\n" +
            "      \"price\":{\n" +
            "        \"type\": \"integer\"\n" +
            "      },\n" +
            "      \"score\":{\n" +
            "        \"type\": \"integer\"\n" +
            "      },\n" +
            "      \"brand\":{\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"city\":{\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"copy_to\": \"all\"\n" +
            "      },\n" +
            "      \"starName\":{\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"business\":{\n" +
            "        \"type\": \"keyword\"\n" +
            "      },\n" +
            "      \"location\":{\n" +
            "        \"type\": \"geo_point\"\n" +
            "      },\n" +
            "      \"pic\":{\n" +
            "        \"type\": \"keyword\",\n" +
            "        \"index\": false\n" +
            "      },\n" +
            "      \"all\":{\n" +
            "        \"type\": \"text\",\n" +
            "        \"analyzer\": \"ik_max_word\"\n" +
            "      }\n" +
            "    }\n" +
            "  }\n" +
            "}";
}

# 删除索引库

@Test
void deleteHotelIndex() throws IOException {
    DeleteIndexRequest hotel = new DeleteIndexRequest("hotel");
    restHighLevelClient.indices().delete(hotel,RequestOptions.DEFAULT);
}

# 判断索引库

@Test
void existHotelIndex() throws IOException {
    GetIndexRequest hotel = new GetIndexRequest("hotel");
    boolean exists = restHighLevelClient.indices().exists(hotel, RequestOptions.DEFAULT);
    System.out.println(exists);
}

# 新增文档

/**
 * @author 乐心湖
 * @version 1.0
 * @date 2021/9/19 17:18
 */
@SpringBootTest
public class HotelDocumentTest {

    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private IHotelService hotelService;

    @Test
    void testInit(){
        System.out.println(this.restHighLevelClient);
    }

    @Test
    void createHotelIndex() throws IOException {
        Hotel hotel = hotelService.getById(61083L);
        HotelDoc hotelDoc = new HotelDoc(hotel);
        // 1.准备Request对象
        IndexRequest hotelIndex = new IndexRequest("hotel").id(hotelDoc.getId().toString());
        // 2.准备Json文档
        hotelIndex.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
        // 3.发送请求
        restHighLevelClient.index(hotelIndex, RequestOptions.DEFAULT);
    }

    @BeforeEach
    void init(){
        this.restHighLevelClient = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://192.168.211.128:9200")
        ));
    }

    @AfterEach
    void down() throws IOException {
        this.restHighLevelClient.close();
    }
}

# 查询文档

@Test
void testGetDocumentById() throws IOException {
    // 1.准备Request
    GetRequest hotel = new GetRequest("hotel", "61083");
    // 2.发送请求，得到响应
    GetResponse hotelResponse = restHighLevelClient.get(hotel, RequestOptions.DEFAULT);
    // 3.解析响应结果
    String hotelDocSourceAsString = hotelResponse.getSourceAsString();
    // 4.json转实体类
    HotelDoc hotelDoc = JSON.parseObject(hotelDocSourceAsString, HotelDoc.class);
    System.out.println(hotelDoc);
}

# 删除文档

@Test
void testDeleteDocumentById() throws IOException {
    DeleteRequest hotel = new DeleteRequest("hotel", "61083");
    restHighLevelClient.delete(hotel,RequestOptions.DEFAULT);
}

# 修改文档

前面我们说过，修改文档有两种方式：

• 全量修改：直接覆盖原来的文档
• 增量修改：修改文档中的部分字段

在 RestClient 的 API 中，全量修改与新增的 API 完全一致，判断依据是 ID

• 如果新增时，ID 已经存在，则修改
• 如果新增时，ID 不存在，则新增

所以全量修改写法与新增文档一样，下面我们主要是介绍增量修改。

@Test
void testUpdateDocument() throws IOException {
    // 1.准备Request
    UpdateRequest request = new UpdateRequest("hotel", "61083");
    // 2.准备请求参数
    request.doc(
        "price", "952",
        "starName", "四钻"
    );
    // 3.发送请求
    restHighLevelClient.update(request, RequestOptions.DEFAULT);
}

# 批量导入文档

案例需求：利用 BulkRequest 批量将数据库数据导入到索引库中。

• 利用 mybatis-plus 查询酒店数据
• 将查询到的酒店数据（Hotel）转换为文档类型数据（HotelDoc）
• 利用 JavaRestClient 中的 BulkRequest 批处理，实现批量新增文档

批量处理 BulkRequest，其本质就是将多个普通的 CRUD 请求组合在一起发送。

因此 Bulk 中添加了多个 IndexRequest，就是批量新增功能了。示例：

利用这一点，我们可以写出自己需要的代码，如下

@Test
void testBulk() throws IOException {
    BulkRequest bulkRequest = new BulkRequest();
    List<Hotel> hotelList = hotelService.list();
    hotelList.forEach(item -> {
        HotelDoc hotelDoc = new HotelDoc(item);
        bulkRequest.add(new IndexRequest("hotel")
                .id(hotelDoc.getId().toString())
                .source(JSON.toJSONString(hotelDoc), XContentType.JSON));
    });
    restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
}

---

总之，在 Java 代码中，client 针对操作索引库还是文档，基本都是一样的代码

restHighLevelClient.indices ().xxx，代表操作索引库

restHighLevelClient.xxx，代表操作文档

而其中所需要的参数，我们直接通过 ctrl+p 这样的快捷键去查看就可以，不需要单独记住。

# DSL 文档查询

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 /indexName/_search
{
  "query": {
    "match_all": {
    }
  }
}

# 全文检索

使用场景：全文检索查询的基本流程如下：

• 对用户搜索的内容做分词，得到词条
• 根据词条去倒排索引库中匹配，得到文档 id
• 根据文档 id 找到文档，返回给用户

比较常用的场景包括：

• 商城的输入框搜索
• 百度输入框搜索

例如京东：

因为是拿着词条去匹配，因此参与搜索的字段也必须是可分词的 text 类型的字段。

常见的全文检索查询包括：

• match 查询：单字段查询
• multi_match 查询：多字段查询，任意一个字段符合条件就算符合查询条件

match 查询语法如下：

GET /indexName/_search
{
  "query": {
    "match": {
      "FIELD": "TEXT"
    }
  }
}

mulit_match 查询语法如下：

GET /indexName/_search
{
  "query": {
    "multi_match": {
      "query": "TEXT",
      "fields": ["FIELD1", " FIELD12"]
    }
  }
}

因为我们将 brand、name、business 值都利用 copy_to 复制到了 all 字段中，你根据三个字段搜索，和根据 all 字段搜索效果是一样的。

GET /hotel/_search
{
  "query": {
    "match": {
      "all": "7天酒店"
    }
  }
}

GET /hotel/_search
{
  "query": {
    "multi_match": {
      "query": "7天酒店",
      "fields": ["brand","name"]
    }
  }
}

搜索字段越多，对查询性能影响越大，因此建议采用 copy_to 将多个字段合并为一个，然后使用单字段查询的方式。

# 精准查询

精确查询一般是查找 keyword、数值、日期、boolean 等类型字段。所以不会对搜索条件分词。

• term：根据词条精确值查询
• range：根据值的范围查询

# term 查询

因为精确查询的字段搜是不分词的字段，因此查询的条件也必须是不分词的词条。查询时，用户输入的内容跟自动值完全匹配时才认为符合条件。如果用户输入的内容过多，反而搜索不到数据。

语法说明：

// term查询
GET /indexName/_search
{
  "query": {
    "term": {
      "FIELD": {
        "value": "VALUE"
      }
    }
  }
}

示例：

GET /hotel/_search
{
  "query": {
    "term": {
      "brand": {
        "value": "7天酒店"
      }
    }
  }
}

# range 查询

范围查询，一般应用在对数值类型做范围过滤的时候。比如做价格范围过滤。

基本语法：

// range查询
GET /indexName/_search
{
  "query": {
    "range": {
      "FIELD": {
        "gte": 10, // 这里的gte代表大于等于，gt则代表大于
        "lte": 20 // lte代表小于等于，lt则代表小于
      }
    }
  }
}

示例：

精确查询常见的有哪些？

• term 查询：根据词条精确匹配，一般搜索 keyword 类型、数值类型、布尔类型、日期类型字段
• range 查询：根据数值范围查询，可以是数值、日期的范围

# 地理坐标查询

地理坐标查询，其实就是根据经纬度查询，官方文档：https://www.elastic.co/guide/en/elasticsearch/reference/current/geo-queries.html

常见的使用场景包括：

• 携程：搜索我附近的酒店
• 滴滴：搜索我附近的出租车
• 微信：搜索我附近的人

附近的酒店：

附近的车：

矩形范围查询

矩形范围查询，也就是 geo_bounding_box 查询，查询坐标落在某个矩形范围的所有文档

查询时，需要指定矩形的左上、右下两个点的坐标，然后画出一个矩形，落在该矩形内的都是符合条件的点。

// 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" // 圆心
    }
  }
}

我们先搜索陆家嘴附近 15km 的酒店

发现共有 47 家酒店，然后把半径缩短到 3 公里

可以发现，搜索到的酒店数量减少到了 5 家。

GET /hotel/_search
{
  "query": {
    "match_all": {}
  },
  "sort": [
    {
      "_geo_distance" : {
          "location": "31.034661,121.612282", //圆心
          "order" : "asc", //排序
          "unit" : "km" //单位
      }
    }
  ]
}

结果为：

"hits" : [
    {
        "_index" : "hotel",
        "_type" : "_doc",
        "_id" : "2056298828",
        "_score" : null,
        "_source" : {
            ...
        },
        "sort" : [
            4.8541199685347785 //这里的结果为离圆心的距离
        ]
    },

注意：输出结果中的 sort 为距离，比较常用。

排序完成后，页面还要获取我附近每个酒店的具体距离值，这个值在响应结果中是独立的：

# 复合查询

复合（compound）查询：复合查询可以将其它简单查询组合起来，实现更复杂的搜索逻辑。

• fuction score：算分函数查询，可以控制文档相关性算分，控制文档排名
• bool query：布尔查询，利用逻辑关系组合多个其它的查询，实现复杂搜索

# 相关性算分

这部分内容作为了解即可。

当我们利用 match 查询时，文档结果会根据与搜索词条的关联度打分（_score），返回结果时按照分值降序排列。例如，我们搜索 “虹桥如家”，结果如下：

[
  {
    "_score" : 17.850193,
    "_source" : {
      "name" : "虹桥如家酒店真不错",
    }
  },
  {
    "_score" : 12.259849,
    "_source" : {
      "name" : "外滩如家酒店真不错",
    }
  },
  {
    "_score" : 11.91091,
    "_source" : {
      "name" : "迪士尼如家酒店真不错",
    }
  }
]

elasticsearch 早期使用的打分算法是 TF-IDF 算法，公式如下：

在后来的 5.1 版本升级中，elasticsearch 将算法改进为 BM25 算法，公式如下：

TF-IDF 算法有一各缺陷，就是词条频率越高，文档得分也会越高，单个词条对文档影响较大。而 BM25 则会让单个词条的算分有一个上限，曲线更加平滑：

# 算分函数查询

根据相关度打分是比较合理的需求，但有时候也不能够满足我们的需求。

以百度为例，你搜索的结果中，并不是相关度越高排名越靠前，而是谁给的钱多排名就越靠前。

要想认为控制相关性算分，就需要利用 elasticsearch 中的 function score 查询了。

function score 查询中包含四部分内容：

• 原始查询条件：query 部分，基于这个条件搜索文档，并且基于 BM25 算法给文档打分，原始算分（query score)

• 过滤条件：filter 部分，符合该条件的文档才会重新算分

• 算分函数：符合 filter 条件的文档要根据这个函数做运算，得到的函数算分（function score），有四种函数

  • weight：函数结果是常量
  • field_value_factor：以文档中的某个字段值作为函数结果
  • random_score：以随机数作为函数结果
  • script_score：自定义算分函数算法

• 运算模式：算分函数的结果、原始查询的相关性算分，两者之间的运算方式，包括：

  • multiply：相乘
  • replace：用 function score 替换 query score
  • sum、avg、max、min

function score 的运行流程如下：

1. 根据原始条件查询搜索文档，并且计算相关性算分，称为原始算分（query score）
2. 根据过滤条件，过滤文档
3. 符合过滤条件的文档，基于算分函数运算，得到函数算分（function score）
4. 将原始算分（query score）和函数算分（function score）基于运算模式做运算，得到最终结果，作为相关性算分。

因此，其中的关键点是

• 过滤条件：决定哪些文档的算分被修改
• 算分函数：决定函数算分的算法
• 运算模式：决定最终算分结果

例如：我们给 “如家” 这个品牌的酒店排名靠前一些

GET /hotel/_search
{
  "query": {
    "function_score": {
      "query": {  .... }, // 原始查询，可以是任意条件
      "functions": [ // 算分函数
        {
          "filter": { // 满足的条件，品牌必须是如家
            "term": {
              "brand": "如家"
            }
          },
          "weight": 10 // 算分权重为10
        }
      ],
      "boost_mode": "sum" // 加权模式，求和
    }
  }
}

测试，在未添加算分函数时，如家得分如下

添加了算分函数后，如家得分就提升了

布尔查询是一个或多个查询子句的组合，每一个子句就是一个子查询。子查询的组合方式有

• must：必须匹配每个子查询，类似 “与”
• should：选择性匹配子查询，类似 “或”
• must_not：必须不匹配，不参与算分，类似 “非”
• filter：必须匹配，不参与算分

比如在搜索酒店时，除了关键字搜索外，我们还可能根据品牌、价格、城市等字段做过滤

每一个不同的字段，其查询的条件、方式都不一样，必须是多个不同的查询，而要组合这些查询，就必须用 bool 查询了。

需要注意的是，搜索时，参与打分的字段越多，查询的性能也越差。因此这种多条件查询时，建议这样做：

• 搜索框的关键字搜索，是全文检索查询，使用 must 查询，参与算分
• 其它过滤条件，采用 filter 查询，不参与算分

GET /hotel/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"city": "上海" }}
      ],
      "should": [
        {"term": {"brand": "皇冠假日" }},
        {"term": {"brand": "华美达" }}
      ],
      "must_not": [
        { "range": { "price": { "lte": 500 } }}
      ],
      "filter": [
        { "range": {"score": { "gte": 45 } }}
      ]
    }
  }
}

需求：搜索名字包含 “如家”，价格不高于 400，在坐标 31.21,121.5 周围 10km 范围内的酒店。

• 名称搜索，属于全文检索查询，应该参与算分，放到 must 中
• 价格不高于 400，用 range 查询，属于过滤条件，不参与算分，放到 must_not 中
• 周围 10km 范围内，用 geo_distance 查询，属于过滤条件，不参与算分，放到 filter 中

bool 查询的几种逻辑关系

• must：必须匹配的条件，可以理解为 “与”
• should：选择性匹配的条件，可以理解为 “或”
• must_not：必须不匹配的条件，不参与打分
• filter：必须匹配的条件，不参与打分

# 搜索结果处理

# 排序

elasticsearch 默认是根据相关度算分（_score）来排序，但是也支持自定义方式对搜索结果排序。可以排序字段类型有：keyword 类型、数值类型、地理坐标类型、日期类型等

keyword、数值、日期类型排序的语法基本一致。

GET /indexName/_search
{
  "query": {
    "match_all": {}
  },
  "sort": [
    {
      "FIELD": "desc"  // 排序字段、排序方式ASC、DESC
    }
  ]
}

排序条件是一个数组，也就是可以写多个排序条件。按照声明的顺序，当第一个条件相等时，再按照第二个条件排序。

需求描述：酒店数据按照用户评价（score) 降序排序，评价相同的按照价格 (price) 升序排序

地理坐标排序略有不同

GET /indexName/_search
{
  "query": {
    "match_all": {}
  },
  "sort": [
    {
      "_geo_distance" : {
          "FIELD" : "纬度，经度", // 文档中geo_point类型的字段名、目标坐标点
          "order" : "asc", // 排序方式
          "unit" : "km" // 排序的距离单位
      }
    }
  ]
}

GET /hotel/_search
{
  "query": {
    "match_all": {}
  },
  "sort": [
    {
      "_geo_distance" : {
          "location": "31.034661,121.612282", 
          "order" : "asc", 
          "unit" : "km" 
      }
    }
  ]
}

获取你的位置的经纬度的方式：https://lbs.amap.com/demo/jsapi-v2/example/map/click-to-get-lnglat

假设我的位置是：31.034661，121.612282，寻找我周围距离最近的酒店。

# 分页

elasticsearch 默认情况下只返回 top10 的数据。而如果要查询更多数据就需要修改分页参数了。

elasticsearch 通过修改 from、size 参数来控制要返回的分页结果：

• from：从第几个文档开始
• size：总共查询几个文档

类似于 mysql 中的 limit ?, ?

GET /hotel/_search
{
  "query": {
    "match_all": {}
  },
  "from": 0, // 分页开始的位置，默认为0
  "size": 10, // 期望获取的文档总数
  "sort": [
    {"price": "asc"}
  ]
}

深度分页问题

现在，我要查询 990~1000 的数据，查询逻辑要这么写

GET /hotel/_search
{
  "query": {
    "match_all": {}
  },
  "from": 990, // 分页开始的位置，默认为0
  "size": 10, // 期望获取的文档总数
  "sort": [
    {"price": "asc"}
  ]
}

这里是查询 990 开始的数据，也就是 第 990~ 第 1000 条 数据。

注意：elasticsearch 内部分页时，必须先查询 0~1000 条，然后截取其中的 990 ~ 1000 的这 10 条

查询 TOP1000，如果 es 是单点模式，这并无太大影响。

但是 elasticsearch 将来一定是集群，例如我集群有 5 个节点，我要查询 TOP1000 的数据，并不是每个节点查询 200 条就可以了。节点 A 的 TOP200，在另一个节点可能排到 10000 名以外了。

因此要想获取整个集群的 TOP1000，必须先查询出每个节点的 TOP1000，汇总结果后，重新排名，重新截取 TOP1000。

当查询分页深度较大时，汇总数据过多，对内存和 CPU 会产生非常大的压力，因此 elasticsearch 会禁止 from+ size 超过 10000 的请求。

针对深度分页，ES 提供了两种解决方案，官方文档：

• search after：分页时需要排序，原理是从上一次的排序值开始，查询下一页数据。官方推荐使用的方式。
• scroll：原理将排序后的文档 id 形成快照，保存在内存。官方已经不推荐使用。

分页查询的常见实现方案以及优缺点

• from + size

  • 优点：支持随机翻页
  • 缺点：深度分页问题，默认查询上限（from + size）是 10000
  • 场景：百度、京东、谷歌、淘宝这样的随机翻页搜索

• after search

  • 优点：没有查询上限（单次查询的 size 不超过 10000）
  • 缺点：只能向后逐页查询，不支持随机翻页
  • 场景：没有随机翻页需求的搜索，例如手机向下滚动翻页

• scroll

  • 优点：没有查询上限（单次查询的 size 不超过 10000）
  • 缺点：会有额外内存消耗，并且搜索结果是非实时的
  • 场景：海量数据的获取和迁移。从 ES7.1 开始不推荐，建议用 after search 方案。

# 高亮

我们在百度，京东搜索时，关键字会变成红色，比较醒目，这叫高亮显示：

高亮显示的实现分为两步：

• 1）给文档中的所有关键字都添加一个标签，例如 <em> 标签
• 2）页面给 <em> 标签编写 CSS 样式

GET /hotel/_search
{
  "query": {
    "match": {
      "FIELD": "TEXT" // 查询条件，高亮一定要使用全文检索查询
    }
  },
  "highlight": {
    "fields": { // 指定要高亮的字段
      "FIELD": {
        "pre_tags": "<em>",  // 用来标记高亮字段的前置标签
        "post_tags": "</em>" // 用来标记高亮字段的后置标签
      }
    }
  }
}

注意：

• 高亮是对关键字高亮，因此搜索条件必须带有关键字，而不能是范围这样的查询。
• 默认情况下，高亮的字段，必须与搜索指定的字段一致，否则无法高亮
• 如果要对非搜索字段高亮，则需要添加一个属性： required_field_match=false

DSL 总体结构如下：

# RestClient 文档查询

# 发起查询请求

/**
 * @author 乐心湖
 * @version 1.0
 * @date 2021/10/16 17:05
 */
@SpringBootTest
public class HotelSearchTest {

    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private IHotelService hotelService;

    @Test
    public void match_All() throws IOException {
        SearchRequest request = new SearchRequest("hotel");
        request.source()
                .query(QueryBuilders.matchAllQuery());
        SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    }

    @BeforeEach
    void init() {
        this.restHighLevelClient = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://192.168.211.128:9200")
        ));
    }

    @AfterEach
    void down() throws IOException {
        this.restHighLevelClient.close();
    }
}

• 第一步，创建 SearchRequest 对象，指定索引库名

• 第二步，利用 request.source() 构建 DSL，DSL 中可以包含查询、分页、排序、高亮等

  • query() ：代表查询条件，利用 QueryBuilders.matchAllQuery() 构建一个 match_all 查询的 DSL

• 第三步，利用 client.search() 发送请求，得到响应

关键的 API 有两个，一个是 request.source() ，其中包含了查询、排序、分页、高亮等所有功能

另一个是 QueryBuilders ，其中包含 match、term、function_score、bool 等各种查询

# 解析查询响应

Elasticsearch 返回的结果是一个 JSON 字符串，结构包含

• hits ：命中的结果

  • total ：总条数，其中的 value 是具体的总条数值

  • max_score ：所有结果中得分最高的文档的相关性算分

  • hits ：搜索结果的文档数组，其中的每个文档都是一个 json 对象

    • _source ：文档中的原始数据，也是 json 对象

因此，我们解析响应结果，就是逐层解析 JSON 字符串，流程如下

• SearchHits ：通过 response.getHits() 获取，就是 json 中的最外层的 hits，代表命中的结果

  • SearchHits.getTotalHits().value ：获取总条数信息

  • SearchHits.getHits() ：获取 SearchHit 数组，也就是文档数组

    • SearchHit.getSourceAsString() ：获取文档结果中的 _source ，也就是原始的 json 文档数据

/**
 * @author 乐心湖
 * @version 1.0
 * @date 2021/10/16 17:05
 */
@SpringBootTest
public class HotelSearchTest {

    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private IHotelService hotelService;

    @Test
    public void match_All() throws IOException {
        SearchRequest request = new SearchRequest("hotel");
        request.source()
                .query(QueryBuilders.matchAllQuery());
        SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
        SearchHits searchHits = response.getHits();
        System.out.println("hits.getTotalHits().条数 = " + searchHits.getTotalHits().value);
        SearchHit[] hits = searchHits.getHits();
        for (SearchHit hit : hits) {
            String sourceAsString = hit.getSourceAsString();
            HotelDoc hotelDoc = JSON.parseObject(sourceAsString, HotelDoc.class);
            System.out.println(hotelDoc);
        }
    }

    @BeforeEach
    void init() {
        this.restHighLevelClient = new RestHighLevelClient(RestClient.builder(
                HttpHost.create("http://192.168.211.128:9200")
        ));
    }

    @AfterEach
    void down() throws IOException {
        this.restHighLevelClient.close();
    }
}

# match 查询

@Test
public void matchQuery() throws IOException {
    SearchRequest request = new SearchRequest("hotel");
    request.source()
            .query(QueryBuilders.matchQuery("all","如家"));
    SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    SearchHits searchHits = response.getHits();
    System.out.println("hits.getTotalHits().条数 = " + searchHits.getTotalHits().value);
    SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
        String sourceAsString = hit.getSourceAsString();
        HotelDoc hotelDoc = JSON.parseObject(sourceAsString, HotelDoc.class);
        System.out.println(hotelDoc);
    }
}

@Test
public void multiMatchQuery() throws IOException {
    SearchRequest request = new SearchRequest("hotel");
    request.source()
            .query(QueryBuilders.multiMatchQuery("如家","name","brand"));
    SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    SearchHits searchHits = response.getHits();
    System.out.println("hits.getTotalHits().条数 = " + searchHits.getTotalHits().value);
    SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
        String sourceAsString = hit.getSourceAsString();
        HotelDoc hotelDoc = JSON.parseObject(sourceAsString, HotelDoc.class);
        System.out.println(hotelDoc);
    }
}

# 精确查询

精确查询主要是两者

• term：词条精确匹配
• range：范围查询

# 布尔查询

布尔查询是用 must、must_not、filter 等方式组合其它查询，代码示例如下

@Test
void testBool() throws IOException {
    // 1.准备Request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备DSL
    request.source()
            .query(
                    QueryBuilders.boolQuery()
                            .must(QueryBuilders.termQuery("city", "上海"))
                            .filter(QueryBuilders.rangeQuery("price").lte(300))
            );
    // 3.发送请求
    SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    // 4.解析响应
    SearchHits searchHits = response.getHits();
    System.out.println("hits.getTotalHits().条数 = " + searchHits.getTotalHits().value);
    SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
        String sourceAsString = hit.getSourceAsString();
        HotelDoc hotelDoc = JSON.parseObject(sourceAsString, HotelDoc.class);
        System.out.println(hotelDoc);
    }
}

# 排序、分页

搜索结果的排序和分页是与 query 同级的参数，因此同样是使用 request.source() 来设置。

对应的 API 如下

@Test
void testPageAndSort() throws IOException {
    // 页码，每页大小
    int page = 1, size = 5;

    // 1.准备Request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备DSL
    // 2.1.query
    request.source().query(QueryBuilders.matchAllQuery());
    // 2.2.排序 sort
    request.source().sort("price", SortOrder.ASC);
    // 2.3.分页 from、size
    request.source().from((page - 1) * size).size(5);
    // 3.发送请求
    SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
    // 4.解析响应
    SearchHits searchHits = response.getHits();
    System.out.println("hits.getTotalHits().条数 = " + searchHits.getTotalHits().value);
    SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
        String sourceAsString = hit.getSourceAsString();
        HotelDoc hotelDoc = JSON.parseObject(sourceAsString, HotelDoc.class);
        System.out.println(hotelDoc);
    }
}

# 高亮

• 查询的 DSL：其中除了查询条件，还需要添加高亮条件，同样是与 query 同级。
• 结果解析：结果除了要解析 _source 文档数据，还要解析高亮结果

高亮请求的构建 API

上述代码省略了查询条件部分，但是高亮查询必须使用全文检索查询，并且要有搜索关键字，将来才可以对关键字高亮.

@Test
void testHighlight() throws IOException {
    // 1.准备Request
    SearchRequest request = new SearchRequest("hotel");
    // 2.准备DSL
    // 2.1.query
    request.source().query(QueryBuilders.matchQuery("all", "如家"));
    // 2.2.高亮
    request.source().highlighter(new HighlightBuilder().field("name").requireFieldMatch(false));
    // 3.发送请求
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    // 4.解析响应
    handleResponse(response); //代码在下文
}

高亮结果解析

高亮的结果与查询的文档结果默认是分离的，并不在一起。

因此解析高亮的代码需要额外处理：

• 第一步：从结果中获取 source。 hit.getSourceAsString() ，这部分是非高亮结果，json 字符串，需要反序列为 HotelDoc 对象
• 第二步：获取高亮结果。 hit.getHighlightFields() ，返回值是一个 Map，key 是高亮字段名称，值是 HighlightField 对象，代表高亮值
• 第三步：从 map 中根据高亮字段名称，获取高亮字段值对象 HighlightField
• 第四步：从 HighlightField 中获取 Fragments，并且转为字符串。这部分是真正的高亮字符串
• 第五步：用高亮的结果替换 HotelDoc 中的非高亮结果

完整代码如下：

private void handleResponse(SearchResponse response) {
    // 4.解析响应
    SearchHits searchHits = response.getHits();
    // 4.1.获取总条数
    long total = searchHits.getTotalHits().value;
    System.out.println("共搜索到" + total + "条数据");
    // 4.2.文档数组
    SearchHit[] hits = searchHits.getHits();
    // 4.3.遍历
    for (SearchHit hit : hits) {
        // 获取文档source
        String json = hit.getSourceAsString();
        // 反序列化
        HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
        // 获取高亮结果
        Map<String, HighlightField> highlightFields = hit.getHighlightFields();
        if (!CollectionUtils.isEmpty(highlightFields)) {
            // 根据字段名获取高亮结果
            HighlightField highlightField = highlightFields.get("name");
            if (highlightField != null) {
                // 获取高亮值
                String name = highlightField.getFragments()[0].string();
                // 覆盖非高亮结果
                hotelDoc.setName(name);
            }
        }
        System.out.println("hotelDoc = " + hotelDoc);
    }
}

# 地理坐标查询

# 相关性得分

function_score 查询结构如下

对应的 JavaAPI 如下

# DSL 数据聚合

** 聚合（aggregations）** 可以让我们极其方便的实现对数据的统计、分析、运算。

• 什么品牌的手机最受欢迎？
• 这些手机的平均价格、最高价格、最低价格？
• 这些手机每月的销售情况如何？

在 Elasticsearch 实现这些统计功能比数据库的 sql 要方便的多，而且查询速度非常快，可以实现近实时搜索效果。

聚合常见的有三类

• ** 桶（Bucket）** 聚合：用来对文档做分组

  • TermAggregation：按照文档字段值分组，例如按照品牌值分组、按照国家分组
  • Date Histogram：按照日期阶梯分组，例如一周为一组，或者一月为一组

• ** 度量（Metric）** 聚合：用以计算一些值，比如：最大值、最小值、平均值等

  • Avg：求平均值
  • Max：求最大值
  • Min：求最小值
  • Stats：同时求 max、min、avg、sum 等

• ** 管道（pipeline）** 聚合：其它聚合的结果为基础做聚合

注意： 参加聚合的字段必须是 keyword、日期、数值、布尔类型

# Bucket 聚合语法

例如：我们要统计所有数据中的酒店品牌有几种，其实就是按照品牌对数据分组。此时可以根据酒店品牌的名称做聚合，也就是 Bucket 聚合。

GET /hotel/_search
{
  "size": 0,  // 设置size为0，结果中不包含文档，只包含聚合结果
  "aggs": { // 定义聚合
    "brandAgg": { //给聚合起个名字
      "terms": { // 聚合的类型，按照品牌值聚合，所以选择term
        "field": "brand", // 参与聚合的字段
        "size": 20 // 希望获取的聚合结果数量
      }
    }
  }
}

默认情况下，Bucket 聚合会统计 Bucket 内的文档数量，记为 _count ，并且按照 _count 降序排序。

我们可以指定 order 属性，自定义聚合的排序方式

GET /hotel/_search
{
  "size": 0, 
  "aggs": {
    "brandAgg": {
      "terms": {
        "field": "brand",
        "order": {
          "_count": "asc" // 按照_count升序排列
        },
        "size": 20
      }
    }
  }
}

默认情况下，Bucket 聚合是对索引库的所有文档做聚合，但真实场景下，用户会输入搜索条件，因此聚合必须是对搜索结果聚合。那么聚合必须添加限定条件。

我们可以限定要聚合的文档范围，只要添加 query 条件即可；

GET /hotel/_search
{
  "query": {
    "range": {
      "price": {
        "lte": 200 // 只对200元以下的文档聚合
      }
    }
  }, 
  "size": 0, 
  "aggs": {
    "brandAgg": {
      "terms": {
        "field": "brand",
        "size": 20
      }
    }
  }
}

这次，聚合得到的品牌明显变少了

# Metric 聚合语法

上面，我们对酒店按照品牌分组，形成了一个个桶。现在我们需要对桶内的酒店做运算，获取每个品牌的用户评分的 min、max、avg 等值。

这就要用到 Metric 聚合了，例如 stats 聚合：就可以获取 min、max、avg 等结果

GET /hotel/_search
{
  "size": 0, 
  "aggs": {
    "brandAgg": { 
      "terms": { 
        "field": "brand", 
        "size": 20
      },
      "aggs": { // 是brands聚合的子聚合，也就是分组后对每组分别计算
        "score_stats": { // 聚合名称
          "stats": { // 聚合类型，这里stats可以计算min、max、avg等
            "field": "score" // 聚合字段，这里是score
          }
        }
      }
    }
  }
}

这次的 score_stats 聚合是在 brandAgg 的聚合内部嵌套的子聚合。因为我们需要在每个桶分别计算。

另外，我们还可以给聚合结果做个排序，例如按照每个桶的酒店平均分做排序

# RestAPI 数据聚合

聚合条件与 query 条件同级别，因此需要使用 request.source() 来指定聚合条件

聚合的结果也与查询结果不同，API 也比较特殊。不过同样是 JSON 逐层解析

@Test
public void testAggregation() throws IOException {
    SearchRequest request = new SearchRequest("hotel");
    request.source().aggregation(AggregationBuilders.terms("brandAgg").field("brand").size(20));
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    Terms brandAgg = response.getAggregations().get("brandAgg");
    List<? extends Terms.Bucket> buckets = brandAgg.getBuckets();
    for (Terms.Bucket bucket : buckets) {
        String key = bucket.getKeyAsString();
        System.out.println("key = " + key);
    }
}

# 自动补全

当用户在搜索框输入字符时，我们应该提示出与该字符有关的搜索项，提示完整词条的功能，就是自动补全了。

# 拼音分词器

如果我们需要根据拼音字母来推断，因此要用到拼音分词功能。

要实现根据字母做补全，就必须对文档按照拼音分词。插件地址：https://github.com/medcl/elasticsearch-analysis-pinyin

使用 docker volume inspect es-plugins 查看插件目录，将下载的文件解压上传，重启 Elasticsearch

测试用法如下：

POST /_analyze
{
  "text": "如家酒店还不错",
  "analyzer": "pinyin"
}

结果：

# 自定义分词器

默认的拼音分词器会将每个汉字单独分为拼音，而我们希望的是每个词条形成一组拼音，需要对拼音分词器做个性化定制，形成自定义分词器。

elasticsearch 中分词器（analyzer）的组成包含三部分：

• character filters：在 tokenizer 之前对文本进行处理。例如删除字符、替换字符
• tokenizer：将文本按照一定的规则切割成词条（term）。例如 keyword，就是不分词；还有 ik_smart
• tokenizer filter：将 tokenizer 输出的词条做进一步处理。例如大小写转换、同义词处理、拼音处理等

文档分词时会依次由这三部分来处理文档：

声明自定义分词器的语法如下：

PUT /test
{
  "settings": {
    "analysis": {
      "analyzer": { // 自定义分词器
        "my_analyzer": {  // 分词器名称
          "tokenizer": "ik_max_word",
          "filter": "py"
        }
      },
      "filter": { // 自定义tokenizer filter
        "py": { // 过滤器名称
          "type": "pinyin", // 过滤器类型，这里是pinyin
          "keep_full_pinyin": false,
          "keep_joined_full_pinyin": true,
          "keep_original": true,
          "limit_first_letter_length": 16,
          "remove_duplicated_term": true,
          "none_chinese_pinyin_tokenize": false
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text",
        "analyzer": "my_analyzer",
        "search_analyzer": "ik_smart"
      }
    }
  }
}

测试

注意：为了避免搜索到同音字，搜索时不要使用拼音分词器

# 自动补全查询

elasticsearch 提供了 Completion Suggester 查询来实现自动补全功能。这个查询会匹配以用户输入内容开头的词条并返回；为了提高补全查询的效率，对于文档中字段的类型有一些约束

• 参与补全查询的字段必须是 completion 类型。
• 字段的内容一般是用来补全的多个词条形成的数组。

// 创建索引库
PUT test
{
  "mappings": {
    "properties": {
      "title":{
        "type": "completion"
      }
    }
  }
}

然后插入下面的数据

// 示例数据
POST test/_doc
{
  "title": ["Sony", "WH-1000XM3"]
}
POST test/_doc
{
  "title": ["SK-II", "PITERA"]
}
POST test/_doc
{
  "title": ["Nintendo", "switch"]
}

查询的 DSL 语句如下

// 自动补全查询
GET /test/_search
{
  "suggest": {
    "title_suggest": {
      "text": "s", // 关键字
      "completion": {
        "field": "title", // 补全查询的字段
        "skip_duplicates": true, // 跳过重复的
        "size": 10 // 获取前10条结果
      }
    }
  }
}

例如一个酒店的索引库完整案例

// 酒店数据索引库
PUT /hotel
{
  "settings": {
    "analysis": {
      "analyzer": {
        "text_anlyzer": {
          "tokenizer": "ik_max_word",
          "filter": "py"
        },
        "completion_analyzer": {
          "tokenizer": "keyword",
          "filter": "py"
        }
      },
      "filter": {
        "py": {
          "type": "pinyin",
          "keep_full_pinyin": false,
          "keep_joined_full_pinyin": true,
          "keep_original": true,
          "limit_first_letter_length": 16,
          "remove_duplicated_term": true,
          "none_chinese_pinyin_tokenize": false
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "id":{
        "type": "keyword"
      },
      "name":{
        "type": "text",
        "analyzer": "text_anlyzer",
        "search_analyzer": "ik_smart",
        "copy_to": "all"
      },
      "address":{
        "type": "keyword",
        "index": false
      },
      "price":{
        "type": "integer"
      },
      "score":{
        "type": "integer"
      },
      "brand":{
        "type": "keyword",
        "copy_to": "all"
      },
      "city":{
        "type": "keyword"
      },
      "starName":{
        "type": "keyword"
      },
      "business":{
        "type": "keyword",
        "copy_to": "all"
      },
      "location":{
        "type": "geo_point"
      },
      "pic":{
        "type": "keyword",
        "index": false
      },
      "all":{
        "type": "text",
        "analyzer": "text_anlyzer",
        "search_analyzer": "ik_smart"
      },
      "suggestion":{
          "type": "completion",
          "analyzer": "completion_analyzer"
      }
    }
  }
}

# JavaAPI

解析响应的代码如下

# 数据同步

elasticsearch 中的数据来自于 mysq l 数据库，因此 mysql 数据发生改变时，elasticsearch 也必须跟着改变，这个就是 elasticsearch 与 mysql 之间的数据同步

常见的数据同步方案有三种

• 同步调用
• 异步通知
• 监听 binlog

# 同步调用

方案一：同步调用

• hotel-demo 对外提供接口，用来修改 elasticsearch 中的数据
• 酒店管理服务在完成数据库操作后，直接调用 hotel-demo 提供的接口

# 异步通知

方案二：异步通知

• hotel-admin 对 mysql 数据库数据完成增、删、改后，发送 MQ 消息
• hotel-demo 监听 MQ，接收到消息后完成 elasticsearch 数据修改

# 监听 binlog

方案三：监听 binlog

• mysql 开启 binlog 功能
• mysql 完成增、删、改操作都会记录在 binlog 中
• hotel-demo 基于 canal 监听 binlog 变化，实时更新 elasticsearch 中的内容

# 优缺点

方式一：同步调用

• 优点：实现简单，粗暴
• 缺点：业务耦合度高

方式二：异步通知

• 优点：低耦合，实现难度一般
• 缺点：依赖 mq 的可靠性

方式三：监听 binlog

• 优点：完全解除服务间耦合
• 缺点：开启 binlog 增加数据库负担、实现复杂度高

# 实现方式

我们以异步通知为例，使用 MQ 消息中间件

MQ 结构如图：

引入依赖，在 hotel-admin、hotel-demo 中引入 rabbitmq 的依赖：

<!--amqp-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

声明队列交换机

public class MQConstants {
    /**
     * 交换机
     */
    public final static String HOTEL_EXCHANGE = "hotel.topic";
    /**
     * 监听新增和修改的队列
     */
    public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";
    /**
     * 监听删除的队列
     */
    public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";
    /**
     * 新增或修改的RoutingKey
     */
    public final static String HOTEL_INSERT_KEY = "hotel.insert";
    /**
     * 删除的RoutingKey
     */
    public final static String HOTEL_DELETE_KEY = "hotel.delete";
}

消息接收方

@Configuration
public class MqConfig {
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
    }

    @Bean
    public Queue insertQueue() {
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
    }

    @Bean
    public Queue deleteQueue() {
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
    }

    @Bean
    public Binding insertQueueBinding() {
        return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
    }

    @Bean
    public Binding deleteQueueBinding() {
        return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
    }
}

消息发送方

rabbitTemplate.convertAndSend(MQConstants.HOTEL_EXCHANGE, MQConstants.HOTEL_INSERT_KEY, hotel.getId());

rabbitTemplate.convertAndSend(MQConstants.HOTEL_EXCHANGE, MQConstants.HOTEL_DELETE_KEY, id);

消息接收方

@Override
public void insertById(Long id) {
    try {
        // 根据id查询酒店数据
        Hotel hotel = getById(id);
        // 转换为文档类型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
        client.index(request, RequestOptions.DEFAULT);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

@Override
public void deleteById(Long id) {
    try {
        DeleteRequest request = new DeleteRequest("hotel", id.toString());
        client.delete(request, RequestOptions.DEFAULT);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

@Component
public class HotelListener {

    @Autowired
    private HotelService hotelService;

    /**
     * 监听酒店新增或修改的业务
     *
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
    public void listenHotelInsertOrUpdate(Long id) {
        hotelService.insertById(id);
    }

    /**
     * 监听酒店删除的业务
     *
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
    public void listenHotelDelete(Long id) {
        hotelService.deleteById(id);
    }
}

---

# 微服务保护 Sentinel

# 雪崩问题和解决方案

# 雪崩问题:

微服务调用链路中的某个服务故障，引起整个链路的所有微服务都不可用

这里就是服务 A 需要调用服务 D 但是服务 D 因为某种原因故障了，那么服务 A 中依赖于服务 D 的业务会被阻塞住 (一直等待调用服务 D 的结果), 阻塞的话就不会释放 tomcat 连接，如果之后服务 A 多业务次请求服务 D 都一样都被阻塞住，直到服务 A 内部的所有连接都被占用住 ->Tomcat 资源耗尽，此时如果有服务调用服务 A 这个请求是不会进来的，所以可以认为服务 A 也故障了，so on…

也就是一个服务的故障了导致了依赖这个服务的也出现故障，微服务关系很复杂，可能之后所有的都故障了 -> 雪崩

# 解决方式 (4 个)

# 1. 超时处理

服务 A 请求服务 C, 如果等待了 1 秒还没接收到请求的结果，那么立即断掉这个连接

这种方式只是缓解了雪崩问题，因为我们设置的等待时间比如说是 1 秒，那么如果进来的请求是每秒两个，释放速度没有进入的速度快，总有一天还是会服务 A 资源还是会被耗尽，只是缓解雪崩，并没有从根本上解决问题

# 2. 舱壁模式

也就是我们给每一个服务 (每一个 Tomcat 装着的服务) 都按照业务划分线程池，给每一个业务分配一个线程__池__, 那么访问一个服务 A 的业务 1, 线程池只有 10 个线程，访问一个服务 A 的业务 2, 线程池只有 10 个线程.

因为业务 2 调用了服务 C, 服务 C 故障了，给服务 A 的业务 2 产生了阻塞，但是最用十个线程，我们服务 A 的业务 1 依旧可以调用服务 B. 避免整个 tomcat 资源耗尽，只是阻塞部分 tomcat 的线程

确实解决了雪崩问题，但是资源上讲还是有些浪费，因为服务 C 已经挂了，每次请求服务 A 的业务 2 来访问服务 C 还是会请求，但是已经挂了，还占用了那 10 个线程，浪费

# 3. 熔断降级

也就是服务 A 如果调用服务 D, 然后计算异常比例 (所有请求中有几次发生阻塞 (故障)), 如果达到了某个 (我们设置的) 值，暗恶魔就会直接熔断这个服务 D, 之后要是服务 A 想要请求服务 D, 会直接被拦截，请求直接失败，1 毫秒都不等，不占用资源 (因为压根不让访问服务 D 了)

解决雪崩比较好的一个方案

# 4. 流量控制

这种方式就需要用到我们的 Sentinel 了

如果有多个请求到我们的一个微服务，Sentinel 按照我们给这个微服务设置的 QPS (query per second) 这个频率来释放请求，我们这个微服务就可以在可以承受的频率处理请求避免发生故障，他不产生故障就不会把故障传递给依赖于这个服务的服务了，不会出现雪崩问题

这种方式相当于避免出现故障 (预防雪崩问题)

当然这不完全解决雪崩问题， 因为高并发导致服务故障可能只是其中一个，还有各种其他方式可能会导致一个服务故障 (比如说网络问题等等等), 这里只是预防了高并发引发的服务故障

# 总结

# 服务保护技术对比

# Sentinel 介绍和安装

# 微服务整合 Sentinel

端点 (endpoint)-> 我们的 springmvc 中任意一个 controller 接口都是一个端点，所以只需要访问一下触发了 sentinel 监控就能看到监控信息

访问完成后，再看我们之前开启的 sentinel 后台管理页面:

# 限流规则

# 快速入门

# 簇点链路

所以就是一般是监控我们写的 controller 层的那些方法 (controller 里面每一个方法都是被监控的资源), 如果我们的 service 和 mapper 也想被监控就需要 sentinel 提供的一些注解来实现

针对来源就是从哪访问过来的请求需要做流控 (因为当前在设置的就是流控),default-> 代表一切进来的请求都做限流

然后 QPS 之前说过，然后设为了 1, 也就是每秒最多放行让这个接口处理 1 个请求，超出的会被拦截并报错

# 案例

# 流控模式

# 关联模式

也就是如果有多个修改订单的请求 (会对数据库做出更改), 超过我们设置的 QPS 值，我们希望把查询订单的请求 (也会用到数据库但是是查询，比修改 less important) 做限流，这样高并发情况下 (多请求 (高 QPS)), 修改业务先做 (更多资源来处理这个请求了), 查询业务 less important, 等修改业务做完或者做的差不多再去做.

当优先级高的资源被请求超过设置的 QPS 给优先级低的做限流

# 链路模式

这么设置如果从 /test1 进入到 /common 请求就不会管，所以 /test2 访问 /common 过多超过了设置的 QPS, 那么来自 /test2 访问 /common 就会被限流，而从 /test1 进入到 /common 请求就不会被限流

链路理解是 controller->service->mapper 这种的，不是访问地址！

注意我们需要让 sentinel 监控 service 层的 queryGoods 方法，默认只是监控 controller 层的那些方法:

这么做之后那个 service 的方法会被监控，我们才可以做限流等操作

# 总结

# 流控效果

# warm up

# 排队等待

所有的请求都会放入队列，如果其中有一个请求超出了我们设置的 timeout, 那么就会被拒绝并报出异常 (就像是我们之前那样 -> 正常的和 warmup)

能做到流量整形的效果，波浪形的请求进来，(从这里 (这个请求队列) 出去的) 接收也会是很平缓的:

# 总结

# 热点参数限流

但是，注意！！！

所以我们需要给那些 controller 里面的那个方法加上 @SentinelResource 才可以配置这些参数限流

比如说:

# 隔离和降级

# FeignClient 整合 Sentinel

一整合，sentinel 就会监控 feign 客户端，把他当做链路中的一个资源.

这样我们就可以配置各种限流，隔离，降级级别等等等

给 feign 客户端 (就那个自己写的接口！要是有什么错了该怎么怎么办) 编写失败的降级逻辑，这样要是真的触发了我们的设置的各种 (降级的) 规则，那么就会按照我们设置的失败的降级逻辑来操作，而不是直接报错 (不友好)

可以理解为写个备用方案

# 线程隔离 (舱壁模式)

# 熔断降级

# 断路器熔断策略 - 慢调用

# 断路器熔断策略 - 异常比例和异常数

• 异常比例就是异常的比例
• 异常数就是超过给的异常数

这样给 feign client 设置了这些熔断降级的，只要现在有微服务使用另外一个微服务的 feign client 来调用那个微服务，如果调用符合了我们设定的熔断规则 (以及策略), 那么任何调用这个微服务的请求都会被熔断 (其他的微服务调用这个微服务请求也会被熔断)

# 总结

# 授权规则

sentinel 也需要这个，因为要是有人直接访问我们的微服务而不是通过网关那么网关就没效果了

所以需要给 sentiental 也配置个授权规则，判断访问当前微服务是不是通过网关来访问的

这个 origin 的请求头里面根本没有

需要约定好了，到时候再传过来这个就行了

# 自定义异常结果

# 规则持久化

# 规则管理模式

# pull 模式

因为是读取，要是别的微服务要用这个客户端设置的规则但是还没到时间从那个 Machine1 里面读取那个本地文件，就会发送错误 (其他机器上的本地规则缓存还是没变), 所以不推荐 pull 模式

# 实现 push 模式

具体的要看 https://www.bilibili.com/video/BV1LQ4y127n4?p=146

---

# 分布式事务 seata

# 分布式事务

每个服务都有着自己再 service 层的自己的事务，自己的部分做完了就提交给自己的数据库了

别的服务要是失败了这些其他服务也不知道照样提交，这样子不行，得要分布式事务

# CAP 定理

这里有两个节点存的是一样的 (主从 -> 备份), 然后如果其中一个节点中的数据变了，另外一个节点数据也需要变，这样用户访问哪一个都是一样的

凡是分布式系统，分区一定会出现，那么就代表必须有 P, 剩下要么是 A 要么是 C.

# BASE 理论

# 分布式事务模型

# Seata