微服务学习笔记-易微帮

微服务学习笔记

文章目录

微服务学习笔记

认识微服务

微服务技术栈

1、服务集群：系统功能的实现

2、注册中心：记录每个服务的IP、端口、功能等。起到服务注册和服务发现的作用。

3、配置中心：实现各个服务的配置问题

4、服务网关：用户访问的时候进行校验、路由到服务、负载均衡。所有的服务调用都调用到网关，然后在网关里配置路由，进行服务的转发，类似于代理的作用。当然网关需要配合注册中心进行使用，去发现转发到哪个服务上去。

5、分布式缓存：缓存数据

6、分布式搜索：快速搜索

7、消息队列：异步通信

8、分布式日志：统一对日志存储、统计、分析

9、系统监控链路追踪：实时追踪每个服务的负载、CPU运行状态、内存等。出问题可以快速追踪。

微服务学习要点

1、微服务治理：SpringCloud相关技术

2、异步通信技术

3、缓存技术

4、搜索技术

5、DevOps

微服务远程调用

Spring提供了 RestTemplate 用于远程调用。

1)注册RestTemplate

在order-service的OrderApplication中注册RestTemplate

@MapperScan("cn.itcast.order.mapper")
@SpringBootApplication
public class OrderApplication {
public static void main(String[] args) {
    SpringApplication.run(OrderApplication.class，args);
    @Bean//返回一个RestTemplate对象交给Spring IOC容器进行管理，方便之后进行自动注入
    public RestTemplate restTemplate(){
        return new RestTemplate0;
    }
}

2) 服务远程调用RestTemplate

修改order-service中的OrderService的queryOrderByld方法

@Service
public class OrderService {
    @Autowired
    private RestTemplate restTemplate:
    public Order queryOrderById(Long orderId) {
        // 1.查询订单
        Order order = orderMapper.findById(orderId);
        // TODO 2.查询用户
        String url ="http:// localhost:8081/user/" + order.getUserId();
        User user = restTemplate.getForObject(url,User.class);//使用restTemplate发送远程调用
        // 3.封装user信息
        order.setUser(user);
        // 4.返回
        return order;
    }
}

Eureka注册中心

简介

在Eureka架构中，微服务角色有两类

**EurekaServer:**服务端，注册中心

记录服务信息
心跳监控
**EurekaClient:**客户端
Provider:服务提供者，例如案例中的 user-service
注册自己的信息到EurekaServer
每隔30秒向EurekaServer发送心跳

consumer:服务消费者，例如案例中的order-service
根据服务名称从EurekaServer拉取服务列表
基于服务列表做负载均衡，选中一个微服务后发起远程调用

消费者该如何获取服务提供者具体信息?

服务提供者启动时向eureka注册自己的信息
eureka保存这些信息
消费者根据服务名称向eureka拉取提供者信息

如果有多个服务提供者，消费者该如何选择?

服务消费者利用负载均衡算法，从服务列表中挑选一个消费者

如何感知服务提供者健康状态?

服务提供者会每隔30秒向EurekaServer发送心跳请求，报告健康状态eureka会更新记录服务列表信息，心跳不正常会被剔除
消费者就可以拉取到最新的信息

操作过程

搭建EurekaServer

1.创建项目，引入spring-cloud-starter-netflix-eureka-server的依赖

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

2.编写启动类，添加@EnableEurekaServer注解

3.添加application.ym[文件，编写下面的配置:

server:
	port: 10086 
spring:
	application:
		name: eurekaserver
eureka:
	client:
		service-url: # 指定eureka服务端的注册地址,可以指定多个
				defaultZone: http://127.0.0.1:10086/eureka

注册user-service

1.在user-service项目引入spring-cloud-starter-netflix-eureka-client的依赖

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

2.在application.yml文件，编写下面的配置

spring:
	application:
		name: userservice
eureka:
	client:
		service-url:
			defaultZone: http://127.0.0.1:10086/eureka

另外，我们可以将user-service多次启动，模拟多实例部署，但为了避免端口冲突，需要修改端口设置.

在Service中选中相关实例然后复制一份，并配置端口

在order-service完成服务拉取

服务拉取是基于服务名称获取服务列表，然后在对服务列表做负载均衡

1.修改OrderService的代码，修改访问的url路径，用服务名代替ip、端口:

String url = "http://userservice/user/" + order.getUserId();

2.在order-service项目的启动类OrderApplication中的RestTemplate添加负载均衡注解:

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

Ribbon负载均衡

请求将会被LoadBalanceInterceptor负载均衡拦截器拦截下来。

这个拦截器会将请求交割RibbonLoadBalancerClient（之后简称R）这个对象，它会将服务id交给DynamicServeListLoadBalancer （后面简称D）这个对象。之后D会向Eureka拉取服务列表，Eureka返回列表后D将会寻找IRule负责负载均衡。IRule将会选择某个服务返回R，R将会修改Url从而发起真实的请求。

IRule负载均衡策略

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

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

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

@Bean
public IRule randomRule(){
    return new RandomRule();
}

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

userservice:
	ribbon:
		NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule # 负载均衡规则

饥饿加载

Ribbon默认是采用懒加载，即第一次访问时才会去创建LoadBalanceClient，请求时间会很长.而饥饿加载则会在项目启动时创建，降低第一次访问的耗时，通过下面配置开启饥饿加载:

ribbon:
	eager-load:
		enabled: true # 开启饥饿加载
		cLients: 
			- userservice # 指定对userservice这个服务饥饿加载

Nacos注册中心

步骤

服务注册到Nacos

1.在cloud-demo父工程中添加spring-cloud-alilbaba的管理依赖:

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

2.注释掉order-service和user-service中原有的eureka依赖
3.添加nacos的客户端依赖:

<!-- nacos客户端依赖 -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>

4.修改user-service&order-service中的application.yml文件，注释eureka地址，添加nacos地址:

spring:
	cloud:
		nacos:
			server-addr: localhost:8848 # nacos 服务端地址

5.启动并测试:

Nacos服务搭建

下载安装后在bin下运行

startup.cmd -m standalone

分级存储模型

服务->集群->实例

服务集群属性

1.修改application.yml，添加如下内容:

spring:
	cloud:
		nacos:
			server-addr: localhost:8848 # nacos 服务端地址
			discovery:
				cluster-name: HZ # 配置集群名称，也就是机房位置，例如: HZ，杭州

2.在Nacos控制台可以看到集群变化:

根据集群负载均衡

1修改order-service中的application.yml，设置集群为HZ:

spring:
	cloud:
		nacos:
			server-addr: localhost:8848 # nacos 服务端地址
			discovery:cluster-name: HZ # 配置集群名称，也就是机房位置

2然后在order-service中设置负载均衡的IRule为NacosRule，这个规则优先会寻找与自己同集群的服务:

userservice:
	ribbon:
		NFLoadBalancerRuleclassName: com.albaba.cloudnacos.ribbon.NacosRule # 负载均衡规则

3.注意将user-service的权重都设置为1

根据权重负载均衡

实际部署中会出现这样的场景:
服务器设备性能有差异，部分实例所在机器性能较好，另一些较差，我们希望性能好的机器承担更多的用户请求
Nacos提供了权重配置来控制访问频率，权重越大则访问频率越高

1.在Nacos控制台可以设置实例的权重值，首先选中实例后面的编辑按钮

2.将权重设置为0.1，测试可以发现8081被访问到的频率大大降低

环境隔离-namespace

Nacos中服务存储和数据存储的最外层都是一个名为namespace的东西，用来做最外层隔离

namespace用来做环境隔离每个namespace都有唯一id不同namespace下的服务不可见

Namespace->Group->Service/Data

1.在Nacos控制台可以创建namespace，用来隔离不同环境

2.然后填写一个新的命名空间信息:

3.保存后会在控制台看到这个命名空间的id:

4.修改order-service的application.yml，添加namespace

spring:
	cloud:
		nacos:
			server-addr: localhost:8848
			discovery:
				cluster-name: SH # 上
				namespace: 492a7d5d-237b-46a1-a99a-fa8e98e4bof9 # 命名空间，填ID

临时实例和非临时实例

服务注册到Nacos时，可以选择注册为临时或非临时实例。

临时实例如果不健康将会直接删除对应的实例。非临时将会实时监测健康状态，等待恢复健康。通过下面的配置来设置

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

Nacos与eureka的共同点

都支持服务注册和服务拉取
都支持服务提供者心跳方式做健康检测

Nacos与Eureka的区别

Nacos支持服务端主动检测提供者状态:临时实例采用心跳模式，非临时实例采用主动检测模式。临时实例心跳不正常会被剔除，非临时实例则不会被剔除
Nacos支持服务列表变更的消息推送模式，服务列表更新更及时
Nacos集群默认采用AP方式，当集群中存在非临时实例时，采用CP模式;Eureka采用AP方式

Nacos配置管理

统一配置管理

配置更改热更新。

在Nacos客户端的配置列表进行配置

配置名称的约定一般如下：功能-运行环境。比如：userservice-dev.yaml

不是所有配置都更新过来，而是有热更新需求的配置放在其中。

配置获取步骤：

项目启动->读取bootstap.yml文件获取nacos地址。读取nacos中配置文件

微服务获取Nacos配置

1.引入Nacos的配置管理客户端依赖

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

2.在userservice中的resource目录添加一个bootstrap.yml文件，这个文件是引导文件，优先级高于
application.yml:

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

配置热更新

Nacos中的配置文件变更后，微服务无需重启就可以感知。不过需要通过下面两种配置实现

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

此时的注入需要使用@Value注解

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

方式二（推荐）:使用@ConfigurationProperties注解完成配置自动加载，注意前缀名和变量名的凭借要和配置一致。

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

此时的注入采用自动装配。

@Autowired
private PatternProperties properties;

多环境配置共享

某些配置是多个环境下一样的。

微服务启动时会从nacos读取多个配置文件:
1、[spring.application.name]-[spring.profiles.active].yaml，例如: userservice-dev.yaml

2、[spring.application.name].yaml，例如: userservice.yaml
无论profile如何变化，[spring.application.name].yaml这个文件一定会加载，因此多环境共享配置可以写入这个文件

多种配置的优先级

服务名-profile.yaml>服务名称yaml >本地配置

搭建Nacos集群

搭建集群的基本步骤:
1、搭建数据库，初始化数据库表结构
2、下载nacos安装包
3、配置nacos

进入nacos的conf目录，修改配置文件cluster.conf.example，重命名为cluster.conf:

添加节点信息，如：

127.0.0.1:8845
127.0.0.1:8846

修改application.properties文件，添加数据库配置：

# 打开数据源
spring.datasource.platform=mysql
#配置url和用户信息

修改application.properties文件，修改对应的端口号

server.port=8845

4、启动nacos集群

在bin目录下启动startup.cmd

5、nginx反向代理

在nginx的conf目录下修改nginx.conf文件,在http中添加以下内容：

upstream nacos-cluster {
    server 127.0.0.1:8845;
    server 127.0.0.1:8846;
    server 127.0.0.1:8847;
}
server {
	listen	80;
	server_name	1ocalhost;
	location /nacos {
        proxy_pass http://nacos-cluster;
    }
}

http客户端Feign

RestTemplate方式调用存在的问题：1、代码可读性差，编程体验不统一；2、参数复杂URL难以维护

Feign是一个声明式的http客户端，其作用就是帮助我们优雅的实现http请求的发送，解决上面提到的问题。

Feign使用ribbon实现了负载均衡

使用步骤

1、引入依赖

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

2、在order-service的启动类添加注解开启Feign功能：

@EnableFeignClients

3、编写Feign客户端：

//指定服务名称
@FeignClient("userservice")
public interface UserClient{
    @GetMapping("/user/{id}")
    User findById( @PathVariable("id") Long id);
}

主要是基于SpringMVC的注解来声明远程调用的信息，比如:
服务名称: userservice
请求方式:GET
请求路径: /user/{id)
请求参数: Long id
返回值类型:User

4、使用Feign进行远程调用

 public Order queryOrderById(Long orderId){
     // 1.查询订单
     Order order = orderMapper.findById(orderId);
     // 2.用Feign远程调用
     User user = userclient.findById(order.getUserId());
     // 3.封装user到Order
     order.setUser(user);
     // 4.返回
     return order;
 }

Feign的自定义配置

Feign运行自定义配置来覆盖默认配置，可以修改的配置如下：

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

配置Feign日志有两种方式:

方式一:配置文件方式
全局生效:

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

局部生效：

feign:
	client:
		config:
			userservice: 
				loggerLevel: FULL

方式二:java代码方式，需要先声明一个Bean:

public class FeignClientConfiguration{
    @Bean
    public Logger.Level feignLogLevel(){
        return Logger.Level.BASIC;
    }
}

如果在启动类上的@EnableFeignClients注解中声明则代表全局

@EnableFeignClients(defaultConfiguration = FeignClientConfiguration.class)

如果在客户端上的@FeignClient注解中声明则代表某服务

@Feignclient(value = "userservice",configuration = FeignclientConfiguration.class)

Feign的性能优化

Feign底层的客户端实现

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

因此优化Feign的性能主要包括:

使用连接池代替默认的URLConnection
日志级别，最好用basic或none

连接池配置

1、引入依赖

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

2、配置连接池，开启Feign对连接池的支持

feign:
	httpclient:
		enabled: true # 开feign对HttpClient的支持
		max-connections: 200 # 最大的连接数
		max-connections-per-route: 50 # 每个路的最大连接数

Feign的最佳实践

方式一(继承)（不推荐）:给消费者的FeignClient和提供者的controller定义统一的父接口作为标准

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

之后定义的客户端和controller接口继承该接口。

方式二(抽取):将FeignClient抽取为独立模块，并且把接口有关的POJO、默认的Feign配置都放到这个模块中，提供给所有消费者使用

1、首先创建一个module，命名为feign-api，然后引入feign的starter依赖

2、将order-service中编写的UserClient、User、DefaultFeignConfiguration都复制到feign-api项目中

3、在order-service中引入feign-api的依赖

4、修改order-service中的所有与上述三个组件有关的import部分，改成导入feign-api中的包
重启测试

5、重启测试

在使用过程中，当定义的FeignClient不在SpringBootApplication的扫描包范围时，这些FeignClient无法使用。有两种方式解决:

方式一:指定FeignClient所在包

@EnableFeignClients(basePackages = "cn.itcast.feign.clients")

方式二（推荐）:指定FeignClient字节码

@EnableFeignClients(clients = {UserClient.class})

统一网关Gateway

网关功能：

身份认证和权限校验
服务路由、负载均衡
请求限流

网关的技术实现：
在SpringCloud中网关的实现包括两种 gateway、zuul

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

搭建网关服务

1、创建新的module，引入SpringCloudGateway的依赖和nacos的服务发现依赖。

<!-- 网关依赖-->
<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>

2、编写路由配置及nacos地址

server:
	port: 10010 # 网关端口
spring:
	application:
		name: gateway # 服务名称
	cloud:
		nacos:
			server-addr: Localhost:8848 # nacos地址
		gateway:
			routes: # 网关路由配置
				- id: user-service # 路由id，自定义，只要唯一即可
				  # uri: http://127.0.0.1:8081 # 路由的目标地址 http就是固定地址
				  uri: lb://userservice # 路由的目标地址 L就是负裁均衡，后面跟服务名称
				  predicates: # 路由断言，也就是判断请求是否符合路由规则的条件
			  		- Path=/user/** # 这个是按照路径配，只要以/user/开头就符合要求
				- id: order-service
				  uri: lb://order-service
				  predicates:
				  	- Path=/order/**

路由断言工厂

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

Spring提供了11种基本的Predicate工厂:

名称	说明	示例
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/Denverj
Cookie	请求必须包含某些cookie	~ Cookie=chocolate, ch.p
Header	请求必须包含某些header	- Header=X-Request-ld, \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	权重处理

使用：

predicates:
	- Path=/order/**
	- After=2023-11-06T16:55:50.123+08:00[Asia/Shanghai]

过滤器

路由过滤器 GatewayFilter

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

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

如果要对所有的路由都生效，则可以将过滤器工厂写到default下。格式如下:

spring:
	cloud:
		gateway:
			default-filters: # 默认过滤器，会对所有的路由请求都生效"
				- AddReqvestHeader=Truth，Itcast is freaking awesome!# 加请求义

全局过滤器GlobalFilter

全局过滤器的作用也是处理一切进入网关的请求和微服务响应，与GatewayFilter的作用一样。
区别在于GatewayFilter通过配置定义，处理逻辑是固定的。而GlobalFilter的逻辑需要自己写代码实现。
定义方式是实现GlobalFilter接口。

public interface GlobalFilter {
/**
*处理当前请求，有必要的话通过{@link GatewayFilterChain}将请求交给下一个过游器处理
*
* @param exchange 请求上下文，里面可以获取Request、Response等信息
* @param chain 用来把请求委托给下一个过滤器
* @return {@code Mono<Void>} 返回标示当前过滤器业务结束
*/
	Mono<Void> filter(ServerWebExchange exchange，GatewayFilterChain chain);
}

实现接口：

需要强调的一点在于在网关中使用的不再是Servlet的编程方式，而是响应式编程。

@Order(-1)//定义优先级
@component
public class AuthorizeFilter implements GlobalFilter{
    // 1.获取请求参数
    ServerHttpRequest request = exchange.getRequest();
    MultiValueMap<String，String> params = request.getQueryParams();
    // 2获取参数中的 authorization 参数
    String auth = params.getFirst( key: "authorization");
    // 3判断参数值是否于 admin
    if ("admin".equals(auth)){
        //4 是，放行
        return chain.filter(exchange);
    }
    // 5.否，拦截
	// 5.1.设置状态码
    exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
    // 5.2.拦截请求
    return exchange.getResponse().setComplete();
}

过滤器顺序

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

GlobalFilter通过实现Ordered接口，或者添加@Order注解来指定order值，由我们自己指定

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

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

跨域问题处理

解决方案：CORS

网关处理跨域采用的同样是CORS方案，并且只需要简单配置即可实现:

spring:
	cloud:
		gateway:
			globalcors: #全局的跨域处理
				add-to-simple-url-handler-mapping: true # 解决options请求被拦截问题
				corsConfigurations:
					'[/**]': #对一切请求使用如下配置
						aLLowedOrigins: # 允许哪些网站的跨域请求
							- "http://localhost:8090"
							- "http://www.leyou.com"
						allowedMethods: # 允许的跨域ajax的请方式
							- "GET" 
							- "POST"
							- "DELETE"
							- "PUT"
							- "OPTIONS"
						aLLowedHeaders: "*" #允许在请求中携带的头信息
						alLowCredentials: true # 是否允许携带cookie
						maxAge: 360000 # 这次跨域检测的有效期

Docker

详见分章节，Docker学习的部分。

服务异步通讯

异步调用方案

常见的实践：事件驱动模式

引入Broker事件代理者：将事件交给Broker进行管理，而对应的事件则订阅Broker。事件发生后Broker将会通知订阅了他的服务。

初识MQ

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

产品

RabbitMQ快速入门

安装

docker pull rabbitmq:3-management

docker run \
-e RABBITMA_DEFAULT_USER=mx \
-e RABBITMA_DEFAULT_PASS=123456 \
--name mq \ 
--hostname mql \
-p 15672:15672 \
-p 5672:5672 \
-d \
rabbitmq:3-management

消息模型

基本消息队列
工作消息队列
发布订阅
- 广播
- 路由
- 主题

下面仅演示基本消息队列：

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

publisher: 消息发布者，将消息发送到队列queue

queue: 消息队列，负责接受并缓存消息

consumer: 订阅队列，处理队列中的消息

消费者基本过程：

1、建立链接(设置参数)

2、创建通道channel

3、利用channel创建队列

4、订阅消息，定义消费行为并将其与队列绑定

生产者基本过程：

1、建立链接(设置参数)

2、创建通道

3、利用channel创建队列

4、利用channel向队列发送消息

SpringAMQP(RabbitMQ)

消息发送

1、在父工程中引入spring-amqp的依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列

2.1、在publisher服务中编写application.yml, 添加mq连接信息：

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

2.2、编写测试类发送消息

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSimpleQueue() {
        String queueName = "simple.queue";
        String message = "hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName,message);
    }
}

3、在consumer服务中编写消费逻辑，绑定（监听）simple.queue这个队列

3.1、在consumer服务中编写application.yml，添加mq连接信息

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

3.2、在consumer服务中新建一个类，编写消费逻辑

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System,out.println("spring 消费者接收到消息 :["+ msg +"]");
    }
}

Work Queue工作队列

对于一种消息给多个消费者共同处理的情况应该使用Work Queue，若使用Basic Queue因为存在消息预取机制，消息在处理之前已经分配完毕。

可以修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限

spring:
	rabbitmq:
		host: 192.168.150.101 # 主机名
		port: 5672 # 端口
		virtual-host: / # 虚拟主机
		username: itcast # 用户名 
		password: 123321 # 密码
		listener:
			simple:
				prefetch: 1

发布(publish)订阅(Subscribe)

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange(交换机)

exchange从生产者中得到消息，之后将消息按照需求转发给多个(或者单个)不同的消息队列

注意:exchange负责消息路由，而不是存储，路由失败则消息丢失

广播Fanout Exchange

Fanout Exchange 会将接收到的消息路由到每一个跟其绑定的queue

实现步骤：

1.在consumer服务中，利用代码声明队列、交换机，并将两者绑定

@Configuration
public class FanoutConfig {
    // 声明FanoutExchange交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("MX.fanout");//交换机名称
    }
    //声明第1个队列
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");//队列名称
    }
    //绑定队列和交换机
    @Bean
    public Binding bindingQueve1(Queve fanoutQueve1，FanoutExchange fanoutExchange){
return BindingBuilder
    .bind(fanoutQueue1)
    .to(fanoutExchange);
    }
    //以相同方式声明第2个队列，并完成绑定
}

2.在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(
    String msg){
    System.out.println("消费者接收到fanout.queve1的消息:["+ msg +"]");
}

3.在publisher中编写测试方法，向交换机MX.fanout发送消息

// 交换机名称
String exchangeName = "itcast.fanout";
// 消息
String message = "hello, every one!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName，routingKey:"",message);

路由Direct Exchange

Direct Exchange 会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式 (routes)

每一个Queue都与Exchange设置一个BindingKey
发布者发送消息时，指定消息的RoutingKey
Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

值得注意的是对于一个消息队列，可以拥有多个BindingKey，只要有一个符合情况就可以成功转发

实现思路如下：

1.利用@RabbitListener声明Exchange、Queue、RoutingKey

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(
        name = "direct.queue1"),
    	exchange = @Exchange(
            name = "itcast.direct", 			
            type = ExchangeTypes.DIRECT),
    	key = {"red","blue"}
    ))
public void listenDirectqueue1(String msg){
    System.out.println("消费者接收到direct.queue1的消息:["+ msg +"]");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(
        name = "direct.queue2"),
    	exchange = @Exchange(
            name = "itcast.direct", 			type = ExchangeTypes.DIRECT),
    	key = {"red","yellow"}
    ))
public void listenDirectqueue2(String msg){
    System.out.println("消费者接收到direct.queue2的消息:["+ msg +"]");
}

2.在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2

略

3.在publisher中编写测试方法，向itcast.direct发送消息

// 交换机名称
String exchangeName = "itcast.fanout";
// 消息
String message = "hello, every one!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName,"blue",message);
//rabbitTemplate.convertAndSend(exchangeName,"red",message);

主题TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以.分割。比如china.news、china.power、US.news

Queue与Exchange指定BindingKey时可以使用通配符：

#:代指0个或多个单词

*:代指一个单词

china.#

#.news

利用@RabbitListener声明Exchange、Queue、RoutingKey

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(
        name = "topic.queue1"),
    	exchange = @Exchange(
            name = "itcast.topic", 			type = ExchangeTypes.TOPIC),
    	key = "china.#"
    ))
public void listenDirectqueue2(String msg){
    System.out.println("消费者接收到direct.queue2的消息:["+ msg +"]");
}

消息转换器

Spring的对消息对象的处理是由org.springframeworkamqp.support.converter.MessageConverter来处理的。而默认实现是simpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。如果要修改只需要定义一个MessageConverter 类型的Bean即可。

推荐用JSON方式序列化，步骤如下:

publisher引入依赖

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

在publisher服务声明MessageConverter:

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

consumer引入依赖
在consumer服务声明MessageConverter
定义一个消费者，监听object.queue队列并消费消息

@RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String， Object> msg) {
System,out.println("收到消息:["+ msg +"]");
}

个人小结

在异步通信时我们采用的产品是基于AMQP协议的由SpringBoot整合的RabbitMQ

配置上

spring:
	rabbitmq:
		host: 192.168.150.101 # 主机名
		port: 5672 # 端口
		virtual-host: / # 虚拟主机
		username: itcast # 用户名 
		password: 123321 # 密码
		listener:
			simple:
				prefetch: 1 # 最大预取消息，也就是在工作前就预先分配给消息队列的数量

发布、订阅的核心在于交换机Exchange。

广播对挂载的所有消息队列发送，路由和主题则是有选择性的，主题可以使用通配符

最核心的部分分为三个过程：中间件、发送方、接收方

中间件：定义消息队列和交换机，并将他们相互绑定。可以使用@Configuration注解定义一个类，但是实际开发时推荐利用@RabbitListener声明接收方监听时添加对应的Exchange、Queue、RoutingKey。

接收方：定义处理消息的行为，以及与消息队列挂载。实际使用时可能会定义网络拓扑。其中key决定了将会处理什么样的消息

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(
        name = "topic.queue1"),//消息队列的名称
    	exchange = @Exchange(//交换机信息
            name = "itcast.topic",//交换机名称
            type = ExchangeTypes.TOPIC),//交换机类型
    	key = "china.#" //转发的tag
    ))
public void listenDirectqueue2(String msg){
    System.out.println("消费者接收到direct.queue2的消息:["+ msg +"]");
}

发送方：指定发送给哪个网络，即交给哪个交换机，交换机根据key转发给符合条件的消息队列

//根据业务，包含以下代码
// 交换机名称
String exchangeName = "MX.topic";
// 消息
String message = "hello, every one!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName，"china.news",message);

过程：业务需求将消息给交换机所指定的网络，交换机根据key指定消息队列转发，接收方监测到消息队列内容根据代码处理消息。

分布式搜索

初识ElasticSearch

ElasticSearch是一款非常强大的开源搜索引擎，可以帮助我们从海量数据中快速找到需要的内容。

Elastic Stack是以elasticsearch为核心的技术栈，包括beats、Logstash、 kibana、elasticsearch。

ElasticSearch的底层是基于Lucene实现的。

Lucene的优势：

易扩展
高性能（基于倒排索引）

Lucene的缺点：

只限于Java语言开发
学习曲线陡峭
不支持水平扩展

相比与lucene，ElasticSearch具备下列优势：

支持分布式，可水平扩展
提供Restful接口，可被任何语言调用

倒排索引

elasticsearch采用倒排索引：对文档内容分词，对词条创建索引，并记录词条所在文档的信息。查询时先根据词条查询到文档id，而后获取到文档。

文档（document）：每条数据就是一个文档
词条（term）：文档按照语义分成的词语

词条 ( term)	文档id
小米	1,3,4
手机	1,2
华为	2,3
充电器	3
手环	4

比较擅长基于文本内容进行搜索。

索引

索引（index）：相同类型的文档的集合

映射（mapping）：索引中文档的字段约束信息，类似表的结构约束

概念对比如下：

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笔记擅长海量数据的搜索、分析、计算

安装ES

查教程吧。

安装kibana

查教程+1

分词器

分词器的作用：es在创建倒排索引时需要对文档分词；在搜索时，需要对用户输入内容分词。

但默认的分词规则对中文处理并不友好。

处理中文分词，一般会使用IK分词器。

安装请查资料。

ik_max_word粒度较细，分出的词更多。

ik_smart粒度较粗，分出的词较少

POST /_analyze
{
    "text": "今天天气真不错",
    "analyzer": "ik_max_word"
}

字典的个性化拓展

要拓展ik分词器的词库，只需要修改一个ik分词器目录中的config目录中的IkAnalyzer.cfg.xml文件：

<?xml version="1.0" encoding="UTF-8"?>

<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
    <comment>IK Analyzer 扩展配置</comment> 
    <!--用户可以在这里配置自己的扩展字典,实际上是一个文件-->
    <entry key="ext_dict">ext.dic</entry>
    <!--用户可以在这里配置自己的扩展停止词字典 ***添加停用词词典-->
    <entry key="ext_stopwords">stopword.dic</entry>
</properties>

索引库操作

mapping映射属性

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

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

{
    "age": 21,
    "weight": 52.1,
    "isMarried": false,
    "info": "黑马程序员Java讲师",
    "email": "zy@itcast.cn",
    "score": [99.1, 99.5, 98.9],
    "name": {
        "firstName": "云",
        "lastName": "赵"
    }
}

索引库的CRUD

创建索引库

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

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

修改索引库

索引库和mapping一旦创建无法修改，但是可以添加新的字段，语法如下:

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

查询索引库

GET /索引库名

删除索引库

DELETE /索引库名

文档操作

添加文档

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

查询文档

GET /索引库名/_doc/文档id

小技巧：

字段拷贝可以使用copy_to属性将当前字段拷贝到指定字段。示例：

"all":{
    "type": "text",
    "analyzer": "ik_smart"
},
"brand":{
    "type": "keyword",
    "copy_to": "all"
}
"pic":{
    "type": "keyword",
    "copy_to": "all"
}
"address":{
    "type": "keyword",
    "copy_to": "all"
}

这样在all字段中就有下面这些字段的值了，从而实现从一个字段中搜索到多个字段的值。

删除文档

DELETE /索引库名/_doc/文档id

修改文档

方式一：全量修改，会删除旧文档，添加新文档

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

方式二：增量修改，修改指定字段值

PUT /索引库名/_update/文档id
{
    "doc":{
        "字段1":"值1",
    	"字段2":"值2",
    }
}

RestClient操作索引库

初始化JavaRestClient

1、引入es的RestHighLevelcient依赖

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

2、因为SpringBoot默认的ES版本是7.6.2，所以我们需要覆盖默认的ES版本

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

3、初始化RestHighLevelClient

RestHighLevelclient client = new RestHighLevelclient(Restclient.builder(//多个客户端只需要用,分割
    HttpHost.create("http://192.168.15.101:9200"),
    HttpHost.create("http://192.168.15.101:9201")
));

创建索引库

@Test
void testCreateHotelIndex() throws IOException {
    //1．创建Request对象,相当于 PUT /hotel
    CreateIndexRequest request = newCreateIndexRequest("hotel");
    //2．请求参数，MAPPING＿TEMPLATE是静态常量字符串，内容是创建索引库的DSL语句
    request.source(MAPPING_TEMPLATE,XContentType.JSON);
    //3．发起请求
    client.indices().create(
        request,RequestOptions.DEFAULT
    ); 
}

删除索引库

@Test
void testDeleteHotelIndex() throws IOException {
    //1．创建Request对象,相当于 Delete /hotel
    DeleteIndexIndexRequest request = newDeleteIndexRequest("hotel");
  
    //2．发起请求
    client.indices().delete(
        request,RequestOptions.DEFAULT
    ); 
}

判断索引库是否存在

@Test
void testExistHotelIndex() throws IOException {
    //1．创建Request对象,相当于 GET /hotel
    GetIndexRequest request = newGetIndexRequest("hotel");
  
    //2．发起请求
    boolean exists = client.indices().exists(
        request,RequestOptions.DEFAULT
    ); 
    if(exists){
        sout.println("存在捏");
    }
}

RestClient操作文档

初始化JavaRestClient

同上

添加酒店数据到索引库

@Test
void testIndexDocument() throws IOException {
    // 1.创建request对象
    IndexRequest request = new IndexRequest("indexName").id("1");
    // 2.准备JSON文档
    request.source("{\"name\": \"Jack\"\"age\": 21}"，XContentType.JSON);
    //3.发送请求
    client.index(request，RequestOptions.DEFAULT);
}

@Test
void testIndexDocument() throws IOException {
    Hotel  ht = hotelService.getById(61083L);
    //注意要有与实体库相关的实现类，实现从数据库数据到索引库数据的转换
    HotelDoc htDoc = new HotelDoc(hotel);//记得定义相关构造函数
    // 1.创建request对象
    IndexRequest request = new IndexRequest("indexName").id(hotel.getId().toString());
    // 2.准备JSON文档,此处使用fastJSON进行序列化
    request.source(JSON.toJSONString(hotelDoc),XContentType.JSON);
    //3.发送请求
    client.index(request，RequestOptions.DEFAULT);
}

根据id查询酒店数据

@Test
void testGetDocumentById() throws IOException {
    // 1.创建request对象
    GetRequest request = new GetRequest("indexName","1");
    // 2.发送请求，得到结果
    GetResponse response = cient.get(request， RequestOptions.DEFAULT);
    // 3.解析结果
    String json = response.getSourceAsString();
    System.out.println(json);
}

@Test
void testGetDocumentById() throws IOException {
    // 1.创建request对象
    GetRequest request = new GetRequest("hotel","61083");
    // 2.发送请求，得到结果
    GetResponse response = cient.get(request， RequestOptions.DEFAULT);
    // 3.解析结果
    String json = response.getSourceAsString();
    //使用fastJSON的解析函数
    HotelDoc htDoc = JSON.parseObject(json,HotelDoc.class);
    
    System.out.println(htDoc);
}

根据id修改文档

修改文档数据有两种方式:
方式一：全量更新。再次写入id一样的文档，就会删除旧文档，添加新文档

方式二：局部更新。只更新部分字段，我们演示方式二

@Test
void testUpdateDocumentById() throws IOException {
    // 1.创建request对像
    UpdateRequest request = new UpdateRequest("indexName"，"1");
    //2.准备参数，每2个参数为一对 key value
    request.doc(
        "age",18,
        "name","Rose"
    );
    // 3.更新文档
    client.update(request，RequestOptions.DEFAULT);
}

根据id删除文档

@Test
void testDeleteDocumentById() throws IOException {
    // 1.创建request对像
    DeleteRequest request = new DeleteRequest("indexName"，"1");
    // 2.删除文档
    client.delete(request，RequestOptions.DEFAULT);
}

批量导入数据

思路
1.利用mybatis-plus查询酒店数据

2.将查询到的酒店数据 (Hotel) 转换为文档类型数据 (HotelDoc)

3，利用JavaRestClient中的Bulk批处理，实现批量新增文档，示例代码如下

@Test
void testBulk() throws IOException {
    // 批量查询酒店数据
    List<Hotel> hotels = hotelService.list();
    
    // 1.创建BuLk请求
    BulkRequest request = new BulkRequest();
    //2.添加要批量提交的请求 
    for (Hotel hotel :hotels){
        // 换为文档类型HoteLDoc
        HotelDoc hotelDoc = new HotelDoc(hotel);
        // 创建新增文档的Request对象
        reguest.add(new IndexReguest("hotel")
                .id("61038").source(JSON.toJSONString(hotelDoc), XContentType.JSON));
    }
    // 3.发起bulk请求
    client.bulk(request，RequestOptions.DEFAULT);
}

ElasticSearch 搜索功能

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查询: 全文检索查询的一种，会对用户输入内容分词，然后去倒排索引库检索，语法:

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

其中indexName是你的索引库名称，FIELD指定字段，TEXT指定内容

multi_match:与match查询类似，只不过允许同时查询多个字段，语法：

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

其中的indexName是你的索引库名称，field指定字段列表，query指定搜索内容。

精确查询

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

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

term查询:

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

其中的indexName是你的索引库名称，FILED指定字段，value指定值。

range查询：

GET /indexName/_search{
    "query" : {
        "range" : {
            "FIELD" : {
                "gte" : "VALUE",
                "lte" : "VALUE"
            }
        }
    }
}

其中的indexName是你的索引库名称，FILED指定字段，gte指定大于等于，lte指定小于等于，gt指定大于，lt指定小于。

地理查询

根据经纬度查询。常见的使用场景包括：携程（搜索我附近的酒店）、滴滴（搜索我附近的出租车）

geo_bounding_box:查询geo_point值落在某个矩形范围的所有文档

GET /indexName/_search{
    "query" : {
        "geo_bounding_box" : {
            "FIELD" : {
                "top_left" : {
                    "lat" : "维度值",
                    "lon" : "经度值"
                },
                "bottom_right" : {
                    "lat" : "维度值",
                    "lon" : "经度值"
                },
            }
        }
    }
}

其中的indexName是你的索引库名称，top_left指定左上角，bottom_right指定右下角。

geo_distance：查询到指定中心点小于某个距离值的所有文档

GET /indexName/_search{
    "query" : {
        "geo_distance" : {
            "distance": "15km",
            "FIELD": "维度值,经度值"
        }
    }
}

其中的indexName是你的索引库名称，distance指定距离，FIELD指定坐标中心点。

复合查询

复合(compound)查询: 复合查询可以将其它简单查询组合起来，实现更复杂的搜索逻辑

function score 查询

function score: 算分函数查询，可以控制文档相关性算分，控制文档排名。例如百度竞价。

当我们利用match查询时，文档结果会根据与搜索词条的关联度打分(_score)，返回结果时按照分值降序排列。

TF-IDF算法：

$IDF(逆文档频率)=Log(\frac{文档总数}{包含词条的文档总数})$

$TF(词条频率)=\frac{词条出现次数}{文档中词条总数}$

$score = sum_i^n{TF*IDF}$

BM25算法，相较于TF-IDF在词频高速增长时受影响较小，曲线趋于水平

使用 function score query，可以修改文档的相关性算分 (query score)，根据新得到的算分排序。

 GET /hotel/_search
{
    "query": {
        "function_score": {
            "query":{
                "match":{
                    "all": "外滩"
                }
            },
            "functions":[
                {
                    "filter": {
                        "term":{
                            "id": "1"
                        }
                    },
                    "weight": 10
                }
            ],
            "boost_mode": "multiply"
        }
    }
}

其中query负责处理原始查询条件，搜索文档并根据相关性打分。

functions中的filter是过滤条件，符合条件的文档才会被重新算分。

functions中的weight是一种算法函数，其结果称为functions score，将来会与query score运算得到新算分，常见的算分函数有:

weight: 给一个常量值，作为函数结果 (function score)
field_value_factor: 用文档中的某个字段值作为函数结果
random_score:随机生成一个值，作为函数结果
script_score:自定义计算公式，公式结果作为函数结果

boost_mode定义了加权模式，定义function score与query score的运算方式，包括:

multiply:两者相乘。默认就是这个
replace: 用function score 替换 query score
其它: sum、avg、max、min

Boolean Query 查询

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

must:必须匹配每个子查询，类似“与”
should:选择性匹配子查询，类似“或“
must_not:必须不匹配，不参与算分，类似“非”
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
                        }
                    }
                },
                {}
   			]
        }
    }
}

搜索结果处理

排序

elasticsearch支持对搜索结果排序，默认是根据相关度算分( score)来排序。可以排序字段类型有: keyword类型数值类型、地理坐标类型、日期类型等

GET /indexName/_search{
    "query" : {
        "match_all": {}
    },
    "sort": [
        {
            "FIELD": "desc" //字段降序排
        },
        {
            "_geo_distance" :{ //坐标排序
                "FIELD" : "中心点纬度值，经度值",
                "order" : "asc",
                "unit" : "km"
            }
        },
        {}
    ]
}

其中indexName是你的索引库名称，sort数组中的元素是用于指定按照字段升序或者降序，相同的则按照下一个元素排序，特殊的_geo_distance用于坐标排序，要给出中心点的坐标

分页

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

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

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

实际上是逻辑分页，截取分页后显示的文档。

深度分页问题

ES是分布式的，所以会面临深度分页问题。例如按price排序后，获取from = 990，size =10的数据：在不同的集群上有不同的数据，所以要对所有节点的数据进行汇总、整理、排序

首先在每个数据分片上都排序并查询前1000条文档。
然后将所有节点的结果聚合，在内存中重新排序选出前1000条文档
最后从这1000条中，选取从990开始的10条文档

如果搜索页数过深，或者结果集 (from +size)越大，对内存和CPU的消耗也越高。因此ES设定结果集查询的上限是10000。

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

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

分页对比

from + size

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

after search:

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

scroll

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

高亮

高亮：就是在搜索结果中把搜索关键字突出显示。

原理是这样的:

将搜索结果中的关键字用标签标记出来，交给前端
前端在页面中给标签添加Css样式，之后激动显示

一定要用带关键字的查询，而不是只有关于内容的查询，且默认需要高亮字段和搜索字段一致

GET /hotel/_search{
    "query" :{
        "match" :{
            "FIELD": "TEXT"
        }
    },
    "highlight" : {
        "fields": {//指定要高亮的字段
            "FIELD": {
                "pre_tags":"<em>", // 用来标记高亮字段的前置标签
                "post_tags":"</em>", // 用来标记高亮字段的后置标签
                "require_field_match" : "false" //无需高亮和搜索一致(匹配)
            }
        }
    }
}

RestClient查询文档

快速入门

我们通过match all来演示下基本的API，先看请求DSL的组织

@Test
void testMatchAll() throws IOException {
    // 1.准备Request
    SearchRequest request = new SearchRequest("hotel");
    // 2.组织DSL参数
request.source()
.query(QueryBuilders.matchAllQuery());
    // 3.发送请求，得到响应结果
SearchResponse response = client.search(request,RequestOptions.DEFAULT);
    //解析响应结果
    // 4.解析结果
    SearchHits searchHits = response.getHits();
    // 4.1.查询的总条数
    long total = searchHits.getTotalHits().value;
    //4,2查询的结果数组
    SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
        // 4.3.得到source
        String json = hit.getSourceAsString();
        // 4.4.打印
        System.out.println(json);
        
        //反序列化
        HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
        System.out.println("hotelDoc = " + hotelDoc);
    }

match查询

全文检索查询
全文检索的match和multi match查询与match all的API基本一致。差别是查询条件，也就是query的部分。同样是利用QueryBuilders提供的方法，仅改动准备DSL的部分

//单字查询
QueryBuilders.matchQuery("all","如家");
//多字段查询
 QueryBuilders.multiMatchQuery("如家","name","business".

所以实际上可以将各个部分抽取出来，在IDEA中选中抽取部分，按下ctrl+alt+m

精确查询

精确查询常见的有term查询和range查询，同样利用QueryBuilders实现

// 词条查询
QueryBuilders.termQuery("city"，"杭酬");
// 范围查询
QueryBuilders.rangeQuery("price").gte(100).lte(150);

复合查询

复合查询-boolean query
精确查询常见的有term查询和range查询，同样利用QueryBuilders实现

// 创建布尔查询
BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
//添加must亲件
boolQuery.must(QueryBuilders.termQuery("city"，"杭州"));
// 添加filter亲件
boolQuery.filter(QueryBuilders.rangeQuery("price").lte(250));

复合查询- function score

Function score查询可以控制文档的相关性算分，使用方式如下

FunctionScoreQueryBuilder functionScoreQueryBuilder = QueryBuilders.functionScoreQuery(
    QueryBuilders.matchQuery("name","外滩"),
    new FunctionScoreQueryBuilder.FilterFunctionBuilder[]{
        new FunctionScoreQueryBuilder.FilterFunctionBuilder(
            QueryBuilders.termQuery("brand"，"如家"),
            ScoreFunctionBuilders.weightFactorFunction(5)
            )
    }
);
sourceBuilder.query(functionScoreQueryBuilder);

总之要构建查询条件，只要记住一个类：QueryBuilders

排序、分页、高亮

搜索结果的排序和分页是与query同级的参数，对应的API如下

//查询
request.source().query(QueryBuilders.matchAllQuery());
//分页
request.source().from(0).size(5);
//价格排序
request.source().sort("price",SortOrder.ASC);

//高亮
request.source().highlighter(new HighlightBuilder().field("name").requireFieldMatch(false));

值得注意的点在于高亮的结果是与source同级别的，需要额外的处理

// 从hit获取source
HotelDoc hotelDoc = JSON.parseObject(hit.getSourceAsString(),HotelDoc.class);
//从hit获取highlight字段，处理高亮
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);
    }

}

数据聚合

聚合的种类

聚合(aggregations)可以实现对文档数据的统计、分析、运算。聚合的字段一定是不分词的。聚合常见的有三类

桶(Bucket)聚合:用来对文档做分组
- TermAggregation:按照文档字段值分组
- Date Histogram:按照日期阶梯分组，例如一周为一组，或者一月为一组
度量(Metric)聚合:用以计算一些值，比如:最大值、最小值、平均值等
- AVg:求平均值
- Max:求最大值
- Min:求最小值
- Stats:同时求max、min、avg、sum等
管道(pipeline)聚合:其它聚合的结果为基础做聚合

DSL实现聚合

bucket聚合

现在，我们要统计所有数据中的酒店品牌有几种，此时可以根据酒店品牌的名称做聚合类型为term类型，DSL示例:

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

默认情况下，Bucket聚合是对索引库的所有文档做聚合，我们可以限定要聚合的文档范围，只要添加query条件即可

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

Metrics聚合

其位置处在bucket聚合中，例如，我们要求获取每个品牌的用户评分的min、max、avg等值我们可以利用stats聚合

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

RestAPI实现聚合

RestAPI实现聚合
我们以品牌聚合为例，演示下Java的RestClient使用，先看请求组装

request.source().size(0);//删除文档结果的影响
request.source().aggregation(
	AggregationBuilders
         .terms("brand_agg")
         .field("brand")
         .size(20)
);

//1.准备Request
SearchRequest request =new SearchRequest("hotel");
// 2.准备DSL
// 2.1.设size
request.source().size(0);
// 2.2.聚合
request.source().aggregation(AggregationBuilders
                             .terms("brandAgg")
                             .field("brand")
.size(10)
);
//3.发出请求
SearchResponse response = client.search(request, RequestOptions.DEFAULT)
 //4.解折结果
    
 Aggregations aggregations =response.getAggregations();
//4.1根据名称获取聚合结果
Terms brandTerms =aggregations.get("brand_agg");
//4.2获取桶
List<? extends Terms.Bucket> buckets = brandTerms.getBuckets();
//4.3 遍历
for(Terms.Bucket bucket : buckets){
    //4.4 获取key，也就是品牌信息
    String brandName = bucket.getKeyAsString();
    System.out.println(brandName);
}

动态查询条件，在做聚合的时候与做查询的时候的查询条件保持一致，即在准备DSL时，在设置size之前添加准备query的部分。

自动补全

拼音分词器

要实现根据字母做补全，就必须对文档按照拼音分词。在GitHub上恰好有elasticsearch的拼音分词插件。

安装方式与IK分词器一样，分三步:

解压
上传到虚拟机中，elasticsearch的plugin目录
重启elasticseakch
测试

自定义分词器

elasticsearch中分词器(analyzer)的组成包含三部分:

character filters:在tokenizer之前对文本进行处理。例如删除字符、替换字符
tokenizer:将文本按照一定的规则切割成词条(term)。例如keyword，就是不分词;还有ik_smart
tokenizer filter:将tokenizer输出的词条做进一步处理。例如大小写转换、同义词处理、拼音处理等

我们可以在创建索引库时，通过settings来配置自定义的analyzer(分词器)

PUT /test
{
    "settings":{
        "analysis":{
            "analyzer":{ // 自定义分词器
                "my_analyzer":{ // 分词器名称
                    "tokenizer": "ik_max_word",//先这个
                    //"tokenizer": "keyword", //直接先不分词的写法
                    "filter": "mypy"  //再这个
                }
            },
            "filter":{//自定义tokenizer filter
                "mypy": { // 过滤器名称
                    "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
                }
            }
        }
    }     
}

拼音分词器适合在创建倒排索引的时候使用，但不能在搜索的时候使用。会导致同音词在同一个倒排索引词条中

因此字段在创建倒排索引时应该用my_analyzer分词器;字段在搜索时应该使用iksmart分词器:

PUT /test
{
    "settings":{
        "analysis":{
            "analyzer":{ // 自定义分词器
                "my_analyzer":{ // 分词器名称
                    "tokenizer": "ik_max_word",
                    "filter": "mypy"
                }
            },
            "filter":{//自定义tokenizer filter
                "mypy": {...}
            }
        }
    },
	"mappings":{
        "properties":{
            "name":{
                "type": "text",
                "analyzer" : "my_analyzer",
                "search_analyzer": "ik_smart"
            }
        }
    }
}

自动补全查询

为了提高补全查询的效率，对于文档中字段的类型有一些约束

参与补全查询的字段必须是completion类型
字段的内容一般是用来补全的多个词条形成的数组

//示例数据
POST test/_doc
{
    "title":["Sony","WH-1000XM3"]
}

POST test/_doc
{
    "title":["SK-II","PITERA"]
}

POST test/_doc
{
    "title":["Nintendo","switch"]
}

查询语法如下：

//自动补全查询
GET /test/_search
{
    "suggest":{//定义自动补全的字段
        "title_suggest":{//自定义的一个名称，作为需求内容
            "text":"s",// 关键字
            "completion":{
                "field":"title",// 补全查询的字段
                "skip_duplicates":true,//跳过重复的
                "size":10 //获取前10条结果
            }
        }
    }
}

//1.准备请求
SearchRequest request = new SearchRequest("hotel");
//2.请求参数
request.source()
    .suggest(new SuggestBuilder().addSuggestion("mysuggestion",
                                                SuggestBuilders
                                                	.completionSuggestion("title")
                                                .prefix("h")
                                                .skipDuplicates(true)
                                                .size(10)
));
//3.发送请求
client.search(request,RequestOptions.DEFAULT);
//4.结果解析，根据返回的json结果逐层解析

Suggest suggest = response.getSuggest();
//4.1.根据名称获取补全结果
CompletionSuggestion suggestion= suggest.getSuggestion("mysuggestion");
//4.2.获取options并遍历
for (completionSuggestion.Entry.0ption option : suggestion.getoptions()){
// 4.3.获取一个option中的text，也就是补全的词条
    String text = option.getText().string();
    System.out.println(text);
}

数据同步

数据同步思路分析

ElasticSearch中的酒店数据来自于mysql数据库，因此mysql数据发生改变时，ElasticSearch也必须跟着改变，这个就是ElasticSearch与mysql之间的数据同步

方案一：同步调用

1.1.管理服务写入数据库

1.2.管理服务调用搜索服务的更新索引库接口

1.3.搜索服务更新ElasticSearch

**优点:**实现简单，粗暴

**缺点:**业务耦合度高方式二:异步通知

方案二：异步通知

将同步中的1.2调用服务的过程采用MQ的方式管理服务发送消息，搜索服务接收消息

**优点:**低耦合，实现难度一般

**缺点:**依赖mq的可靠性

方案三：监听binlog

将同步中的1.2调用服务的过程删除。加入诸如canal的中间件，监听mysql的变化，在数据更新后通知搜索服务更新。

**优点:**完全解除服务间耦合

**缺点:**开启binlog增加数据库负担、实现复杂度高

ElasticSearch集群

搭建ES集群

单机的elasticsearch做数据存储，必然面临两个问题:海量数据存储问题、单点故障问题。

海量数据存储问题:将索引库从逻辑上拆分为N个分片(shard)，存储到多个节点
单点故障问题:将分片数据在不同节点备份(replica)

ES中的集群名称一致就会自动形成同一个集群

具体查攻略吧。

监听ES集群状态可以使用Cerebro，使用查攻略。

对于分片的部署和副本数量等问题，可以在创建索引库时在settings中进行设置。

集群职责与脑裂

elasticsearch中的每个节点角色都有自己不同的职责，因此建议集群部署时，每个节点都有独立的角色

一般是负载均衡链接到多个coordinating，在到data，再到master eligible

节点类型	配置参数	默认值	节点职责
master eligible	node.master	true	备选主节点:主节点可以管理和记录集群状态、决定分片在哪个节点、处理创建和删除索引库的请求
data	node.data	true	数据节点:存储数据、搜索、聚合、CRUD
ingest	node.ingest	true	数据存储之前的预处理
coordinating	上面3个参数都为false 则为coordinating节点	无	路由请求到其它节点合并其它节点处理的结果,返回给用户

默认情况下，每个节点都是master eligible节点，因此一旦master节点宕机，其它候选节点会选举一个成为主节点。当主节点与其他节点网络故障时，可能发生脑裂问题。

为了避免脑裂，需要要求选票超过(eligible节点数量+1)/2才能当选为主，因此eligible节点数量最好是奇数。对应配置项是discovery.zen.minimum_master_nodes，在es7.0以后，已经成为默认配置，因此一般不会发生脑裂问题

集群分布式存储

ElasticSearch会通过hash算法来计算文档应该存储到哪个分片

算法与分片数量有关，因此索引库一旦创建，分片数量不能修改!

新增文档流程：

1、新增文档id到达协调节点

2、hash运算得到存储的分片位置

3、路由到对应的分片

4、对应的分片保存文档

5、同步内容给副本分片存储副本

6、返回结果到协调节点

7、协调节点返回结果给用户

集群分布式查询

ElasticSearch的查询分成两个阶段:

scatter phase：分散阶段，coordinatingnode会把请求分发到每一个分片
gather phase：聚集阶段，coordinating node汇总data node的搜索结果，并处理为最终结果集返回给用户

集群故障转移

集群的master节点会监控集群中的节点状态，如果发现有节点岩机，会立即将宕机节点的分片数据迁移到其它节点，确保数据安全，这个叫做故障转移。

具体来说：如果master宕机后，EligibleMaster选举为新的主节点。

master节点监控分片、节点状态，将故障节点上的分片转移到正常节点，确保数据安全。

微服务保护

初识Sentinel

雪崩问题及解决方案

微服务调用链路中的某个服务故障，引起整个链路中的所有微服务都不可用，这就是雪崩。

解决雪崩问题的常见方式有四种

超时处理: 设定超时时间，请求超过一定时间没有响应就返回错误信息，不会无休止等待
舱壁模式（线程隔离）: 限定每个业务能使用的线程数，避免耗尽整个tomcat的资源，因此也叫线程隔离
熔断降级: 由断路器统计业务执行的异常比例，如果超出阈值则会熔断该业务，拦截访问该业务的一切请求。
流量控制: 限制业务访问的QPS（每秒处理的服务数量），避免服务因流量的突增而故障。

服务保护技术对比

	Sentinel	Hystrix
隔离策略	信号量隔离	线程池隔离/信号量隔离
熔断降级策略	基于慢调用比例或异常比例	基于失败比率
实时指标实现	滑动窗☐	滑动窗☐(基于RxJava)
规则配置	支持多种数据源	支持多种数据源
扩展性	多个扩展点	插件的形式
基于注解的支持	支持	支持
限流	基于QPS,支持基于调用关系的限流	有限的支持
流量整形	支持慢启动、匀速排队模式	不支持
系统自适应保护	支持	不支持
控制台	开箱即用,可配置规则、查看秒级监控、机器发现等	不完善
常见框架的适配	Servlet、 Spring Cloud、 Dubbo、gRPC等	Servlet、 Spring Cloud Netflix

Sentinel介绍和安装

Sentinel是阿里巴巴开源的一款微服务流量控制组件。官网地址:略

安装：

sentinel官方提供了UI控制台，方便我们对系统做限流设置。大家可以在GitHub下载

运行：

java -jar sentinel-dashboard-l.8.1.jar

然后访问localhost:8080 即可看到控制台页面，默认的账户和密码都是sentinel

如果要修改Sentinel的默认端口、账户、密码，可以通过下列配置:

配置项	默认值	说明
server .port	8080	服务端☐
sentinel. dashboard.auth.use rname	senti nel	默认用户名
sentinel. dashboard.auth. password	sentinel	默认密码

例如:

java -jar sentinel-dashboard-1.8.1.jar -Dserver.port=8090

微服务整合Sentinel

我们在order-service中整合Sentinel，并且连接Sentinel的控制台，步骤如下

1.引入sentinel依赖:

<!--sentinel-->
<dependency
            <groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-sentinel</artifactId></dependency>

2.配置控制台地址:

spring:
	cloud:
		sentinel:
			transport:
				dashboard: localhost:8080

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

流量控制

快速入门

簇点链路: 就是项目内的调用链路，链路中被监控的每个接口就是一个资源。默认情况下sentinel会监控SpringMVC的每一个端点(Endpoint)，因此SpringMVC的每一个端点(Endpoint)就是调用链路中的一个资源。

流控、熔断等都是针对簇点链路中的资源来设置的，因此我们可以点击对应资源后面的按钮来设置规则:

点击资源/order/{orderld}后面的流控按钮，就可以弹出表单。表单中可以添加流控规则。

针对来源是设置哪些访问受控，阈值类型设置针对什么情况控制，单击阈值设置超出多少开始限制。

使用jemeter测试。

流控模式

在添加限流规则时，点击高级选项，可以选择三种流控模式:

直接:统计当前资源的请求，触发阈值时对当前资源直接限流，也是默认的模式
关联:统计与当前资源相关的另一个资源触发阈值时，对当前资源限流
链路:统计从指定链路访问到本资源的请求，触发阈值时，对指定链路限流

关联模式

统计与当前资源相关的另一个资源，触发值时，对当前资源限流使用场景

比如用户支付时需要修改订单状态，同时用户要查询订单。查询和修改操作会争抢数据库锁，产生竞争。业务需求是有限支付和更新订单的业务，因此当修改订单业务触发阈值时，需要对查询订单业务限流。

在当前资源/read配置关联模式的时候，选择关联资源/write，则当/write资源访问量触发阈值时，就会对/read资源限流，避免影响/write资源

链路模式

只针对从指定链路访问到本资源的请求做统计，判断是否超过阈值。例如有两条请求链路:
/test1>/common
/test2 >/common
如果只希望统计从/test2进入到/common的请求，则可以这样配置

在/common中配置，设置流控模式为链路，且入口资源为/test2

Sentinel默认只标记controller中的方法为资源，如果要标记其它方法，需要利用@SentinelResource注解

@SentinelResource("goods")
public void queryGoods(){
    System.err.println("查询商品”);
}

Sentinel默认会将Controller方法做context整合，导致链路模式的流控失效，需要修改application.yml，添加配置:

spring:
	cloud:
		sentinel:
			web-context-unify: false # 关闭context整合

流控效果

流控效果是指请求达到流控阈值时应该采取的措施，包括三种:

快速失败: 达到阈值后，新的请求会被立即拒绝并抛出FowException异常。是默认的处理方式。
warm up: 预热模式，对超出阈值的请求同样是拒绝并抛出异常。但这种模式阈值会动态变化，从一个较小值逐渐增加到最大阈值。
排队等待:让所有的请求按照先后次序排队执行，两个请求的间隔不能小于指定时长

warm up 预热模式

是应对服务冷启动的一种方案。请求阈值初始值是threshold/coldFactor，持续指定时长后，逐渐提高到threshold值。而coldFactor的默认值是3。

这样可以避免冷启动时并发过高的情况。

排队等待

当请求超过QPS阈值时，快速失败和warmup会拒绝新的请求并抛出异常。

而排队等待则是让所有请求进入一个队列中然后按照阈值允许的时间间隔依次执行。后来的请求必须等待前面执行完成，如果请求预期的等待时间超出最大时长则会被拒绝。

例如:QPS=5，意味着每200ms处理一个队列中的请求;timeout=2000，意味着预期等待超过2000ms的请求会被拒绝并抛出异常

热点参数限流

之前的限流是统计访问某个资源的所有请求，判断是否超过QPS阈值。而热点参数限流是分别统计参数值相同的请求判断是否超过QPS阈值。

配置示例

![外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传](https://img-home.csdnimg.cn/images/20230724024159.png?origin_url=assets%5C%E5%BE%AE%E4%BF%A1%E6%88%AA%E5%9B%BE_20240130141854.png&pos_id=img-DdrLq3lP-1;

代表的含义是:对hot这个资源的0号参数(第一个参数)做统计，每1秒相同参数值的请求数不能超过5

在热点参数限流的高级选项中，可以对部分参数设置例外配置:

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

结合上一个配置，这里的含义是对0号的long类型参数限流，每1秒相同参数的QPS不能超过5，有两个例外:

如果参数值是100，则每1秒允许的QPS为10
如果参数值是101，则每1秒允许的QPS为15

值得注意的一点在于：热点参数限流对默认的SpringMVC资源无效

所以要添加@SentinelResource注解

还有一点需要注意的点在于热点参数规制的参数类型只允许基本类型和String

隔离和降级

虽然限流可以尽量避免因高并发而引起的服务故障，但服务还会因为其它原因而故障。而要将这些故障控制在一定范围避免雪崩，就要靠线程隔离(舱壁模式)和熔断降级手段了。

不管是线程隔离还是熔断降级，都是对客户端(调用方)的保护。

FeignClient整合Sentinel

1、修改OrderService的application.yml文件，开启Feign的Sentinel功能

feign:
	sentinel:
		enabled: true # 开Feign的sentinel功能

2、给FeignClient编写失败后的降级逻辑

方式一：FallbackClass，无法对远程调用的异常做处理
方式二：FallbackFactory，可以对远程调用的异常做处理，我们选择这种

步骤一：在feing-api项目中定义类，实现FallbackFactory

@slf4j
public class UserclientFallbackFactory implements FallbackFactory<Userclient>{
    @Override
    public UserClient create(Throwable throwable){
        //创建UserClient接口实现类，实现其中的方法，编写失败降级的处理逻辑
        return new UserClient(){
            @Override 
            public User findById(Long id){
                // 记录异常信息
            	log.error("查询用户失败"，throwable);
            	//根据业务需求返回默认的数据，这里是空用户
            	return new User();
            }
        };
    }
}

步骤二：在feing-api项目中的DefaultFeignConfiguration类中将UserClientFallbackFactory注册为一个Bean

@Bean
public UserclientFallbackFactory userclientFallback(){
    return new UserClientFallbackFactory();
}

步骤三：在feing-api项目中的UserClient接口中使用UserClientFallbackFactory

@FeignClient(value = "userservice", fallbackFactory = UserclientFallbackfactory.class)
public interface Userclient {
    @GetMapping("/user/{id}")
    User findById(@PathVariable("id") Long id);
}

线程隔离(舱壁模式)

线程隔离有两种方式实现:

线程池隔离
信号量隔离(Sentinel默认采用)

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

线程池隔离

优点：1、支持主动超时 2、支持异步调用

缺点：1、线程的额外开销比较大

场景：低扇出（依赖于n个其他服务，n较小）

操作方法

信号量隔离
优点：1、轻量级 2、无额外开销

缺点：1、不支持主动超时 2、不支持异步调用

场景：高频调用高扇出

在Sentinel中默认采用的就是信号量隔离的方式，可以在添加限流规则时，可以选择两种阈值类型，将类型选择为线程数即可启动信号量隔离。

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

QPS：就是每秒的请求数，在快速入门中已经演示过
线程数:是该资源能使用用的tomcat线程数的最大值。也就是通过限制线程数量，实现舱壁模式。

熔断降级

熔断降级是解决雪崩问题的重要手段。其思路是由断路器统计服务调用的异常比例、慢请求比例，

如果超出阈值则会熔断该服务。即拦截访问该服务的一切请求；而当服务恢复时，断路器会放行访问该服务的请求。

断路器控制熔断和放行是通过状态机来完成的

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

授权规则状态机包括三个状态：

closed：关闭状态，断路器放行所有请求，并开始统计异常比例、慢请求比例。超过阈值则切换到open状态
open：打开状态，服务调用被熔断，访问被熔断服务的请求会被拒绝，快速失败，直接走降级逻辑。Open状态5秒后会进入half-open状态
half-open：半开状态，放行一次请求，根据执行结果来判断接下来的操作。
请求成功：则切换到closed状态
请求失败：则切换到open状态

熔断策略

断路器熔断策略有三种：慢调用、异常比例、异常数

慢调用：业务的响应时长（RT）大于指定时长的请求认定为慢调用请求。在指定时间内，如果请

求数量超过设定的最小数量，慢调用比例大于设定的阈值，则触发熔断。

例如：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

解读：RT超过500ms的调用是慢调用，统计最近10000ms内的请求，如果请求量超过10次，并且慢调用比例不低于0.5，则触发熔断，熔断时长为5秒。然后进入half-open状态，放行一次请求做测试。

异常比例或异常数：统计指定时间内的调用，如果调用次数超过指定请求数，并且出现异常的比例

达到设定的比例阈值（或超过指定异常数），则触发熔断。

例如，一个异常比例设置：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

解读：统计最近1000ms内的请求，如果请求量超过10次，并且异常比例不低于0.4，则触发熔断。

一个异常数设置：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

解读：统计最近1000ms内的请求，如果请求量超过10次，并且异常比例不低于2次，则触发熔断。

授权规制

基本规则

授权规则可以对调用方的来源做控制，有白名单和黑名单两种方式。

白名单：来源（origin）在白名单内的调用者允许访问
黑名单：来源（origin）在黑名单内的调用者不允许访问

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

点击左侧菜单的授权，可以看到授权规则

资源名：就是受保护的资源，例如/order/{orderId}
流控应用：是来源者的名单，即origin。怎么获得看下面
- 如果是勾选白名单，则名单中的来源被许可访问。
- 如果是勾选黑名单，则名单中的来源被禁止访问。

如何获取origin

Sentinel是通过RequestOriginParser这个接口的parseOrigin来获取请求的来源的。

public interface RequestOriginParser {
    /**
     * 从请求request对象中获取origin，获取方式自定义
     */
    String parseOrigin(HttpServletRequest request);
}

这个方法的作用就是从request对象中，获取请求者的origin值并返回。

默认情况下，sentinel不管请求者从哪里来，返回值永远是default，也就是说一切请求的来源都被认为是一样的值default。

因此，我们需要自定义这个接口的实现，让不同的请求，返回不同的origin。

例如order-service服务中，我们定义一个RequestOriginParser的实现类：

@Component
public class HeaderOriginParser implements RequestOriginParser {
    @Override
    public String parseOrigin(HttpServletRequest request) {
        // 1.获取请求头
        String origin = request.getHeader("origin");
        // 2.非空判断
        if(StringUtils.isEmpty(origin)){
            origin = "blank";
        }
        return origin;
    }
}

我们会尝试从request-header中获取origin值，这就要求我们提前给网关添加请求头。

给网关添加请求头

既然获取请求origin的方式是从reques-header中获取origin值，我们必须让所有从gateway路由到微服务的请求都带上origin头。

这个需要利用之前学习的一个GatewayFilter来实现，AddRequestHeaderGatewayFilter。

修改gateway服务中的application.yml，添加一个defaultFilter：

spring:
  cloud:
    gateway:
      default-filters:
        - AddRequestHeader=origin,gateway # 添加名为origin的请求头，值为gateway
      routes:
       # ...略

这样，从gateway路由的所有请求都会带上origin头，值为gateway。而从其它地方到达微服务的请求则没有这个头。

自定义异常结果

默认情况下，发生限流、降级、授权拦截时，都会抛出异常到调用方。异常结果都是flow limmiting（限流）。这样不够友好，无法得知是限流还是降级还是授权拦截。

而如果要自定义异常时的返回结果，需要实现BlockExceptionHandler接口：

public interface BlockExceptionHandler {
    /**
     * 处理请求被限流、降级、授权拦截时抛出的异常：BlockException
     */
    void handle(HttpServletRequest request, HttpServletResponse response, BlockException e) throws Exception;
}

这个方法有三个参数：

HttpServletRequest request：request对象
HttpServletResponse response：response对象
BlockException e：被sentinel拦截时抛出的异常

这里的BlockException包含多个不同的子类：

异常	说明
FlowException	限流异常
ParamFlowException	热点参数限流的异常
DegradeException	降级异常
AuthorityException	授权规则异常
SystemBlockException	系统规则异常

下面，我们就在order-service定义一个自定义异常处理类：

@Component
public class SentinelExceptionHandler implements BlockExceptionHandler {
    @Override
    public void handle(HttpServletRequest request, HttpServletResponse response, BlockException e) throws Exception {
        String msg = "未知异常";
        int status = 429;
 
        if (e instanceof FlowException) {
            msg = "请求被限流了";
        } else if (e instanceof ParamFlowException) {
            msg = "请求被热点参数限流";
        } else if (e instanceof DegradeException) {
            msg = "请求被降级了";
        } else if (e instanceof AuthorityException) {
            msg = "没有权限访问";
            status = 401;
        }
 
        response.setContentType("application/json;charset=utf-8");
        response.setStatus(status);
        response.getWriter().println("{\"msg\": " + msg + ", \"status\": " + status + "}");
    }
}

规制持久化

规则管理模式

原始模式：Sentinel的默认模式，将规则保存在内存，重启服务会丢失，
pull模式：保存在本地文件或数据库，定时去读取
push模式：保存在nacos，监听变更实时更新

pul模式：控制台将配置的规则推送到Sentinel客户端，而客户端会将配置规则保存在本地文件或数据库中。以后会定时去本地文件或数据库中查询，更新本地规则。

**push模式：**控制台将配置规则推送到远程配置中心，例如Nacos。Sentinel客户端监听Nacos，获取配置变更的推送消息，完成本地配置更新。

实现push模式

push模式实现最为复杂，依赖于nacos，并且需要修改Sentinel控制台源码。

1）修改order-service服务

修改OrderService，让其监听Nacos中的sentinel规则配置。具体步骤如下:

1、引入依赖
在order-service中引入sentinel监听nacos的依赖

<dependency>
    <groupId>com.alibaba.csp</groupId>
    <artifactId>sentinel-datasource-nacos</artifactId>
</dependency>

2、配置nacos地址

在order-service中的application.yml文件配置nacos地址及监听的配置信息

spring:
  cloud:
    sentinel:
      datasource:
        flow:
          nacos:
            server-addr: localhost:8848 # nacos地址
            dataId: orderservice-flow-rules
            groupId: SENTINEL_GROUP
            rule-type: flow # 还可以是：degrade、authority、param-flow

2)修改sentinel-dashboard源码

SentinelDashboard默认不支持nacos的持久化，需要修改源码。

1、解压；解压sentinel源码包

然后并用IDEA打开这个项目，结构如下

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

2、修改nacos依赖

在sentinel-dashboard源码的pom文件中，nacos的依赖默认的scope是test，只能在测试时使用，这里要去除

<dependency>
    <groupId>com.alibaba.csp</groupId>
    <artifactId>sentinel-datasource-nacos</artifactId>
</dependency>

3、添加nacos支持
在sentinel-dashboard的test包下，已经编写了对nacos的支持，我们需要将其拷贝到main下。

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

4、修改nacos地址
然后，还需要修改测试代码中的NacosConfig类：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

修改其中的nacos地址，让其读取application.properties中的配置：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

在sentinel-dashboard的application.properties中添加nacos地址配置：

nacos.addr=localhost:8848

5、配置nacos数据源
另外，还需要修改com.alibaba.csp.sentinel.dashboard.controller.v2包下的FlowControllerV2类：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

让我们添加的Nacos数据源生效：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

6、修改前端页面
接下来，还要修改前端页面，添加一个支持nacos的菜单。
修改src/main/webapp/resources/app/scripts/directives/sidebar/目录下的sidebar.html文件。将其中的这部分注释打开

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

修改其中的文本：

<li ui-sref-active="active" ng-if="entry.appType==0">
    <a ui-sref="dashboard.flow({app: entry.app})">
        <i class="glyphicon glyphicon-filter"></i>&nbsp;&nbsp;流控规则-NACOS</a>
</li>

7、重新编译、打包项目
运行IDEA中的maven插件，编译和打包修改好的Sentinel-Dashboard
8、启动
启动方式跟官方一样：

java -jar sentinel-dashboard.jar

如果要修改nacos地址，需要添加参数：

java -jar -Dnacos.addr=localhost:8848 sentinel-dashboard.jar

分布式事务

理论基础

CAP定理

1998年，加州大学的计算机科学家 EricBrewer提出，分布式系统有三个指标:

Consistency(一致性)
Availability(可用性)
Partition tolerance (分区容错性)

Consistency(一致性)：用户访问分布式系统中的任意节点，得到的数据必须一致

Availability (可用性)：用户访问集群中的任意健康节点，必须能得到响应，而不是超时或拒绝

Partition(分区)：因为网络故障或其它原因导致分布式系统中的部分节点与其它节点失去连接，形成独立分区

Tolerance(容错):在集群出现分区时，整个系统也要持续对外提供服务

Eric Brewer说，分布式系统无法同时满足这三个指标。
这个结论就叫做 CAP 定理

BASE理论

BASE理论是对CAP的一种解决思路，包含三个思想:

Basically Available(基本可用)：分布式系统在出现故障时，允许损失部分可用性，即保证核心可用。
Soft state(软状态)：在一定时间内，允许出现中间状态，比如临时的不一致状态。
Eventurally Consistent(最终一致性)：虽然无法保证强一致性，但是在软状态结束后，最终达到数据一致。

解决分布式事务的思想和模型

全局事务：整个分布式事务
分支事务：分布式事务中包含的每个子系统的事务

而分布式事务最大的问题是各个子事务的一致性问题，因此可以借鉴CAP定理和BASE理论:

AP模式：各子事务分别执行和提交，允许出现结果不一致，然后采用弥补措施恢复数据即可，实现最终一致。
CP模式：各个子事务执行后互相等待，同时提交，同时回滚，达成强一致。但事务等待过程中，处于弱可用状态。

初识Seata

Seata的架构

Seata事务管理中有三个重要的角色:

TC(Transaction Coordinator)-事务协调者：

维护全局和分支事务的状态，协调全局事务提交或回滚。
TM (Transaction Manager)-事务管理器：

定义全局事务的范围：开始全局事务、提交或回滚全局事务
RM(Resource Manager)-资源管理器：

管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

其中，TC 为单独部署的 Server 服务端，TM 和 RM 为嵌入到应用中的 Client 客户端

Seata提供了四种不同的分布式事务解决方案：

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

部署TC服务

1、下载、解压seata

2、修改配置

修改conf目录下的registry.conf文件

3、在nacos添加配置

4、创建数据库表

5、启动TC服务

微服务集成Seata

1、首先，引入seata相关依赖

2、然后，配置application.yml，让微服务通过注册中心找到seata-tc-server

seata:
	registry: 
	# TC服务注册中心的配置，微服务根据这些信息去注册中心获取tc服务地址
	# 参考tc服务自己的registry.conf中的配置，
	#包括:地址、namespace、group、application-name 、cluster
	type: nacos
	nacos: #tc
		server-addr: 127.0.0.1:8848
		namespace: ""
		group: DEFAULT_GROUP
		application: seata-tc-server # tc服务在nacos中的服务名称
	tx-service-group: myseata-demo # 事务组，根据这个获取tc服务的cluster名称
	service:
		vgroup-mapping: # 事务组与TC服务cluster的映射关系
			myseata-demo: SH

总结

nacos服务名称组成包括?

namespace +group+ serviceName + cluster

seata客户端获取tc的cluster名称方式?

以tx-group-service的值为key到vgroupMapping中查找

XA模式

XA规范是 X/0pen 组织定义的分布式事务处理(DTP，Distributed Transaction processing)标准，XA规范描述了全局的TM与局部的RM之间的接口，几乎所有主流的数据库都对XA规范提供了支持。

成功的情况：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

失败的情况：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

seata的XA模式做了一些调整，但大体相似

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

RM一阶段的工作:

注册分支事务到TC
执行分支业务sql但不提交报告执行状态到TC

TC二阶段的工作:

TC检测各分支事务执行状态
- 如果都成功，通知所有RM提交事务
- 如果有失败，通知所有RM回滚事务

RM二阶段的工作:

接收TC指令，提交或回滚事务

XA模式的优缺点

优点：

事务的强一致性，满足ACID原则。
常用数据库都支持，实现简单，并且没有代码侵入

缺点：

因为一阶段需要锁定数据库资源，等待二阶段结束才释放，性能较差
依赖关系型数据库实现事务

实现XA模式

Seata的starter已经完成了XA模式的自动装配，实现非常简单，步骤如下:

1、修改application.yml文件(每个参与事务的微服务)，开启XA模式

seata:
	data-source-proxy-mode: XA # 开启数据源代理的XA模式

2、给发起全局事务的入口方法添加@GlobalTransactional注解，本例中是0rderServicelmpl中的create方法

@0verride
@GlobalTransactional
public Long create(Order order){
    // 创建订单
    orderMapper.insert(order);
    // 扣余额
    // 扣减库存...
    return order.getId();
}

3、重启服务并测试

AT模式

AT模式同样是分阶段提交的事务模型，不过缺弥补了XA模型中资源锁定周期过长的缺陷

核心：分支事务直接提交，快照技术保证回滚

阶段一RM的工作

注册分支事务
记录undo-log(数据快照)
执行业务sql并提交
报告事务状态

阶段二提交时RM的工作

删除undo-log即可

阶段二回滚时RM的工作

根据undo-log恢复数据到更新前

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

值得注意的是AT模式因为一阶段执行事务并提交，所以释放了DB(DataBase)锁，此时其他线程若获取DA锁后修改当前的数据，后续回滚则会丢失本次修改。因此引入了全局锁。

全局锁：由TC记录当前正在操作某行数据的事务该事务持有全局锁，具备执行权。

除此之外，针对非seata管理的事务修改数据产生的影响，我们认为这个情况发生概率较小，但是还是要考虑，于是在产生快照时，设置两份快照，分别是修改前的快照和修改后的，修改前的用于回滚数据，修改后的用于回滚前的比较，若是比较不一致，说明在这个过程中数据被修改过了，不能回滚。

AT模式的优缺点

AT模式的优点:

一阶段完成直接提交事务，释放数据库资源，性能比较好
利用全局锁实现读写隔离
没有代码侵入，框架自动完成回滚和提交

AT模式的缺点:

两阶段之间属于软状态，属于最终一致
框架的快照功能会影响性能，但比XA模式要好很多

实现

AT模式中的快照生成、回滚等动作都是由框架自动完成，没有任何代码侵入，因此实现非常简单

1、导入课前资料提供的Sql文件:seata-at.sql，其中lock_table（全局锁表）导入到TC服务关联的数据库，undo_log（快照表）导入到微服务关联的数据库

2、给发起全局事务的入口方法添加@GlobalTransactional注解

3、修改application.yml文件，将事务模式修改为AT模式即可

seata:
	data-source-proxy-mode: AT #开启数据源代理的AT模式

4、重启服务并测试

TCC模式

追求性能，不加锁。

TCC模式与AT模式非常相似，每阶段都是独立事务，不同的是TCC通过人工编码来实现数据恢复。需要实现三个方法:

Try:资源的检测和预留;
Confirm:完成资源操作业务;要求Try成功 Confirm 一定要能成功。
Cancel:预留资源释放，可以理解为try的反向操作。

举例，一个扣减用户余额的业务。假设账户A原来余额是100，需要余额扣减30元。

阶段一(Try)：检查余额是否充足，如果充足则冻结金额增加30元，可用余额扣除30
阶段二：假如要提交(Confirm)，则冻结金额扣减30
阶段二：如果要回滚(Cancel)则冻结金额扣减30，可用余额增加30

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

TCC的空回滚和业务悬挂

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

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

解决方案：为了实现空回滚、防止业务悬挂，以及幂等性要求。我们必须在数据库记录冻结金额的同时，记录当前事务id和执行状态，为此我们设计了一张表:

CREATE TABLE `account_freeze_tbl`(

    `xid` varchar(128) NOT NULL,
    `user_id` varchar(255) DEFAULT NULL COMMENT '用户id',
    `freeze_money` int(11) unsigned DEFAULT '0' COMMENT '冻结金额',
    `state` int(1) DEFAULT NULL COMMENT'事务状态,0:try,1:confirm,2:cancel'
    PRIMARY KEY(xid)USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=Utf8 ROW_FORMAT=COMPACT;

Try业务:
- 记录冻结金额和事务状态到account_freeze表
- 扣减account表可用金额
Confirm业务
- 根据xid删除account_freeze表的冻结记录
Cancel业务
- 修改account_freeze表，冻结金额为0，state为2
- 修改account表，恢复可用金额
如何判断是否空回滚
- cancel业务中，根据xid查询account_freeze,如果为null则说明try还没做，需要空回滚
如何避免业务悬挂
- try业务中，根据xid查询account_freeze ，如果已经存在则证明Cancel已经执行，拒绝执行try业务

优缺点

优点：

一阶段完成直接提交事务，释放数据库资源，性能好
相比AT模型，无需生成快照，无需使用全局锁，性能最强
不依赖数据库事务，而是依赖补偿操作，可以用于非事务型数据库

缺点：

有代码侵入，需要人为编写try、Confirm和Cancel接口，太麻烦
软状态，事务是最终一致
需要考虑Confirm和Cancel的失败情况，做好幂等处理

实现

TCC的Try、Confirm、Cancel方法都需要在接口中基于注解来声明，语法如下

@LocalTCC
public interface TCCService {
    /**
    Try逻辑,@TwoPhaseBusinessAction中的name属性要与当前方法名一致，用于指定Try逻辑对应的方法
    */
    @TwoPhaseBusinessAction(name = "prepare", commitMethod = "confirm", rollbackMethod = "cancel")
    void prepare(@BusinessActionContextParameter(paramName = "param")String param);
    /**
    *二二阶段confirm确认方法、可以另命名，但要保证与commitMethod一致
    *@param context 上下文，可以传递try方法的参数
    *@return boolean 执行是否成功
    */
    boolean confirm(BusinessActionContext context);
    /**
	*二阶段回滚方法，要保证与rollbackMethod一致
	*/
	boolean 		cancel(BusinessActionContext context);
}

实现类的写法参考

@Override
@Transactional
public void deduct(string userId, int money)
{
    // 0.获取事务id
	String xid=RootContext.getXID();
    // 0.1.判断freeze中是否有冻结记录，如果有，一定是CANCEL执行过，我要拒绝业务
    AccountFreeze oldFreeze =freezeMapper.selectById(xid);
    if(oldFreeze != null){
        //CACEL执行过，我要拒绝业务
        return;
    }
    
    // 1.扣减可用余额
accountMapper.deduct(userId, money);
    // 2.记录冻结金额，事务状态
    AccountFreeze freeze=new AccountFreezefreeze.setUserId(userId);
    freeze.setFreezeMoney(money);
    freeze.setState(AccountFreeze.State.TRY);
    freeze.setXid(xid);
freezeMapper.insert(freeze);
}

@Override
public boolean confirm(BusinessActioncontext ctx){
    //1.获取事务id
    String xid = ctx.getXid();
    //2.根据id删除冻结记录
    int count = freezeMapper.deleteById(xid);
    return count == 1;
}

@Override
public boolean cancel(BusinessActioncontext ctx){
    // 0.查询冻结记录
    String xid =ctx.getXid();
    AccountFreeze freeze = freezeMapper.selectById(xid);
    //1.空回滚的判断，判freeze是否为null，为null证明try没执行，需要回滚
    if(freeze == null){
        // 证明try没执行，需要空回滚
        freeze = new AccountFreeze();
        freeze.setUserId(userId);
        freeze.setFreezeMoney(0);
        freeze.setState(AccountFreeze.State.CANCEL);
        freeze.setXid(xid);
        freezeMapper.insert(freeze);
        return true;
    }
    // 2.幂等判断
    if(freeze.getState() == AccountFreeze.State.CANCEL){
        //已经处理过一次CANCEL了，无需重复处理
        return true;
    }
    
    //3.恢复可用余额
    accountMapper.refund(freeze.getUserId(),freeze.getFreezeMoney());
    //4.将冻结金额清等，状态改为CANCEL
    freeze.setFreezeMoney(0);
    freeze.setState(AccountFreeze.State.CANCEL);
    int count=freezeMapper.updateById(freeze);
    return count == 1;
}

SAGA模式

Saga模式是SEATA提供的长事务解决方案。也分为两个阶段:

一阶段：直接提交本地事务
二阶段：成功则什么都不做；失败则通过编写补偿业务来回滚

事务之间没有隔离性，可能出现脏写的问题。

优缺点

Saga模式优点：

事务参与者可以基于事件驱动实现异步调用，吞吐高
一阶段直接提交事务，无锁，性能好
不用编写TCC中的三个阶段，实现简单

缺点:

软状态持续时间不确定，时效性差
没有锁，没有事务隔离，会有脏写

四种模式对比

	XA	AT	TCC	SAGA
一致性	强一致	弱一致	弱一致	最终一致
隔离性	完全隔离	基于全局锁隔离	基于资源预留隔离	无隔离
代码侵入	无	无	有,要编写三个接口	有,要编写状态机和补偿业务
性能	差	好	非常好	非常好
场景	对一致性、隔离性有高要求的业务	基于关系型数据库的大多数分布式事务场景都可以	1、对性能要求较高的事务。 2、有非关系型数据库要参与的事务。	1、业务流程长、业务流程多 2、参与者包含其它公司或遗留系统服务,无法提供TCC 模式要求的三个接

高可用

高可用集群结构

TC服务作为Seata的核心服务，一定要保证高可用和异地容灾，需要分布式搭载多个集群提供服务。

采用Nacos配置中心配置，将事务组映射到Nacos，微服务读取Nacos配置。

具体查询其他文档。

分布式缓存

单节点Redis的问题与解决

1、数据丢失问题

实现Redis数据持久化

2、并发能力问题

搭建主从集群，实现读写分离

3、故障恢复问题

利用Redis哨兵，实现健康检测和自动恢复

4、存储能力问题

搭建分片集群，利用插槽机制实现动
态扩容

数据持久化

RDB

RDB全称Redis Database Backup file(Redis数据备份文件)，也被叫做Redis数据快照。简单来说就是把内存中的所有数据都记录到磁盘中。当Redis实例故障重启后，从磁盘读取快照文件，恢复数据。

快照文件称为RDB文件，默认是保存在当前运行目录。

redis-cli
/
save # Redis主进程执行RDB，会阻塞所有命令
/
bgsave # 后台保存，开启子进程执行RDB，避免主进程受影响

默认的情况下，停机就会执行持久化。

Redis内部有触发RDB的机制，可以在redis.conf文件中找到，格式如下

# 900秒内，如果至少有1个key被修改，则执行bgsave，如果是save ""则表示禁用RDB
save 900 1
save 300 10
save 60 10000

RDB的其它配置也可以在redis.conf文件中设置:

# 是否压缩,建议不开启，压缩也会消耗cpu，磁盘的话不值钱
rdbcompression yes
# RDB文件名称
dbfilename dump.rdb
# 文件保存的路径目录
dir ./

RDB原理

RDB方式bgsave的基本流程如下：

1、bqsave开始时会fork主进程得到子进程，子进程共享主进程的内存数据
2、子进程完成fork后读取内存数据并写入 RDB 文件。
3、用新RDB文件替换旧的RDB文件。

fork采用的是copy-on-write技术:

当主进程执行读操作时，访问共享内存;
当主进程执行写操作时，则会拷贝一份数据，执行写操作

RDB的缺点：

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

AOF持久化

AOF全称为Append OnlyFile(追加文件)。Redis处理的每一个写命令都会记录在A0F文件，可以看做是命令日志文件。

AOF默认是关闭的，需要修改redis.conf配置文件来开启AOF

#是否开启AOF功能，默认是no
appendonly yes
# AOF文件的名称
appendfilename "appendonly.aof"

AOF的命令记录的频率也可以通过redis.conf文件来配：

# 表示每执行一次写命令立即记录到AOF文件
appendfsync always

# 写命令执行完先放入AOF缓冲区，然后表示每隔1秒将缓冲区数据写到A0F文件，是默认方案
appendfsync everysec

# 写命令执行完先放入AOF缓冲区，由操作系统决定何时将缓冲区内容写回磁盘
appendfsync no

配置项	刷盘时机	优点	缺点
Always	同步刷盘	可靠性高,几乎不丢数据	性能影响大
everysec	每秒刷盘	性能适中	最多丢失1秒数据
no	操作系统控制	性能最好	可靠性较差,可能丢失大量数据

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

Redis也会在触发阈值时自动去重写AOF文件。值也可以在redis.conf中配置:

# AOF文件比上次文件 增长超过多少百分比则触发重写
auto-aof-rewrite-percentage 100
# AOF文件体积最小多大以上才触发重写
auto-aof-rewrite-min-size 64mb

AOF和RDB的优缺点

RDB和AOF各有自己的优缺点，如果对数据安全性要求较高，在实际开发中往往会结合两者来使用

	RDB	AOF
持久化方式	定时对整个内存做快照	记录每一次执行的命令
数据完整性	不完整,两次备份之间会丢失	相对完整,取决于刷盘策略
文件大小	会有压缩,文件体积小	记录命令,文件体积很大
宕机恢复速度	很快	慢
数据恢复优先级	低,因为数据完整性不如AOF	高,因为数据完整性更高
系统资源占用	高,大量CPU和内存消耗	低,主要是磁盘I0资源但AOF重写时会占用大量CPU和内存资源
使用场景	可以容忍数分钟的数据丢失,追求更快的启动速度	对数据安全性要求较高常见

主从复制

搭建主从架构

单节点Redis的并发能力是有上限的，要进一步提高Redis的并发能力，就需要搭建主从集群，实现读写分离。

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

具体步骤可以分为，

1、准备多个实例

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

2、启动

3、开启主从关系

可以通过修改配置文件，或者slaveof命令

详细步骤查询其他文档(Redis集群文档)

主从数据同步原理

全量同步

主从第一次同步是全量同步:

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

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

master如何判断slave是不是第一次来同步数据?

这里会用到两个很重要的概念:

Replicationld：简称replid，是数据集的标记，id一致则说明是同一数据集。每一个master都有唯一的replid,slave则会继承master节点的replid
offset：偏移量，随着记录在repl baklog中的数据增多而逐渐增大。slave完成同步时也会记录当前同步的offset。
如果slave的offset小于master的offset，说明slave数据落后于master，需要更新。

因此slave做数据同步，必须向master声明自己的replicationid 和offset，master才可以判断到底需要同步哪些数据

第一次的时候master判断replid不一致，返回主节点replid和offset。

增量同步

主从第一次同步是全量同步，但如果slave重启后同步，则执行增量同步

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

注意！

repl_baklog大小有上限，写满后会覆盖最早的数据。如果slave断开时间过久，导致尚未备份的数据被覆盖，则无法基于log做增量同步，只能再次全量同步。

可以从以下几个方面来优化Redis主从就集群:

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

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

总结

简述全量同步和增量同步区别?

全量同步:master将完整内存数据生成RDB，发送RDB到slave。后续命令则记录在repl baklog，逐个发送给slave。
增量同步:slave提交自己的offset到master，master获取repl_baklog中从offset之后的命令给slave

什么时候执行全量同步?

slave节点第一次连接master节点时
slave节点断开时间太久，repl_baklog中的offset已经被覆盖时

什么时候执行增量同步?

slave节点断开又恢复，并且在repl_baklog中能找到offset时

Redis哨兵

哨兵的作用和原理

作用

Redis提供了哨兵(Sentinel)机制来实现主从集群的自动故障恢复。哨兵的结构和作用如下

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

服务状态监控（判断实例是否健康）

Sentinel基于心跳机制监测服务状态，每隔1秒向集群的每个实例发送ping命令:

主观下线:如果某sentinel节点发现某实例未在规定时间响应，则认为该实例主观下线。
客观下线:若超过指定数量(quorum)的sentinel都认为该实例主观下线，则该实例客观下线。quorum值最好超过Sentinel实例数量的一半。

选举新的master

一旦发现master故障，sentinel需要在salve中选择一个作为新的master，选择依据是这样的

首先会判断slave节点与master节点断开时间长短，如果超过指定值(down-after-miliseconds*10)则会排除该slave节点
然后判断slave节点的slave-priority值，越小优先级越高，如果是0则永不参与选举
如果slave-prority一样，则判断slave节点的offset值，越大说明数据越新，优先级越高
最后是判断slave节点的运行id大小，越小优先级越高。

如何实现故障转移

当选中了其中一个slave为新的master后(例如slave1)，故障的转移的步骤如下:

sentinel给备选的slave1节点发送slaveofnoone命令，让该节点成为master
sentinel给所有其它slave发送slaveof192.168.150.1017002命令，让这些slave成为新master的从节点，开始从新的master上同步数据。
最后，sentinel将故障节点标记为slave，当故障节点恢复后会自动成为新的master的slave节点

搭建哨兵集群

在准备集群和配置后启动、运行、测试。

具体参考Redis集群文档。

RedisTemplate的哨兵模式

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

1、在pom文件中引入redis的starter依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2、然后在配置文件application.yml中指定sentinel相关信息

spring:
	redis:
		sentinel:
			master: mymaster # 指定master名称
			nodes: # 指定redis-sentinel集群信息
				- 192.168.150.101:27001
				- 192.168.150.101:27002
				- 192.168.150.101:27003

3、配置主从读写分离

@Bean
public LettuceclientConfigurationBuilderCustomizer configurationBuilderCustomizer(){
    return configBuilder ->configBuilder.readFrom(ReadFrOM.REPLICA PREFERRED);
}

这里的ReadFrom是配置Redis的读取策略，是一个枚举，包括下面选择:

MASTER：从主节点读取
MASTER PREFERRED：优先从master节点读取，master不可用才读取replica
REPLICA：从slave(replica)节点读取
REPLICA_PREFERRED：优先从slave(replica)节点读取，所有的slave都不可用才读取master

Redis分片集群

分片集群结构

主从和哨兵可以解决高可用、高并发读的问题。但是依然有两个问题没有解决

海量数据存储问题
高并发写的问题

使用分片集群可以解决上述问题，分片集群特征

集群中有多个master，每个master保存不同数据
每个master都可以有多个slave节点
master之间通过ping监测彼此健康状态
客户端请求可以访问集群任意节点，最终都会被转发到正确节点

搭建参考Redis集群文档。

散列插槽

Redis会把每一个master节点映射到0~16383共16384个插槽(hash slot)上，查看集群信息时就能看到

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

数据key不是与节点绑定，而是与插槽绑定。redis会根据key的有效部分计算插槽值，分两种情况:

key中包含"{}"，且“{}”中至少包含1个字符，“{}”中的部分是有效部分
key中不包含“{}”，整个key都是有效部分

例如：key是num，那么就根据num计算，如果是{itcast}num，则根据itcast计算。计算方式是利用CRC16算法得到一个hash值，然后对16384取余，得到的结果就是slot值。

Redis如何判断某个key应该在哪个实例?

将16384个插槽分配到不同的实例
根据key的有效部分计算哈希值，对16384取余
余数作为插槽，寻找插槽所在实例即可

如何将同一类数据固定的保存在同一个Redis实例?

这一类数据使用相同的有效部分，例如key都以{typeld}为前缀

集群伸缩

redis-cli --cluster提供了很多操作集群的命令，可以通过下面方式查看

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

比如，添加节点的命令:

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

需要老节点用于联系上集群，从而将新节点通知给集群中的每一个节点。默认新增节点为master节点，加上可选参数为从节点。

转移插槽

新创建的节点的插槽数量为0，因此没有任何数据可以存储到新节点上

我们可以将0~3000的插槽从7001转移到7004，命令格式如下：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

复制接收插槽的节点的id和给出插槽的节点的id，作为参数

故障转移

自动故障转移

当集群中有一个master宕机会发生什么呢？

直接停止一个redis实例，例如7002：

1）首先是该实例与其它实例失去连接
2）然后是疑似宕机：
3）最后是确定下线，自动提升一个slave为新的master：
4）当7002再次启动，就会变为一个slave节点了：

手动故障转移

利用cluster failover命令可以手动让集群中的某个master宕机，切换到执行cluster failover命令的这个slave节点，实现无感知的数据迁移。其流程如下：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

这种failover命令可以指定三种模式：

缺省：默认的流程，如图1~6歩
force：省略了对offset的一致性校验
takeover：直接执行第5歩，忽略数据一致性、忽略master状态和其它master的意见

RedisTemplate访问分片集群

RedisTemplate底层同样基于lettuce实现了分片集群的支持，而使用的步骤与哨兵模式基本一致：

1）引入redis的starter依赖
2）配置分片集群地址
3）配置读写分离

与哨兵模式相比，其中只有分片集群的配置方式略有差异，如下：

spring:
  redis:
    cluster:
      nodes:
        - 192.168.150.101:7001
        - 192.168.150.101:7002
        - 192.168.150.101:7003
        - 192.168.150.101:8001
        - 192.168.150.101:8002
        - 192.168.150.101:8003

多级缓存

多级缓存介绍

传统缓存的问题

传统缓存策略一般是请求到tomcat后，先查询redis，如果未命中则查询数据库

请求要先经过tomcat处理，tomcat的性能成为整个系统的瓶颈

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

多级缓存方案

多级缓存就是利用请求处理的每个环节，分别添加缓存，减轻tomcat压力，提升服务器性能：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

JVM进程缓存

实际上相当于实现的是上述图片中的Tomcat进程缓存

初始Caffeine

缓存分类

我们把缓存分为两类：分布式缓存和进程本地缓存

分布式缓存，例如Redis：

优点：缓存容量更大，可靠性更好，可以在集群间共享
缺点：访问缓存网络开销大
场景：缓存数据量较大、可靠性要求较高、需要在集群间共享

进程本地缓存，例如hashMap、GuavaCache

优点：读取本地内存，没有网络开销，速度快
缺点：存储量有限，可靠性低，无法共享
场景：性能要求较高，缓存数据量较小

Caffeine入门

Cafeine是一个基于Java8开发的，提供了近乎最佳命中率的高性能的本地缓存库。目前Spring内部的缓存使用的就是Caffeine。

可以通过item-service项目中的单元测试来学习Caffeine的使用

@Test
void testBasicOps(){
    //创建缓存对象
    Cache<String,string> cache = Caffeine.newBuilder().build();
    // 存数据
    cache.put("gf"，"迪丽热巴");
    //取数据，不存在则返回null
    String gf = cache.getIfPresent("gf");
    System.out.println("gf=" + gf);
    //取数据，不存在则去数据库查询
    String defaultGF =cache.get("defaultGr",key ->{
        //这里可以去数据库根据 key查询value
        return "柳岩";
});
    System.out.println("defaultGF ="+ defaultGF);