springcloud
一、概述
1.1. 微服务技术对比
1.1.1 SpringCloud + Feign
- 使用SpringCloud技术栈
- 服务接口采用Restful风格
- 服务调用采用Feign方式
1.1.2 SpringCloudAlibaba + Feign
- 使用SpringCloudAlibaba技术栈
- 服务接口采用Restful风格
- 服务调用采用Feign方式
1.1.3 SpringCloudAlibaba + Dubbo
- 使用SpringCloudAlibaba技术栈
- 服务接口采用Dubbo协议标准
- 服务调用采用Dubbo方式
1.1.4 Dubbo原始模式
- 局域Dubbo老旧技术体系
- 服务接口采用Dubbo协议标准
- 服务调用采用Dubbo方式
- 可升级为SpringCloudAlibaba + Dubbo
注册中心：Eureka、Nacos，内部都由Ribbon作负载均衡
1.2. 微服务框架
- 微服务治理：二~七
- Docker：八
- 异步通信：九
- 分布式搜索：十
- 微服务保护：十一
- 分布式事务：十二
- 多级缓存：十三
- 分布式缓存：十四
- 可靠消息服务：十五
二、远程调用
2.1. 服务拆分
- 根据业务模块拆分，做到单依职责，不重复开发相同业务
- 可以将业务暴露为借口，供其他微服务使用
- 不同为服务都应该有自己独立的数据库
2.2. RestTemplate
- 由于微服务拆分，各项功能对外暴露REST的api接口，使用RestTemplate进行对REST接口的请求
- 将RestTemplate注册到Spring容器中
@MapperScan("cn.itcast.order.mapper") |
2.3. 测试接口
接口地址1：http://localhost:8081/user/1
返回对象：
{ |
接口地址2：http://localhost:8081/user/1
返回对象：
{ |
2.4. 远程调用应用
①注入RestTemplate
|
②增加业务代码
@Service |
③实现效果
{ |
2.5. 服务调用关系
- Provider服务提供者：暴露接口给其他微服务调用
- Consumer服务消费者：调用其他微服务提供的接口
- 提供者与消费者角色其实是相对的
- 一个服务可以同时是服务提供者和服务消费者
三、Eureka注册中心
- 服务调用出现的问题：硬编码 “http://localhost:8081/user/“ + order.getUserId()
- 负载均衡时
- 服务消费者如何获取服务提供者的地址信息
- 多个服务提供者，消费者怎样选择
- 消费者如何得知服务提供者的健康状态
3.1. 作用
Eureka分为两个模块：
eureka-server注册中心
保存eureka-client发送过来的信息，以供服务消费者调用
例如：
user-service:
localhost:8081
localhost:8082
order-service:
localhost:8080
eureka-client
- 服务消费者和服务提供者都在client中
- 服务消费者和多个服务提供者的信息被注册到eureka-server中
- 每隔30秒eureka-client会对eureka-server进行一次心跳续约，保证服务提供者没有宕掉
执行顺序：
- 注册服务信息：eureka-client中的服务消费者和服务提供者向eureka-server注册中心发送信息
- 拉取服务【定时拉取服务pull 30s/次】：服务消费者去拉取eureka-server注册中心中的服务提供者信息
- 负载均衡：服务消费者去选择调用服务提供者的服务器
- 远程调用
- 心跳续约：eureka-client每过30秒刷新一次
- 消费者获取提供者具体信息
- 服务提供者启动时向eureka注册自己的信息
- eureka保存这些信息
- 消费者根据服务名称向eureka拉取提供者信息
- 多个服务提供者时消费者的选择
- 服务消费者利用负载均衡算法从服务列表中挑选一个
- 消费者感知服务提供者健康状态
- 服务提供者每隔30秒向EurekaServer发送心跳请求，报告健康状态
- eureka会更新记录服务列表信息，心跳不正常被剔除
- 消费者可以拉取到最新的信息
3.2. 搭建EurekaServer服务
创建eureka-server项目并引入依赖
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-server</artifactId>
</dependency>编写启动类，添加@EnableEurekaServer注解
public class EurekaApplication {
public static void main(String[] args) {
SpringApplication.run(EurekaApplication.class, args);
}
}添加applicaiton.yml文件编写配置
server:
port: 10086 # 服务端口
spring:
application:
name: eurekaserver # eureka服务名称
eureka:
client:
service-url: # eureka地址信息：将自己也注册进eureka,后续多个eureka,集群使用
defaultZone: http://127.0.0.1:10086/eureka
3.3. eureka服务注册
服务提供者模块添加依赖
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>服务提供者模块添加yml配置eureka地址
spring:
application:
name: userservice # eureka服务名称
eureka:
client:
service-url: # eureka地址信息：将自己也注册进eureka,后续多个eureka,集群使用
defaultZone: http://127.0.0.1:10086/eureka让某一服务多次启动
- 在IDEA的Service中，右键已经启动的服务
- 选择Copy Configuration… 或Ctrl+D
- 修改VMoptions: -Dserver.port=8082
- 如果找不到VMoptions，则选择Modify options选择Add VM Options
- 效果：** UP** (2) - 192.168.209.1:userservice:8082 , 192.168.209.1:userservice:8081
3.4. 服务拉取
修改Service代码，修改访问的url路径，用服务名代替ip、端口
public class OrderService {
private OrderMapper orderMapper;
private RestTemplate restTemplate;
public Order queryOrderById(Long orderId) {
// 1.查询订单
Order order = orderMapper.findById(orderId);
// 2.利用RestTemplate发起http请求，查询用户
// 2.1url路径
String url = "http://userservice/user/" + order.getUserId();
// 2.2发哦是哪个http请求，实现远程调用
User user = restTemplate.getForObject(url, User.class);
// 3.封装User到Order
order.setUser(user);
// 4.返回
return order;
}
}在项目启动类注册的RestTemplate添加负载均衡注解：@LoadBalanced
public class OrderApplication {
public static void main(String[] args) {
SpringApplication.run(OrderApplication.class, args);
}
/**
* 创建RestTemplate并注入到Spring容器
* @return
*/
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
四、Ribbon负载均衡
4.1. 负载均衡流程
- 服务消费者order-service发起请求
- http://userservice/user/1
- 此地址不是一个可以访问的地址
- Ribbon负载均衡作用
- 接收服务消费者发来的请求地址http://userservice/user/1
- 拉取eureka-server中的userservice
- eureka-server给Ribbon返回服务列表
- localhost:8081
- localhost:8082
- Ribbon对服务器进行负载均衡
- 轮询到8081实现访问
- 轮询操作：一个服务器访问完换下一个服务器
- @LoadBalanced原理
- 被LoadBalancerInterceptor.java拦截
4.2. 负载均衡策略定义
4.2.1 代码方式
启动类中添加IRule的实现类为@Bean
|
4.2.2 配置文件方式
yml配置
userservice: # Eureka服务名 |
4.3. Ribbon饥饿加载
Ribbon默认采用懒加载，即第一次访问时创建LoadBalanceClient，请求时间会很长
饥饿加载会在项目启动时创建，降低第一次访问的耗时
yml配置
ribbon: |
五、Nacos注册中心
认识Nacos：Nacos是阿里巴巴的产品，现在是SpringCloud中的一个组件，相比于Eureka功能更加丰富
5.1. Nacos安装指南
登录默认账号密码都是nacos
5.1.1 Windows安装
开发阶段采用单机安装即可。
①下载安装包
在Nacos的GitHub页面，提供有下载链接，可以下载编译好的Nacos服务端或者源代码：
GitHub主页：https://github.com/alibaba/nacos
GitHub的Release下载页：https://github.com/alibaba/nacos/releases
②解压
目录说明：
- bin：启动脚本
- conf：配置文件
③端口配置
Nacos的默认端口是8848，如果你电脑上的其它进程占用了8848端口，请先尝试关闭该进程。
如果无法关闭占用8848端口的进程，也可以进入nacos的conf目录，修改配置文件中的端口：
application.properties
④启动
启动非常简单，进入bin目录，然后执行命令即可：
windows命令：
startup.cmd -m standalone
⑤访问
在浏览器输入地址：http://127.0.0.1:8848/nacos即可
5.1.2 Linux安装
Linux或者Mac安装方式与Windows类似。
①安装JDK
Nacos依赖于JDK运行，索引Linux上也需要安装JDK才行。
上传jdk安装包：
上传jdk-8u144-linux-x64.tar.gz到某个目录，例如：/usr/local/
然后解压缩：
tar -xvf jdk-8u144-linux-x64.tar.gz |
然后重命名为java
配置环境变量：
export JAVA_HOME=/usr/local/java |
设置环境变量：
source /etc/profile |
②上传安装包
上传nacos-server-1.4.1.tar.gz到Linux服务器的某个目录，例如/usr/local/src目录下
③解压
命令解压缩安装包：
tar -xvf nacos-server-1.4.1.tar.gz |
然后删除安装包：
rm -rf nacos-server-1.4.1.tar.gz |
④端口配置
与windows中类似
⑤启动
在nacos/bin目录中，输入命令启动Nacos：
sh startup.sh -m standalone |
5.1.3 Nacos的依赖
父工程：
<dependency> |
客户端：
<!-- nacos客户端依赖包 --> |
5.2. 使用Nacos
①父工程引入依赖
这个依赖声明会将 spring-cloud-alibaba-dependencies 添加到项目的依赖管理器中，从而可以方便地管理 Spring Cloud Alibaba 组件的版本。
<dependencyManagement> |
②所有的Provider和Consumer引入依赖
<!-- nacos客户端依赖包 --> |
③配置application.yml
spring: |
5.3. nacos服务分级存储模型
5.3.1 集群概念
①一级是服务，例如userservice
②二级是集群，例如杭州或上海
③三级是服务实例，例如杭州机房的某台部署了userservice的服务器
- nacos
- HZ集群
- userservice1
- userservice2
- SH集群
- userservice1
- userservice2
- HZ集群
5.3.2 配置集群
# Consumer和Provider服务都进行设置集群 |
注：Consumer会优先调用相同集群区域的Provider
5.3.3 配置集群负载均衡
# Consumer消费者服务设置 |
5.3.4 跨集群查询
当前Consumer调用的Provider无法在相同集群中找到时，会去查找其他集群，同时在Consumer日志中报如下警告
07-17 22:57:28:729 WARN 16336 --- [nio-8080-exec-5] c.alibaba.cloud.nacos.ribbon.NacosRule : A cross-cluster call occurs，name = userservice, clusterName = HZ, instance = [Instance{instanceId='192.168.209.1#8083#SH#DEFAULT_GROUP@@userservice', ip='192.168.209.1', port=8083, weight=1.0, healthy=true, enabled=true, ephemeral=true, clusterName='SH', serviceName='DEFAULT_GROUP@@userservice', metadata={preserved.register.source=SPRING_CLOUD}}] |
5.4. 根据权重负载均衡
- 服务器设备性能有所差异，部分实例所在机器性能好一些，另一些较差
- Nacos提供了权重配置来控制访问频率，权重越大访问频率越高
- 权重值：[0,1]
- 权重为0时，不会访问此服务
设置权重
- 打开Nacos本地服务的网站Console: http://192.168.209.1:8848/nacos/index.html
- 服务管理->服务列表->设置权重
5.5. 环境隔离-namespace
Nacos中服务存储和数据存储的最外层都是一个名为namespace的东西，用来做最外层隔离
如果想让服务之间可以相互调用，放到同一个环境下，否则访问不到各自的环境
- Namespace
- Group
- Service/Data
- Group
创建环境隔离[命名空间] - namespace
Nacos本地服务地址中配置：http://192.168.209.1:8848/nacos/index.html
找到命名空间
新建命名空间，设置命名空间和描述
拿到自动生成的命名空间id
在application.yml中进行配置命名空间
spring:
application:
name: orderservice
cloud:
nacos:
server-addr: localhost:8848 #nacos服务端地址
discovery:
cluster-name: HZ # 集群配置
# 命名空间配置
namespace: bfeb4944-eeb0-4f6e-952a-f290298c9654 # dev环境
总结
- namespace用来做环境隔离
- 每个namespace都有唯一Id
- 不同namespace下的服务不可见
5.6. Nacos注册中心细节
- Eureka对Consumer服务消费者只有定时拉取服务pull的功能
- Eureka对Provider服务提供者只有心跳监测
- Eureka集群采用AP方式
5.6.1 服务消费者Consumer
- Consumer服务消费者定时拉取服务pull,30s/次
- nacos主动推送变更消息push
5.6.2 服务提供者Provider
- 临时实例：采用与Eureka相同的心跳监测
- 非临时实例：Nacos主动询问，当此实例挂掉后，Nacos并不会从实例中剔除
5.6.3 Nacos集群
- Nacos集群默认采用AP方式，当集群中存在非临时实例时，采用CP模式
配置非临时实例
spring: |
5.7. Nacos配置管理
5.7.1 统一配置管理
启动Nacos进入管理页面
选择配置管理->配置列表->命名空间->新建
设置Data ID : userservice-dev.yaml
设置Group : 默认即可
添加描述
选择配置格式： yaml和properties
填写配置内容，内容为核心或后续具备变化的配置
pattern:
dateformat: yyyy-MM-dd HH:mm:ss
5.7.2 配置获取
源springboot项目：
- 项目启动->读取本地配置文件application.yml->创建Spring容器->加载bean
Nacos配置后的springboot项目：
- 项目启动->读取Nacos的配置文件->读取本地配置文件application.yml->创建Spring容器->加载bean
①引入Nacos配置管理客户端依赖
<dependency> |
②userservice中resource目录添加一个bootstrap.yml文件，这个文件是引导文件，优先级高于application.yml
spring: |
注：
- 删除application.yml中的重复配置，如application.name，cloud.nacos
- 配置的命名空间一定要在同一个，否则会启动服务失败，因为找不到这个配置属性
③Controller层注入在Nacos中的配置进行测试
此配置在7.1可查看
|
5.7.3 配置刷新
Nacos配置更改后，微服务可以实现热更新，方式：
- ① 通过@Value注解注入，结合@RefreshScope来刷新
- ② 通过@ConfigurationProperties注入，自动刷新
注意事项：
- 不是所有的配置都适合放到配置中心，维护麻烦
- 建议将一些关键参数，需要运行时调整的参数放到nacos配置中心，一般都是自定义配置
①方式一
在@Value注入的变量所在类上添加注解@RefreshScope
|
②方式二
使用@ConfigurationProperties注解
添加配置类
// com.bamboo.user.config.PatternProperties
public class PatternProperties {
private String dateformat;
}添加spring-boot-configuration-processor注解驱动
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>使用@Autowired进行注入
// nacos配置热部署
//@RefreshScope
public class UserController {
private UserService userService;
// 注入热部署配置
private PatternProperties properties;
public String now() {
return LocalDateTime.now().format(DateTimeFormatter.ofPattern(properties.getDateformat()));
}
}
5.8. Nacos多环境配置共享
微服务启动时会从nacos读取多个配置文件：
- [spring.application.name]-[spring.profiles.active].yaml，例如：userservice-dev.yaml
- [spring.application.name].yaml，例如：userservice.yaml
无论profile如何变化，[spring.application.name].yaml这个文件一定会加载，因此多环境共享配置可以写入这个文件
文件配置优先级：Nacos配置中的application-dev.yaml > Nacos配置中的application.yaml > 本地application.yaml配置
- 即 服务名-profile.yaml > 服务名.yaml > 本地配置
使用
Nacos服务上添加配置列表
新建配置
设置Data ID：userservice.yaml
添加描述：环境共享
设置配置格式：YAML
设置配置内容：
pattern:
envSharedValue: 环境共享属性值
修改读取配置类
public class PatternProperties {
private String dateformat;
private String envSharedValue;
}自动注入并调用接口
private PatternProperties properties;
public PatternProperties loadProp() {
return properties;
}测试
- 测试结果通过，不同profiles环境下只有多环境配置userservice.yaml可以读取到
- 测试多环境方式：IDEA右键Edit Configuration，选择Active profiles填写test，说明这是一个userservice-test.yaml配置的SpringBoot
5.9. Nacos集群搭建
搭建级别：
- nacos client
- Nginx
- Nacos node1
- Nacos node2
- Nacos node3
- MySQL主
- MySQL从
- MySQL从
- MySQL主
- Nginx
Nacos node统一接入MySQL
5.9.1 搭建数据库和Nacos
①初始化数据库
Nacos默认数据存储在内嵌数据库Derby中，不属于生产可用的数据库。
官方推荐的最佳实践是使用带有主从的高可用数据库集群。
这里我们以单点的数据库为例来讲解。
首先新建一个数据库，命名为nacos，而后导入下面的SQL：
CREATE TABLE `config_info` ( |
②下载nacos
nacos在GitHub上有下载地址：https://github.com/alibaba/nacos/tags，可以选择任意版本下载。
③配置nacos
目录说明：
- bin：启动脚本
- conf：配置文件
进入nacos的conf目录，修改配置文件cluster.conf.example，重命名为cluster.conf
然后添加内容：
127.0.0.1:8845 |
然后修改application.properties文件，添加数据库配置
spring.datasource.platform=mysql |
④启动
将nacos文件夹复制三份，分别命名为：nacos1、nacos2、nacos3
然后分别修改三个文件夹中的application.properties，
nacos1:
server.port=8845 |
nacos2:
server.port=8846 |
nacos3:
server.port=8847 |
然后分别启动三个nacos节点：
startup.cmd |
5.9.2 配置nginx反向代理
修改conf/nginx.conf文件，配置如下：
upstream nacos-cluster { |
而后在浏览器访问：http://localhost/nacos即可。
代码中application.yml文件配置如下：
spring: |
六、http客户端Feign
RestTemplate方式调用存在的弊端
源代码：
String url = "http://userservice/user/" + order.getUserId(); |
- 代码可读性差，编程体验不统一
- 参数复杂的URL难以维护
6.1. Feign概述
声明式的http客户端，作用是帮助我们优雅的实现http请求的发送
6.2. 使用
服务消费者Consumer引入依赖
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>在服务消费者Consumer的启动类中加入注解开启Feign功能
public class OrderApplication {
public static void main(String[] args) {
SpringApplication.run(OrderApplication.class, args);
}
}编写Feign客户端
public interface UserClient {
User findById( Long id);
}主要基于SpringMVC注解来声明远程调用的信息：
- 服务名称：userservice
- 请求方式：GET
- 请求路径：/user/{id}
- 请求参数：Long id
- 返回值类型：User
调用
public class OrderService {
private OrderMapper orderMapper;
private RestTemplate restTemplate;
// 注入Feign
private UserClient userClient;
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;
}
}使用总结
- 引入依赖
- 添加@EnableFeignClients注解
- 编写FeignClient接口
- 使用FeignClient中定义的方法代替RestTemplate
6.3. 自定义配置
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.xxx.loggerLevel
如果是xxx是default则代表全局
# 配置feign
feign:
client:
config:
default: # 全局配置
logger-level: full # 日志级别如果xxx是服务名称，例如userservice则代表某服务
# 配置feign
feign:
client:
config:
userservice: # 局部日志
logger-level: full # 日志级别
方式二：Java代码方式，配置Logger.Level的Bean
定义Bean
public class FeignClientConfiguration {
public Logger.Level feignLogLevel() {
return Logger.Level.BASIC;
}
}如果在@EnableFeignClients注解声明则代表全局
public class OrderApplication {
public static void main(String[] args) {
SpringApplication.run(OrderApplication.class, args);
}
}如果在@FeignClient注解中声明则代表某服务
public interface UserClient {
User findById( Long id);
}
6.4. 性能优化
Feign底层he护短实现：
- URLConnection：默认实现，不支持连接池
- Apache HttpClient：支持链接吃
- OKHttp：支持连接池
优化Feign性能主要包括：
① 使用连接池代替默认的URLConnection
② 日志级别，最好使用basic或none
Feign添加HttpClient支持：
引入依赖：
<dependency>
<groupId>io.github.openfeign</groupId>
<artifactId>feign-httpclient</artifactId>
</dependency>配置连接池
feign:
# client: # 已使用java代码对日志进行了配置
# config:
# default: # 全局配置
# logger-level: full # 日志级别
httpclient:
enabled: true # 开启feign对HttpClient的支持
max-connections: 200 # 最大连接数
max-connections-per-route: 50 # 每个路径最大的连接数
6.5. 最佳实践
- 让controller和FeignClient继承同一接口
- 让FeignClient、POJO、Feign的默认配置豆丁一道一个项目中，供所有消费者使用
①方式一(继承)
给消费者的FeignClient和提供者的controller定义统一的父接口作为标准。
父接口
public interface USerAPI{
User findById( Long id);
}FeignClient
public interface UserClient extends UserAPI{}服务提供者Controller
public class UserController implements UserAPI{
public User findById( Long id){
// ...实现业务
}
}
*通常不建议在服务器和客户机之间共享一个接口。它引入了紧密耦合，并且实际上也不能与当前形式的Spring MVC一起工作(方法参数映射不能继承)。
② 方式二(抽取)
将FeignClient抽取为独立模块，并且把接口有关的POJO、默认的Feign配置都放到这个模块中，提供给所有的消费者使用
创建一个module，命名为feign-api，然后引入feign的starter依赖
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
</dependencies>将order-service中编写的UserClient、User、DefaultFeignConfiguration都复制到feign-api项目中
UserClient
package cn.itcast.feign.clients;
import cn.itcast.feign.pojo.User;
import cn.itcast.feign.config.FeignClientConfiguration;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
public interface UserClient {
User findById( Long id);
}User
package cn.itcast.feign.pojo;
import lombok.Data;
public class User {
private Long id;
private String username;
private String address;
}DefaultFeignConfiguration
package cn.itcast.feign.config;
import feign.Logger;
import org.springframework.context.annotation.Bean;
/**
* @version v1.0
* @auther Bamboo
* @create 2023/7/19 13:25
*/
public class FeignClientConfiguration {
public Logger.Level feignLogLevel() {
return Logger.Level.BASIC;
}
}
在order-service中引入feign-api的依赖
<!--引入feign-api-->
<dependency>
<groupId>com.bamboo</groupId>
<artifactId>feign-api</artifactId>
<version>1.0</version>
</dependency>修改Order-service中的所有与上述三个组件有关的import部分，改成导入feign-api中的包
重启测试
扫不到FeignClient的解决方案
方式一：指定FeignClient所在包
方式二：指定FeignClient字节码
七、统一网关Gateway
网关作用
- 对用户请求做身份认证和权限校验
- 将用户请求路由到微服务，并实现负载均衡
- 对用户请求做限流
网关的技术实现
在SpringCloud中网关的实现包括两种：
- gateway
- zuul
Zuul是基于Servlet的实现，属于阻塞式编程。而SpringCloudGateway则是基于Spring5中提供的WebFlux，属于响应式编程的实现，具备更好的性能。
7.1. 搭建网关服务
创建新的module【gateway】，引入SpringCloudGateway的依赖和nacos的服务发现依赖，并创建SpringBoot启动类
<dependencies>
<!--网关依赖-->
<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>
</dependencies>package cn.itcast.gateway;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @version v1.0
* @auther Bamboo
* @create 2023/7/19 22:35
*/
public class GatewayApplication {
public static void main(String[] args) {
SpringApplication.run(GatewayApplication.class,args);
}
}配置application.yml，包括服务基本信息、nacos地址、路由
server:
port: 10010
spring:
application:
name: gateway
cloud:
nacos:
server-addr: localhost:8848 # nacos地址
gateway:
routes:
- id: user-service # 路由标识，必须唯一
# uri: http://127.0.0.1:8081 路由的目标地址，http就是固定地址
uri: lb://userservice # 路由目标地址 lb是负载均衡loadBalance 后面跟服务名称
predicates: # 路由断言，判断请求是否符合规则
- Path=/user/** # 路径断言，判断路径是否是以/user开头，如果是则符合
- id: order-service
uri: lb://orderservice
predicates:
- Path=/order/**测试
路由配置包括：
- 路由id：路由的唯一标识
- 路由目标（uri）：路由的目标地址，http代表固定地址，lb代表根据服务名负载均衡
- 路由断言（predicates）：判断路由的规则
- 路由过滤器（filters）：对请求或响应做处理
7.2. 路由断言工厂 Route Predicate Factory
- 配置文件中的predicates
- predicates: 路由断言，判断请求是否符合要求，符合则转发到路由目的地
- 配置文件中写的断言规则只是字符串，这些字符串会被Predicate Factory读取并处理，转变为路由判断的条件
- 例如Path=/user/*是按照路径匹配，这个规则是由org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory类来处理的
11中基本的Predicate工厂
| 名称 | 说明 | 示例 |
|---|---|---|
| After | 是某个时间点后的请求 | - After=2017-01-20T17:42:47.789-07:00[Asia/Shanghai] |
| Before | 是某个时间点之前的请求 | - Before=2017-01-20T17:42:47.789-07:00[America/Denver] |
| Between | 是某两个时间点之间的请求 | - Between=2017-01-20T17:42:47.789-07:00[America/Denver], 2017-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=green 或 - Query=red, gree. |
| RemoteAddr | 请求者的ip必须是指定范围 | - RemoteAddr=192.168.1.1/24 |
| Weight | 权重处理 | - Weight=group1, 2 |
7.3. 路由过滤器 GatewayFilter
过滤器链，网关中包含一系列过滤器，形成过滤器链
Spring提供了31种不同的路由过滤器工厂
| 名称 | 说明 |
|---|---|
| AddRequestHeader | 给当前请求添加一个请求头 |
| RemoveRequestHeader | 移除请求中的一个请求头 |
| AddResponseHeader | 给响应结果中添加一个响应头 |
| RemoveResponseHeader | 从响应结果中移除已有的一个响应头 |
| RequestRateLimiter | 限制请求的流量 |
实现方式：在gateway中修改application.yml文件，给userservice的路由添加过滤器
server: |
服务中测试请求头信息
package cn.itcast.user.web; |
默认过滤器
server: |
7.4. 全局过滤器GlobalFilter
不同于路由过滤器中的default-filters，全局过滤器定义在yml之外，具有高可控性
全局过滤器也是处理一切进入网关的请求和微服务响应，与GatewayFilter作用一样。
区别在于GatewayFilter通过配置定义，处理逻辑是固定的。而GlobalFilter的逻辑需要自己写代码实现，定时方式是实现GlobalFilter接口
public interface GlobalFilter { |
案例：定义全局过滤器，拦截并判断用户身份
- 参数中有authorization
- 参数值为admin
package cn.itcast.gateway; |
@Order注解与Ordered接口实现的功能一致，都是优先级
测试：http://localhost:10010/user/3?authorization=admin
7.5. 过滤器执行顺序
- 请求进入网关会碰到三类过滤器: 当前路由的过滤器、DefaultFilter、GlobalFilter
- 请求路由后，会将当前路由过滤器和DefaultFilter、GlobalFilter，合并到一个过滤器链 (集合)中，排序后依次执行每个过滤器
- DefaultFilter >>> 路由过滤器 >>> GlobalFilter
- 每一个过滤器都必须指定一个int类型的order值，order值越小，优先级越高，执行顺序越靠前
- GlobalFilter通过实现Ordered接口，或者添加@Order注解来指定order值，由我们自己指定路由过滤器和defaultFilter的order由Spring指定，默认是按照声明顺序从1递增。
- 当过滤器的order值一样时，会按照 defaultFilter > 路由过滤器>GobalFilter的顺序执行。
7.6. 网关cors跨域配置
跨域：
- 协议不同
- 域名不同
- 端口不同
跨域问题:浏览器禁止请求的发起者与服务端发生跨域ajax请求，请求被浏览器拦截的问题
解决方案:CORS
spring: |
yaml全部参数
server: |
八、Docker
微服务项目部署环境过多，例如：Nodejs、Redis、MySQL、MQ等等
Docker解决依赖的兼容问题
- 将应用的Libs（函数库）、Deps（依赖）、配置与应用一起打包
- 将每个应用放到一个隔离容器去运行，避免互相打扰
服务器详解
- 内核与硬件交互，提供操作硬件的指令
- 系统应用封装内核指令作为函数，便于程序员调用
- 用户程序基于系统函数库实现功能
Docker解决不同系统环境的问题
- Docker将用户程序与所需要调用的系统(比如Ubuntu)函数库一起打包
- Docker运行到不同操作系统时，直接基于打包的库函数，借助于操作系统的Linux内核来运行
Docker解决大型项目依赖关系复杂，不同组件依赖的兼容性问题
- Docker允许开发中将应用、依赖、函数库、配置一起打包，形成可移植镜像
- Docker应用运行在容器中，使用沙箱机制，相互隔离
Docker解决开发、测试、生产环境有差异的问题
- Docker镜像中包含完整运行环境，包括系统函数库，仅依赖系统的Linux内核，因此可以在任意Linux操作系统上运行
Docker是一个快速交付应用、运行应用的技术：
- 可以将程序及其依赖、运行环境一起打包为一个镜像，可以迁移到任意Linux操作系统
- 运行时利用沙箱机制形成隔离容器，各个应用互不干扰
- 启动、移除都可以通过一行命令完成，方便快捷
Docker与虚拟机
虚拟机（Virtual Machine）是在操作系统中模拟硬件设备，然后运行另一个操作系统，比如在windows系统中运行Ubuntu系统，这样就可以运行任意Ubuntu应用了
Docker和虚拟机的茶语：
- docker是一个系统进程；虚拟机是在操作系统中的操作系统
- docker体积小、启动速度快、性能好；虚拟机体积大、启动速度慢、性能一般
| 特性 | Docker | 虚拟机 |
|---|---|---|
| 性能 | 接近原生 | 性能较差 |
| 硬盘占用 | 一般为MB | 一般为GB |
| 启动 | 秒级 | 分钟级 |
8.1. Docker概述
镜像和容器
- 镜像（Image）：Docker将应用程序及其所需的依赖、函数库、环境、配置等文件打包在一起，称为镜像。
- 容器（Container）：镜像中的应用程序运行后形成的进程就是容器，只是Docker会给容器做隔离，对外不可见。
Docker和DockerHub
- DockerHub：DockerHub是一个Docker镜像的托管平台。这样的平台称为Docker Registry。
- 国内也有类似于DockerHub的公开服务，比如网易云镜像服务、阿里云镜像库等。
Docker架构
Docker是一个CS架构的程序，由两部分组成：
- 服务端（server）：Docker守护进程，负责处理Docker指令，管理镜像、容器等
- 客户端（client）：通过命令或RestAPI向Docker服务端发送指令。可以在本地或远程向服务端发送指令。
架构分析
- Client发送命令
- docker build
- docker pull
- docker run
- DockerServer
- docker daemon守护进程
- 存放了镜像Image和数个相互隔离的容器
- Registry
- 镜像托管服务器
总结：
- 镜像：应用程序及其依赖、环境、配置打包在一起
- 容器：镜像运行起来就是容器，一个镜像可以运行多个容器
- Docker结构
- 服务端：接收命令或远程请求，操作镜像或容器
- 客户端：发送命令或者请求到Docker服务器
- DockerHub：一个镜像托管的服务器，它们统称为DockerRegistry
8.2. Centos安装Docker
①卸载原有Docker
如果之前安装过旧版本的Docker，可以使用下面命令卸载：
yum remove docker \ |
②安装docker
首先需要虚拟机联网，安装yum工具
yum install -y yum-utils \ |
然后更新本地镜像源：
设置docker镜像源 |
然后输入命令：
yum install -y docker-ce |
docker-ce为社区免费版本。稍等片刻，docker即可安装成功。
③启动docker
Docker应用需要用到各种端口，逐一去修改防火墙设置。非常麻烦，因此建议直接关闭防火墙！
# 关闭 |
通过命令启动docker：
systemctl start docker # 启动docker服务 |
然后输入命令，可以查看docker版本：
docker -v |
④配置镜像加速
docker官方镜像仓库网速较差，我们需要设置国内镜像服务：
参考阿里云的镜像加速文档：https://cr.console.aliyun.com/cn-hangzhou/instances/mirrors
# linux控制台命令 |
8.3. Docker命令
8.3.1 Docker命令帮助文档
- docker –help
- docker images –help
8.3.2 镜像相关命令
- 镜像名称一般分为两部分组成：[repository]:[tag]
- 在没有指定tag时，默认是latest，代表最新版本的镜像
- 例：mysql:5.7
- repository就是mysql
- tag就是5.7
- 构建镜像：Dockerfile使用命令docker build进行对本地Local的Docker环境进行构建镜像
- 推送镜像：本地Local的Docker使用命令docker push推送镜像到服务(镜像服务器Docker Registry)
- 拉取镜像：本地Local的Docker使用明亮docker pull从服务(镜像服务器Docker Registry)拉取镜像
- 查看镜像：docker images
- 删除镜像：docker rmi
- 保存镜像为一个压缩包tar：docker save
- 加载压缩包为镜像：docker load
使用案例
拉取nginx
- ①打开镜像仓库DockerHubhttps://hub.docker.com/搜索nginx
- ②复制docker pull nginx到控制台进行安装Nginx
- ③使用docker images查看拉取到本地的镜像
导出导入磁盘
利用docker xx –help命令查看docker save和docker load的语法
导出
docker save –help
查询到以下关键信息
# 使用方式
Usage: docker save [OPTIONS] IMAGE [IMAGE...]
Options:
# option选择
-o, --output string Write to a file, instead of STDOUT查看镜像：docker images
保存：docker save -o nginx.tar nginx:latest
查看保存的镜像：ll
导入
查看镜像：docker images
删除Nginx镜像：docker rmi -f nginx:latest
docker load –help
查到以下帮助
Usage: docker load [OPTIONS]
Options:
-i, --input string Read from tar archive file, instead of STDIN
-q, --quiet Suppress the load output导入：docker load -i nginx.tar
8.3.3 容器相关命令
- 运行：docker run
- 暂停：docker pause
- 取消暂停：docker unpause
- 停止：docker stop
- 停止后启动：docker start
- 进入容器执行命令：docker exec
- 查看容器运行日志：docker logs
- 查看所有运行的容器及状态：docker ps
- 删除指定容器：docker rm
使用案例1
创建运行一个Nginx容器
①docker hub查看Nginx容器运行命令
docker run --name some-nginx -d -p 8080:80 some-content-nginx |
命令解读：
- docker run：创建并运行起一个容器
- –name：给容器起名，mn【my nginx】
- -p：将宿主机端口与容器端口映射，冒号左侧是宿主机端口，右侧是容器端口
- -d：后台运行容器
- some-content-nginx：镜像名称，例如nginx
- 原理：客户端通过http://192.168.49.10:80访问Linux，通过绑定的80端口链接到Docker容器
②运行
docker run --name mn -p 80:80 -d nginx:latest |
③查看容器状态
docker ps |
④查看nginx容器日志
# 持续查看日志 |
使用案例2
进入Nginx容器修改HTML内容
①进入容器
docker exec -it mn bash |
命令解读：
- docker exec：进入容器内部，执行一个命令
- -it：给当前进入的容器创建一个标准输入、输出终端，允许我们与容器交互
- mn：要进入的容器的名称
- bash：进入容器后执行的命令，bash是一个linux终端的交互命令
②进入nginx容器的HTML所在目录
具体可在docker hub中查看
cd /usr/share/nginx/html/ |
③修改index.html的内容
sed -i 's#Welcome to nginx#Bamboo welcome to you#g' index.html |
④其他命令调用
- docker stop mn：停止nginx容器
- docker ps -a：查看所有状态的容器
- docker rm -f mn：不能删除运行中的容器，除非添加 -f 参数
- 进入容器
- 命令是docker exec -it [容器名] [要执行的命令]
- exec命令可以进入容器修改文件，但是在容器内修改文件是不推荐的
使用案例3
进入redis容器，执行redis-cli客户端命令，存入num=666
官网：start with persistent storage
docker run --name some-redis -d redis redis-server --save 60 1 --loglevel warning |
①启动redis容器
docker run --name mr -p 6379:6379 -d redis redis-server --save 60 1 --loglevel warning |
②进入redis容器
docker exec -it mr bash |
③连接并使用redis
# 连接redis服务 |
8.4. 数据卷
容器与数据耦合的问题：
- 不便于修改：修改Nginx的html内容时，需要进入容器内部修改，不方便
- 数据不可复用：在容器内的修改对外是不可见的，所有修改对新创建的容器是不可复用的
- 升级维护困难：数据在用期内，如果需要升级容器必然删除旧容器，所有数据都跟着删除了
数据卷（volume）是一个虚拟目录，指向宿主机文件系统中的某个目录
- DockerHost存在以下两个
- 宿主机文件系统
- /var/lib/docker/volumes/html
- /var/lib/docker/volumes/conf
- Volumes
- html：指向宿主机文件系统的html
- conf：指向宿主机文件系统的conf
- 宿主机文件系统
- Container
- /etc/nginx/conf指向Volumes中的conf
- /usr/share/nginx/html指向Volumes中的html
8.4.1 数据卷基本语法
docker volume [COMMAND] |
docker volume命令是数据卷操作，根据命令后跟随的command来确定下一步的操作：
- create 创建一个volume
- inspect 显示一个或多个volume的信息
- ls 列出所有的volume
- prune 删除未使用的volume
- rm 删除一个或多个指定的volume
使用案例
①创建数据卷
docker volume create html |
②查看所有数据
docker volume ls |
③查看数据卷详细信息卷
docker volume inspect html |
④删除数据卷
# 删除未使用的数据卷 |
8.4.2 数据卷挂载
方式一：volume数据卷挂载
- -v volumeName: /targetContainerPath
- 如果容器运行时volume不存在，会自动被创建出来
docker run \ # 创建并运行容器 |
方式二：目录挂载
提示：目录挂载与数据卷挂载的语法是类似的：
- -v [宿主机目录]:[容器内目录]
- -v [宿主机文件]:[容器内文件]
使用案例1
创建nginx容器，修改容器内的html目录中的index.html
# 创建容器并挂载html数据卷到容器内HTML目录 |
使用案例2
创建并运行MySQL 容器，将宿主机目录直接挂载到容器
①搭建初始环境
# 拉取mysql |
②将hmy.cnf上传到/tmp/mysql/conf
# hmy.cnf |
③查阅docker hub挂载
docker run \ |
8.4.3 数据卷挂载总结
- docker run命令通过-v参数挂载文件或目录到容器中：
- -v volume名称:容器内目录
- -v 宿主机文件:容器内文件
- -v 宿主机目录:容器内目录
- 数据卷挂载与目录直接挂载的区别
- 数据卷挂载耦合度低，由docker来管理目录，但是目录较深，不好找
- 目录挂载耦合度高，需要我们自己管理目录，不过目录容易寻找查看
8.5. Dockerfile自定义镜像
镜像结构
- 镜像是将应用程序及其需要的系统函数库、环境、配置、依赖打包而成
- 它是一个分层结构，每一层称为一个Layer
- Entrypoint：镜像运行入口，一般是程序启动的脚本和参数
- 层：Layer，夹在Entrypoint和BaseImage中间，在BaseImage基础上添加安装包、依赖、配置等，每次操作都形成新的一层
- BaseImage层：基础镜像层，包含基本的系统函数库、环境变量、文件系统
Dockerfile
Dockerfile就是一个文本文件，其中包含一个个指令(Instruction)，用指令来说明要执行说明操作来构建镜像，每个指令都会形成一层Layer
| 指令 | 说明 | 示例 |
|---|---|---|
| FROM | 指定基础镜像 | FROM centos:6 |
| ENV | 设置环境变量，可在后面指令使用 | ENV key value |
| COPY | 拷贝本地文件到镜像的指定目录 | COPY ./mysql-5.7.rpm /tmp |
| RUN | 执行Linux的shell命令，一般是安装过程的命令 | RUN yum install gcc |
| EXPOSE | 指定容器运行时监听的端口，是给镜像使用者看的 | EXPOSE 8080 |
| ENTRYPOINT | 镜像中应用的启动命令，容器运行时调用 | ENTRYPOINT java -jar xx.jar |
①使用案例
基于Ubuntu镜像构建一个新镜像，运行一个java项目
Dockerfile文件内容
# 1.指定基础镜像 |
新建一个空文件夹，将docker-demo.jar、jdk8.tar.gz、Dockerfile拷贝到此目录下
mkdir -p /tmp/docker-demo
进入docker-demo目录运行命令
docker build -t javaweb:1.0 .
查看镜像
docker images
运行docker容器
docker run --name web -p 8090:8090 -d javaweb:1.0
测试地址
http://192.168.49.10:8090/hello/count
②基于java:8-alpine镜像
优化前一个使用案例的Dockerfile
# 指定基础镜像 |
8.6. DockerCompose
什么是DockerCompose
Docker Compose可以基于Compose文件帮我们快速部署分布式应用，而无需手动一个个创建和运行容器
Compose文件是一个文本文件，通过指令定义集群中的每个容器如何运行
version:"3.8"
services:
mysql:
image: mysql:5.7.42
environment:
MYSQL_ROOT_PASSWORD: 123
volumes:
- /tmp/mysql/data:/var/lib/mysql
- /tmp/mysql/conf/hmy.cnf:/etc/mysql/conf.d/hmy.cnf
web:
build: .
ports:
- 8090: 8090
CentOS7安装DockerCompose
①下载
Linux下需要通过命令下载：
# 安装 |
上传docker-compose到/usr/local/bin/目录也可以
②修改文件权限
# 修改权限 |
③Base自动补全命令
# 补全命令 |
如果这里出现错误，需要修改自己的hosts文件：
echo "199.232.68.133 raw.githubusercontent.com" >> /etc/hosts |
实例：部署微服务项目至Docker
①文件目录
cloud-demo
gateway
app.jar
Dockerfile
FROM java:8-alpine
COPY ./app.jar /tmp/app.jar
ENTRYPOINT java -jar /tmp/app.jar
mysql
conf
hmy.cnf
[mysqld]
skip-name-resolve
character_set_server=utf8
datadir=/var/lib/mysql
server-id=1000
data
order-service
app.jar
Dockerfile
FROM java:8-alpine
COPY ./app.jar /tmp/app.jar
ENTRYPOINT java -jar /tmp/app.jar
user-service
app.jar
Dockerfile
FROM java:8-alpine
COPY ./app.jar /tmp/app.jar
ENTRYPOINT java -jar /tmp/app.jar
docker-compose.yml
version: "3.2"
services:
nacos:
# docker pull nacos/nacos-server 相当于拉取了一个nacos镜像
image: nacos/nacos-server
environment:
MODE: standalone
ports:
- "8848:8848"
mysql:
image: mysql:5.7.42
environment:
MYSQL_ROOT_PASSWORD: root
volumes:
# $PWD为当前linux目录
- "$PWD/mysql/data:/var/lib/mysql"
- "$PWD/mysql/conf:/etc/mysql/conf.d/"
userservice:
# 在当前目录的suer-service下找docker进行构建
build: ./user-service
orderservice:
build: ./order-service
gateway:
build: ./gateway
ports:
- "10010:10010"
②修改本地项目yml
将地址改成服务名称
spring: |
③本地项目打包
<build> |
注：
- feign-api不需要添加此配置，只需要增加
<packaging>jar</packaging>即可 - 父pom不要设置spring-boot-maven-plugin
④上传linux
- 将cloud-demo上传至/tmp并进入文件夹内
- docker-compose up -d 后台运行启动
- docker-compose –help 查看命令帮助
- docker ps 查看状态
- docker-compose logs -f 查看并跟踪日志
- 发现错误，nacos未启动时其他微服务注册
- docker-compose restart gateway userservice orderservice 重启服务解决nacos注册问题
8.7. Docker镜像仓库
8.7.1 搭建私有仓库
搭建镜像仓库可以基于Docker官方提供的DockerRegistry来实现。
官网地址：https://hub.docker.com/_/registry
①简化版镜像仓库
Docker官方的Docker Registry是一个基础版本的Docker镜像仓库，具备仓库管理的完整功能，但是没有图形化界面。
搭建方式比较简单，命令如下：
docker run -d \ |
命令中挂载了一个数据卷registry-data到容器内的/var/lib/registry 目录，这是私有镜像库存放数据的目录。
访问http://YourIp:5000/v2/_catalog 可以查看当前私有镜像服务中包含的镜像
②带有图形化界面版本
使用DockerCompose部署带有图象界面的DockerRegistry，命令如下：
version: '3.0' |
③配置Docker信任地址
我们的私服采用的是http协议，默认不被Docker信任，所以需要做一个配置：
# 打开要修改的文件 |
8.7.2 私有镜像仓库拉取推送
①重新tag本地镜像，名称前缀为私有仓库地址：192.168.49.10:8080
docker tag nginx:latest 192.168.49.10:8080/nginx:1.0 |
②推送镜像
docker push 192.168.49.10:8080/nginx:1.0 |
③拉取镜像
docker pull 192.168.49.10:8080/nginx:1.0 |
九、RabbitMQ服务异步通讯
9.1. 同步与异步
同步调用问题：微服务之间基于Feign的调用就属于同步方式
- 如多个服务之间逐个调用，调用时间过长
- 如中间调用某一服务宕机，则会将调用卡到此处无法继续调用其他服务
会导致：
- 耦合度高
- 每次加入新的需求，都要修改原来的代码
- 性能下降
- 调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用时间之和
- 资源浪费
- 调用链中的每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下回嫉妒浪费系统资源
- 级联失败
- 如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌，迅速导致整个微服务群故障
同步调用优点：
- 时效性较强，可以立即得到结果
同步调用问题：
- 耦合度高
- 性能和吞吐能力下降
- 有额外的资源消耗
- 有级联失败问题
异步调用常见实现是事件驱动模式，Broker实现
异步通信的优点：
- 耦合度低
- 吞吐量提升
- 故障隔离
- 流量削峰
异步通信缺点：
- 依赖于Broker的可靠性、安全性、吞吐能力
- 架构复杂，业务没有明显的流程线，不好追踪管理
9.2. MQ概念
MQ（MessageQueue），中文消息队列，存放消息的队列，事件驱动架构中的Broker
| RabbitMQ | ActiveMQ | RocketMQ | Kafka | |
|---|---|---|---|---|
| 公司/社区 | Rabbit | Apache | 阿里 | Apache |
| 开发语言 | Erlang | Java | Java | Scala&Java |
| 协议支持 | AMQP,XMPP,SMTP,STOMP | OpenWire,STOMP,REST,XMPP,AMQP | 自定义协议 | 自定义协议 |
| 可用性 | 高 | 一般 | 高 | 高 |
| 单机吞吐量 | 一般 | 差 | 高 | 非常高 |
| 消息延迟 | 微秒级 | 毫秒级 | 毫秒级 | 毫秒以内 |
| 消息可靠性 | 高 | 一般 | 高 | 一般 |
9.3. RabbitMQ部署
①下载镜像
本地拉取：将my.tar放进/tmp中，执行命令
docker load -i mq.tar
在线拉取
docker pull rabbitmq:3-management
②安装容器
执行下面的命令来运行MQ容器： |
RabbitMQ中的几个概念
- channel：操作MQ的工具
- exchange：路由消息到队列中
- queue：缓存消息
- virtual host：虚拟主机，是对queue、excahnge等资源的逻辑分组
9.4. 常见消息模型
MQ官方文档给出了五个MQ的Demo示例，对应了几种不同的用法：
- 基本消息队列（BasicQueue）
- 工作消息队列（WorkQueue）
- 发布订阅（Publish、Subscribe），又根据交换机类型不同分为三种：
- Fanout Exchange：广播
- Direct Exchange：路由
- Topic Exchange：主题
HelloWorld案例
官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色：
- publisher：消息发布者，将消息发送到队列queue
- queue：消息队列，负责接受并缓存消息
- consumer：订阅队列，处理队列中的消息，处理完后此消息在队列中被销毁
publisher —> queque —> consumer
基本消息队列的消息发送流程：
- 建立connection
- 创建channel
- 利用channel声明队列
- 利用channel向队列发送消息
package cn.itcast.mq.helloworld; |
基本消息队列的消息接收流程：
- 建立connection
- 创建channel
- 利用channel声明队列
- 定义consumer的消费行为handleDelivery()
- 利用channel将消费者与队列绑定
package cn.itcast.mq.helloworld; |
9.5. SpringAMQP
AMQP：全称Advanced Message Queuing Protocal（高级消息队列协议），是用于应用程序之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。
Spring AMQP是基于AMQP协议定义的一套API规范，提供了模块来发送和接收消息。包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现。
使用事项
- 发送
- 注入RabbitTemplate
- 调用convertAndSend()方法发送
- 接收
- 创建类添加@Componet组件，方法注入@RabbitListener(queues = “simple.queue”)
- 启动SpringBoot自动接收
①引入SpringAMQP依赖
<!--AMQP依赖，包含RabbitMQ--> |
②编辑application.yml
spring: |
③发送/接收消息
9.5.1 BasicQueue基本消息队列
发送simple.queue
|
接收simple.queue
添加一个组件，接收消息
public class SpringRabbitListener {
// basicQueue
public void listenSimpleQueue(String msg) {
System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
}
}启动SpringBoot
public class ConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(ConsumerApplication.class, args);
}
}
9.5.2 WorkQueue工作消息队列
可以提高消息处理速度，避免队列消息堆积
- publisher
- queue
- consumer1
- consumer2
- queue
案例
模拟WorkQueue，实现一个队列绑定多个消费者
Publisher
public class SpringAMQPTest {
private RabbitTemplate rabbitTemplate;
public void testSendMessageWorkQueue() throws InterruptedException {
String queueName = "simple.queue";
String message = "hello, message__";
for (int i = 1; i <= 50; i++) {
rabbitTemplate.convertAndSend(queueName, message + i);
Thread.sleep(20);
}
}
}Consumer
public class SpringRabbitListener {
// workQueue
public void listenWorkQueue1(String msg) throws InterruptedException {
System.out.println("消费者1接收到消息：【" + msg + "】 " + LocalTime.now());
Thread.sleep(20);
}
public void listenWorkQueue2(String msg) throws InterruptedException {
System.err.println("消费者2接收到消息：【" + msg + "】 " + LocalTime.now());
Thread.sleep(200);
}
}Consumer下的application.yml设置prefetch
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 192.168.49.10 # 主机名
port: 5672 # 端口号
virtual-host: / # 虚拟主机
username: bamboo # 用户名
password: root # 密码
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息，默认接收消息无限制
9.5.3 发布订阅（Publish、Subscribe）
发布订阅模式的特点是允许将同一消息发送给多个消费者，实现方式是加入exchange（交换机），常见exchange类型包括：
- Fanout：广播
- Direct：路由
- Topic：话题
注意：exchange负责消息路由，而不是存储，路由失败则消息丢失
① Fanout Exchange
Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue
- publisher
- excahnge【itcast.fanout】
- fanout.queue1
- consumer1
- fanout.queue2
- consumer2
- fanout.queue1
- excahnge【itcast.fanout】
①在consumer服务声明Exchange、Queue、Binding作为FanoutConfig
//创建config包下的FanoutConfig类 |
②修改监听组件
package cn.itcast.mq.listener; |
③启动consumer服务等待消息接收
④编写publisher代码发送消息
|
⑤结果：publisher发布一次消息，consumer的多个订阅将接收到信息
② Direct Exchange
Direct Exchange会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式(routes)
- 每一个Queue都与Exchange设置一个BindingKey
- 发布者发布消息时，指定消息的RoutingKey
- Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
@RabbitListenner注解
@Queue注解
@Exchange注解
①使用@RabbitListener设置consumer
/** |
②publisher发送消息
/** |
③ Topic Exchange
与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以.分割。
Queue与Exchange指定BindingKey时可以使用通配符：
- #：代指0个或多个单词
- *：代指一个单词
①编写consumer监听
/** |
②编写publisher代码
/** |
9.6. SpringAMQP消息转换器
消息转换器实现序列化和反序列化
- 原生是利用JDK的序列化进行默认实现MessageConverter
- 注意发送方与接收方必须使用相同的MessageConverter
简单测试
发送消息：注册一个队列bean
package cn.itcast.mq.config; |
使用publisher发送object消息
/** |
默认序列化成：
rO0ABXNyABFqYXZhLnV0aWwuSGFzaE1hcAUH2sHDFmDRAwACRgAKbG9hZEZhY3RvckkACXRocmVzaG9sZHhwP0AAAAAAAAx3CAAAABAAAAACdAAEbmFtZXQA |
添加MessageConverter
①依赖引入
父pom引入或模块引入都可
父pom引入【父pom基于spring-boot-starter-parent】
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>模块引入
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-xml</artifactId>
<version>2.9.10</version>
</dependency>
②在publisher服务声明MessageConverter
package cn.itcast.mq; |
接收消息：重复上述步骤，注册一个MessageConverter的@Bean组件，使用与发送类型相同的类型参数接收信息
package cn.itcast.mq.listener; |
十、Elasticsearch
elasticsearch是一款非常强大的开源搜索引擎，可以帮助我们从海量数据中快速找到需要的内容。
elasticsearch是elastic stack (ELK)的核心，负责存储、搜索、分析数据。
- 数据可视化：Kibana
- 存储、计算、搜索数据：Elasticsearch
- 数据抓取：Logstash、Beats
Elasticsearch底层实现：Lucene
- 是Apache的开源搜索引擎类库，提供了搜索引擎的核心API
elasticsearch的发展 Lucene是一个ava语言的搜索引擎类库，是Apache公司的顶级项目，由DougCutting于1999年研发官网地址: https://lucene.apache.org/。
Lucene的优势:
- 易扩展
- 高性能(基于倒排索引)
2004年shay Banon基于Lucene开发了Compass
2010年shay Banon 重写了Compass，取名为Elasticsearch。官网地址: https://www.elastic.co/cn/
相比与lucene，elasticsearch具备下列优势
- 支持分布式，可水平扩展
- 提供Restful接口，可被任何语言调用
10.1. 概述
正向索引和倒排索引
elasticsearch采用倒排索引:
文档(document):每条数据就是一个文档
词条(term):文档按照语义分成的词语
什么是文档和词条?
- 每一条数据就是一个文档
- 对文档中的内容分词，得到的词语就是词条
什么是正向索引?
- 基于文档id创建索引。查询词条时必须先找到文档，而后判断是否包含词条
什么是倒排索引?
- 对文档内容分词，对词条创建索引，并记录词条所在文档的信息。查询时先根据词条查询到文档id，而后获取到文档
索引(Index)
- 索引(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中是Json格式
- 字段:Json文档中的字段
- 索引:同类型文档的集合
- 映射:索引中文档的约束，比如字段名称、类型
- elasticsearch与数据库的关系
- 数据库负责事务类型操作
- elasticsearch负责海量数据的搜索、分析、计算
10.2. 安装
10.2.1 部署单点es
创建网络：还需要部署kibana容器，因此需要让es和kibana容器互联
docker network create es-net
加载镜像：这里采用elasticsearch的7.12.1版本的镜像，这个镜像体积非常大，接近1G。不建议pull
# 导入数据
docker load -i es.tar运行：运行docker命令，部署单点es
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.49.10:9200 即可看到elasticsearch的响应结果
// 20230725160609
// http://192.168.49.10:9200/
{
"name": "9d54b23c2bcb",
"cluster_name": "docker-cluster",
"cluster_uuid": "-0wNglZGQtSCxh3oAvAirA",
"version": {
"number": "7.12.1",
"build_flavor": "default",
"build_type": "docker",
"build_hash": "3186837139b9c6b6d23c3200870651f10d3343b7",
"build_date": "2021-04-20T20:56:39.040728659Z",
"build_snapshot": false,
"lucene_version": "8.8.0",
"minimum_wire_compatibility_version": "6.8.0",
"minimum_index_compatibility_version": "6.0.0-beta1"
},
"tagline": "You Know, for Search"
}
10.2.2 部署kibana
kibana可以给我们提供一个elasticsearch的可视化界面，便于学习
导入kibana.tar到/tmp下
docker load -i kibana.tar
部署
docker run -d \
--name kibana \
-e ELASTICSEARCH_HOSTS=http://es:9200 \
--network=es-net \
-p 5601:5601 \
kibana:7.12.1
# 版本需要与es一致--network es-net：加入一个名为es-net的网络中，与elasticsearch在同一个网络中-e ELASTICSEARCH_HOSTS=http://es:9200"：设置elasticsearch的地址，因为kibana已经与elasticsearch在一个网络，因此可以用容器名直接访问elasticsearch-p 5601:5601：端口映射配置
kibana启动一般比较慢，需要多等待一会，可以通过命令：
docker logs -f kibana
查看运行日志，当查看到下面的日志，说明成功
"message":"http server running at http://0:5601"
10.2.3 安装IK分词器
分词器
es在创建倒排索引时需要对文档分词;在搜索时，需要对用户输入内容分词。但默认的分词规则对中文处理并不友好我们在kibana的DevTools中测试：
GET _search |
语法说明:
- POST:请求方式
- /_analyze: 请求路径，这里省略了http://192.168.150.101:9200，有kibana帮我们补充
- 请求参数，json风格：
- analyzer:分词器类型，这里是默认的standard分词器
- text:要分词的内容
安装IK分词器
在线安装：
# 进入容器内部
docker exec -it elasticsearch /bin/bash
# 在线下载并安装
./bin/elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.12.1/elasticsearch-analysis-ik-7.12.1.zip
#退出
exit
#重启容器
docker restart elasticsearch离线安装【建议】
查看数据目录
安装插件需要知道elasticsearch的plugins目录位置，而我们用了数据卷挂载，因此需要查看elasticsearch的数据卷目录，通过下面命令查看:
docker volume inspect es-plugins
显示结果：
[
{
"CreatedAt": "2023-07-25T14:46:01+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这个目录中。解压缩分词器安装包
解压ik压缩包，将文件重命名为ik上传到es容器的插件数据卷中也就是
/var/lib/docker/volumes/es-plugins/_data重启容器
4、重启容器
docker restart es# 查看es日志
docker logs -f es测试
IK分词器包含两种模式：
ik_smart：最少切分ik_max_word：最细切分
GET /_analyze
{
"analyzer": "ik_max_word",
"text": "黑马程序员学习java太棒了"
}结果：
{
"tokens" : [
{
"token" : "黑马",
"start_offset" : 0,
"end_offset" : 2,
"type" : "CN_WORD",
"position" : 0
},
{
"token" : "程序员",
"start_offset" : 2,
"end_offset" : 5,
"type" : "CN_WORD",
"position" : 1
},
{
"token" : "程序",
"start_offset" : 2,
"end_offset" : 4,
"type" : "CN_WORD",
"position" : 2
},
{
"token" : "员",
"start_offset" : 4,
"end_offset" : 5,
"type" : "CN_CHAR",
"position" : 3
},
{
"token" : "学习",
"start_offset" : 5,
"end_offset" : 7,
"type" : "CN_WORD",
"position" : 4
},
{
"token" : "java",
"start_offset" : 7,
"end_offset" : 11,
"type" : "ENGLISH",
"position" : 5
},
{
"token" : "太棒了",
"start_offset" : 11,
"end_offset" : 14,
"type" : "CN_WORD",
"position" : 6
},
{
"token" : "太棒",
"start_offset" : 11,
"end_offset" : 13,
"type" : "CN_WORD",
"position" : 7
},
{
"token" : "了",
"start_offset" : 13,
"end_offset" : 14,
"type" : "CN_CHAR",
"position" : 8
}
]
}扩展词词典
随着互联网的发展，“造词运动”也越发的频繁。出现了很多新的词语，在原有的词汇列表中并不存在。比如：“奥力给”，“传智播客” 等。
所以我们的词汇也需要不断的更新，IK分词器提供了扩展词汇的功能。
打开IK分词器config目录：
./ik/config/IKAnalyzer.cfg.xml
<properties>
<comment>IK Analyzer 扩展配置</comment>
<!--用户可以在这里配置自己的扩展字典 *** 添加扩展词典-->
<entry key="ext_dict">ext.dic</entry>
</properties>新建一个 ext.dic，可以参考config目录下复制一个配置文件进行修改
传智播客
奥力给重启elasticsearch
docker restart es
# 查看 日志
docker logs -f elasticsearch测试效果
GET /_analyze
{
"analyzer": "ik_max_word",
"text": "传智播客Java就业超过90%,奥力给！"
}
停用词词典
在互联网项目中，在网络间传输的速度很快，所以很多语言是不允许在网络上传递的，如：关于宗教、政治等敏感词语，那么我们在搜索时也应该忽略当前词汇。
IK分词器也提供了强大的停用词功能，让我们在索引时就直接忽略当前的停用词汇表中的内容。
IKAnalyzer.cfg.xml配置文件内容添加:
<properties>
<comment>IK Analyzer 扩展配置</comment>
<!--用户可以在这里配置自己的扩展字典-->
<entry key="ext_dict">ext.dic</entry>
<!--用户可以在这里配置自己的扩展停止词字典 *** 添加停用词词典-->
<entry key="ext_stopwords">stopword.dic</entry>
</properties>在 stopword.dic 添加停用词
习大大
重启elasticsearch
# 重启服务
docker restart elasticsearch
docker restart kibana
# 查看 日志
docker logs -f elasticsearch测试效果：
GET /_analyze
{
"analyzer": "ik_max_word",
"text": "传智播客Java就业率超过95%,习大大都点赞,奥力给！"
}
10.2.4 部署es集群
部署es集群可以直接使用docker-compose来完成，不过要求你的Linux虚拟机至少有4G的内存空间
首先编写一个docker-compose文件，内容如下：
version: '2.2' |
Run docker-compose to bring up the cluster:
docker-compose up |
10.3. 索引库
mapping属性
mapping是对索引库中文档的约束，常见的mapping属性包括
- type:字段数据类型，常见的简单类型有:
- 字符串:text (可分词的文本)、keyword (精确值，例如:品牌、国家、ip地址)
- 数值: long、integer、short、byte、double、float、
- 布尔: boolean
- 日期:date
- 对象:object
- index:是否创建索引，默认为true
- analyzer:使用哪种分词器
- properties:该字段的子字段
10.3.1 索引库操作
创建索引库
# 创建索引库 /heima为索引库名 |
查看、删除索引库
查看：
GET /索引库名 |
删除：
DELETE /索引库名 |
修改索引库
索引库和mapping一旦创建无法修改，但是可以添加新的字段
PUT /索引库名/_mapping |
索引库总结：索引库操作
- 创建索引库 :
PUT /索引库名 - 查询索引库 :
GET /索引库名 - 删除索引库 :
DELETE /索引库名 - 添加字段 :
PUT /索引库名/_mapping
10.3.2 文档操作
- 添加文档：
POST /索引库名/_doc/文档id{json文档} - 查询文档：
GET /索引库名/_doc/文档id - 删除文档：
DELETE /索引库名/_doc/文档id
增、删、查DSL语法
POST /索引库名/_doc/文档id |
修改文档
- 全量修改：既能修改也能新增
PUT /索引库名/_doc/文档id{json文档} - 增量【局部】修改：
POST /索引库名/_update/文档id{"doc":{字段}}
# 全局修改 |
文档操作总结：
- 创建文档：
POST /索引库名/_doc/文档id{json文档} - 查询文档：
GET /索引库名/_doc/文档id - 删除文档：
DELETE /索引库名/_doc/文档id - 修改文档
- 全量修改：
PUT/索引库名/_doc/文档id{json文档} - 增量修改：
POST /索引库名/_update/文档id{"doc":{字段}}
- 全量修改：
10.4. RestClient
数据库导入
CREATE TABLE `tb_hotel` ( |
10.4.1 操作索引库
索引表语法
# 酒店的mapping |
ES中支持两种地理坐标数据类型:
- geo_point：由纬度(latitude)和经度(longitude)确定的一个点。例如:”32.8752345,120.2981576”
- geo_shape：有多个geo_point组成的复杂几何图形。例如一条直线，”LINESTRING(-77.03653 38.8976761, -77.009051 38.889939)”
字段拷贝可以使用copy_to属性将当前字段拷贝到指定字段。示例:
"all": { |
1）简单应用
①导入依赖
<!--JavaRestClient--> |
②测试
/** |
① 创建索引库
创建CreateIndexRequest对象
CreateIndexRequest request = new CreateIndexRequest("hotel");
准备请求的参数，DSL语句
request.source(MAPPING_TEMPLATE, XContentType.JSON);
发送请求
client.indices().create(request, RequestOptions.DEFAULT);
注意事项：CreateIndexRequest导包易错点
//导入这个包就可以正常运行了 |
①添加常量
package cn.itcast.hotel.constants; |
②测试创建索引
package cn.itcast.hotel; |
② 删除索引库
创建DeleteIndexRequest对象
DeleteIndexRequest request = new DeleteIndexRequest("hotel");
发送请求
client.indices().delete(request, RequestOptions.DEFAULT);
import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE; |
③ 判断索引库是否存在
创建GetIndexRequest对象
GetIndexRequest request = new GetIndexRequest("hotel");
发送请求
boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
import static cn.itcast.hotel.constants.HotelConstants.MAPPING_TEMPLATE; |
④ 总结
索引库操作的基本步骤：
- 初始化RestHighLevelClient
- 创建XxxlndexRequest。XXX是CREATE、Get、Delete
- 准备DSL (CREATE时需要)
- 发送请求。调用RestHighLevelClient#indices().xxx()方法xxx是create、exists、delete
10.4.2 操作文档
① 新建文档
准备Request对象
IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
准备Json文档
request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
发送请求
client.index(request, RequestOptions.DEFAULT);
/** |
② 查询文档
准备Request
GetRequest request = new GetRequest("hotel","394796");
发送请求，得到响应
GetResponse response = client.get(request, RequestOptions.DEFAULT);
解析响应结果
String json = response.getSourceAsString();
JSON处理数据为Object
HotelDoc hotelDoc = JSON.parseObject(json, HotelDoc.class);
/** |
③ 修改文档
修改文档数据有两种方式:
- 方式一：全量更新。再次写入id一样的文档，就会删除旧文档，添加新文档
- 方式二：局部更新。只更新部分字段，我们演示方式二
方式
准备Request对象
UpdateRequest request = new UpdateRequest("hotel", "394796");
准备请求参数
request.doc(
"price","9999",
"starName","超星级"
);发送请求
client.update(request, RequestOptions.DEFAULT);
/** |
④ 删除文档
准备Request
DeleteRequest request = new DeleteRequest("hotel","394796");
发送请求
client.delete(request, RequestOptions.DEFAULT);
/** |
⑤ 批量导入文档
- 使用Bulk批处理，实现批量新增文档
- 批量查询：
GET /hotel/_search
/** |
⑥ 总结
文档操作的基本步骤
- 初始化RestHighLevelClient
- 创建XxxRequest。XXX是Index、Get、Update、Delete
- 准备参数 (Index和Update时需要)
- 发送请求。调用RestHighLevelClient#.xxx()方法，xxx是index、get、update、delete
- 解析结果 (Get时需要)
10.5. DSL语法
10.5.1 DSL语法
Elasticsearch提供了基于JSON的DSL（Domain Specific Language）来定义查询。常见的查询类型包括：
- 查询所有
- 全文检索（full text）
- 精确查询
- 地理（geo）查询
- 复合（compound）查询
①查询所有：查询出所有数据，一般用于测试。例如：match_all
GET /indexName/_search |
②全文检索（full text）查询：利用分词器对用户输入内容粉刺，然后去倒排索引库中匹配。例如：
match_query
# match查询
GET /indexName/_search
{
"query": {
"match": {
"FIELD": "TEXT"
}
}
}
# 示例：match查询
GET /hotel/_search
{
"query": {
"match": {
"all": "外滩"
}
}
}multi_match_query：与match查询相似，但允许同时查询多个字段
# multi_match查询
GET /indexName/_search
{
"query": {
"multi_match": {
"query": "TEXT",
"fields": ["FIELD1","FIELD2"]
}
}
}
# 示例
GET /hotel/_search
{
"query": {
"multi_match": {
"query": "外滩如家",
"fields": ["brand","name","business"]
}
}
}【意见】：建议使用copy_to拷贝多个属性到一个字段中，使用match查询；multi_match查询是根据多个字段查询，参与查询字段越多，查询性能越差。
③精确查询：根据精确词条值查找数据，一般是查找keyword、数值、日志、boolean等类型字段。例如：
range：根据值的范围查询（数值、日期的范围）
# range查询
GET /indexName/_search
{
"query": {
"range": {
"FIELD": {
"gte": 10,
"lte": 20
}
}
}
}
# 示例
GET /hotel/_search
{
"query": {
"range": {
"price": {
"gte": 100,
"lte": 300
}
}
}
}term：根据词条精确值查询（keyword类型、数值类型、布尔类型、日期类型字段）
# term查询
GET /indexName/_search
{
"query": {
"term": {
"FIELD": {
"value": "VALUE"
}
}
}
}
# 示例
GET /hotel/_search
{
"query": {
"term": {
"city": {
"value": "上海"
}
}
}
}
④地理（geo）查询：根据经纬度查询。例如：
geo_distance【常用】：半径查询，即以坐标为原点半径以内的数据被检索
# distance查询
GET /indexName/_search
{
"query": {
"geo_distance":{
"distance": "15km",
"location": "31.21,121.5"
}
}
}
# 示例
GET /hotel/_search
{
"query": {
"geo_distance":{
"distance": "15km",
"location": "31.21,121.5"
}
}
}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
}
}
}
}
}
⑤复合（compound）查询：复合查询可以将上述各种查询条件组合起来，合并查询条件。例如：
bool
- 一个或多个查询子句的组合。组合方式有：
- must：必须匹配每个子查询，类似“与”
- should：选择性匹配子查询，类似“或”
- must_not：必须不匹配，不参与算分，类似“非”
- filter：必须匹配，不参与算分
GET /hotel/_search
{
"query": {
"bool": {
"must": [
{
"match": {
"name": "如家"
}
}
],
"must_not": [
{
"range": {
"price": {
"gt": 400
}
}
}
],
"filter": [
{
"geo_distance": {
"distance": "10km",
"location": {
"lat": 31.21,
"lon": 121.5
}
}
}
]
}
}
}- 一个或多个查询子句的组合。组合方式有：
function_score
- 定义的三要素
- 过滤条件：哪些文档要加分
- 算分函数：如何计算Function score
- 加权方式：function score 与 query score如何运算
GET /indexName/_search
{
"query": {
"function_score": {
// 原始查询条件，搜索文档并根据相关性打分（query score）
"query": { "match": {"all":"外滩"}},
"functions": [
{
// 过滤条件，符合条件的文档才会被重新打分
"filter": {"term": {"id": "1"}},
// 算分函数，算分函数的结果称为function score,将来会与query score运算得到新算分，常见算分函数有：
// weight: 给一个常量值，作为函数结果（function score）
// field_value_factor: 用文档中某个字段作为函数结果
// random_score: 随机生成一个值，作为函数结果
// script_score: 自定义计算公式，公式结果作为函数结果
"weight": 10
}
],
// 加权模式，定义function score与query score的运算方式，包括：
// multiply: 两者相乘，默认使用
// replace: 永function score替换query score
// 其他: sum、avg、max、min
"boost_mode": "multiply"
}
}
}
# function_score查询
GET /hotel/_search
{
"query": {
"function_score": {
"query": {
"match": {
"all": "外滩"
}
},
"functions": [ // 算分函数
{
"filter": { // 满足的条件
"term": {
"brand": "如家"
}
},
"weight": 10 // 算分权重2
}
],
"boost_mode": "sum"
}
}
}- 算分函数查询，可以控制文档相关性算分，控制文档排名。
- 相关性打分算法
- 利用match查询时，文档结果会根据与搜索词的关联度打分（_score），返回结果按照分值降序排列
- TF-IDF：在es5.0之前，随着词频增加而越来越大
- BM25：在es5.0之后，随着词频增加而增大，但增长曲线会趋于水平。
- 定义的三要素
10.5.2 DSL搜索结果处理
- 排序
- 分页
- 高亮
①排序
elasticsearch支持对搜索结果排序，默认是根据相关度算分（_score）来排序。可以排序字段类型有：keyword类型、数值类型、地理坐标类型、日期类型等。
GET /indexName/_search |
②分页
elasticsearch默认情况下只返回top10的数据。而如果要查询更多数据就需要修改分页参数，elasticsearch中通过修改from、size参数控制返回的分页结果：
GET /hotel/_search |
深度分页问题：ES是分布式的，会面临深度分页问题，例如按price排序后，获取from=990，size=10的数据：
- 在ES集群中每个数据分片上都排序查询前1000条文档
- 所有节点结果聚合，在内存中重新排序选出前1000条文档
- 最后从这1000条中，选取从990开始的10条文档
注：如搜索页数过深，或者结果集（from+size）越大，对内存和CPU的消耗也越高。因此ES设定结果集查询的上限是10000，超出会报错
深度分页解决方案：
- search after：分页时需要排序，原理是从上一次排序的排序值开始，查询下一页数据。【官方推荐】
- scroll：原理将排序数据形成快照，保存在内存。【不推荐】
from+size
- 优点：支持随机反野
- 缺点：深度分页问题，默认查询上限（from+size）是10000
- 场景：百度、京东、谷歌、淘宝这样的随机反野搜索
after search
- 优点：没有查询上限（单词查询的size不超过10000）
- 缺点：只能向后逐页查询，不支持随机翻页
- 场景：没有随机翻页需求的搜索，例如手机乡下滚动翻页
scroll
- 优点：没有查询上限（单词查询的size不超过10000）
- 缺点：会有额外内存消耗，并且搜索结果是非实时的
- 场景：海量数据的获取和迁移。从ES7.1开始不推荐，建议使用after search方案
③高亮
搜索结果中把搜索关键字突出显示
原理：搜索结果关键字用标签标记出来，在页面中给标签添加css样式
GET /hotel/search |
④搜索结果处理总结
GET /hotel/_search |
10.5.3 RestClient查询文档【高级】
① match_all查询
查询基本步骤：
- 创建SearchRequest对象
- 准备Requesst.source()，也就是DSL
- QueryBuilders来构建查询条件
- 传入Request.source()的query()方法
- 发送请求，得到结果
- 解析结果（参考JSON结果，从外到内，逐层解析）
/** |
RestAPI中，其中构建DSL是通过HighLevelRestClient中的resource()来实现的，其中包含了查询、排序、分页、高亮等所有功能。
② 抽离结果处理Ctrl Alt M
private static void handleResponse(SearchResponse response) { |
③ 全文检索查询
全文检索的match和multi_match查询与match_all的API基本一致。差别是查询条件，也就是query的部分。
// 单字段查询 |
代码实现：
void testMatch() throws IOException { |
④ 精确查询
常见有term查询和range查询，同样利用QueryBuilders实现
// 词条查询 |
代码实现：
|
⑤ 复合查询
// 创建布尔查询 |
代码实现：
|
总结：要构建查询条件时，只需要认准QueryBuilders
⑥ 排序和分页
搜索结果的排序和分页是与query同级的参数，对应API如下：
// 查询 |
代码实现：
|
⑦ 高亮
高亮API包括请求DSL构建和结果解析两部分
构建
request.source().highlighter(new HighlightBuilder() |
高亮结果解析
// 解析结果 |
代码实现：
|
高亮结果解析参考JSON结果，逐层解析
应用
|
10.6. 数据聚合
聚合的分类
聚合（aggregations）可以实现对文档数据的统计、分析、运算。聚合常见的有三类:
- 桶（Bucket）聚合:用来对文档做分组
- TermAggregation：按照文档字段值分组
- Date Histogram：按照日期阶梯分组，例如一周为一组，或者一月为一组
- 度量 （Metric）聚合：用以计算一些值，比如：最大值、最小值、平均值等
- Avg：求平均值
- Max：求最大值
- Min：求最小值
- Stats：同时求max、min、avg、sum等
- 管道 （pipeline）聚合: 其它聚合的结果为基础做聚合
参与聚合的字段类型：keyword、数值、日期、布尔
10.6.1 Bucket聚合
# 聚合功能 |
聚合结果排序：默认情况下Bucket聚合会统计Bucket内的文档数量，记为_count，并且按照_count降序排序
# 聚合功能，自定义排序规则 |
限定聚合范围：默认情况下Bucket聚合对索引库所有文档做聚合，我们可以先定要聚合的文档范围，只需要增加query条件即可
# 聚合功能，限定聚合范围 |
aggs代表聚合，与query同级，此时query的作用：
- 限定聚合的的文档范围
聚合必须的三要素：
- 聚合名称
- 聚合类型
- 聚合字段
聚合可配置属性有：
- size：指定聚合结果数量
- order：指定聚合结果排序方式
- field:：指定聚合字段
10.6.2 Metrics聚合
例：获取每个品牌的用户评分的min、max、avg等值，可以利用stats聚合：
# 嵌套聚合metric |
10.6.3 RestClient实现聚合
聚合条件
request.source().aggregation(AggregationBuilders |
聚合结果解析
// 4.解析结果 |
代码实现：
/** |
10.7. 自动补全
网页搜索框自动补全
安装拼音分词器
网站：https://github.com/medcl/elasticsearch-analysis-pinyin
① 解压elasticsearch-analysis-pinyin-7.12.1.zip到挂载目录下/var/lib/docker/volumes/es-plugins/_data/py
② docker restart es 重启
③ DSL测试
POST /_analyze |
自定义分词器
elasticsearch中分词器 (analyzer) 的组成包含三部分：
- character filters：在tokenizer之前对文本进行处理。例如删除字符、替换字符
- tokenizer：将文本按照一定的规则切割成词条 (term)。例如keyword，就是不分词；还有ik_smart
- tokenizer filter：将tokenizer输出的词条做进一步处理。例如大小写转换、同义词处理、拼音处理等
使用
创建索引库时，通过settings来配置自定义的analyzer（分词器）
PUT /test |
拼音分词器适合在创建倒排索引的时候使用，但不能在搜索的时候使用。
因此，字段在创建倒排索引时应该使用my_analyzer分词器，字段在搜索时应该使用ik_smart分词器
PUT /test |
completion suggester查询
elasticsearch提供了Completion Suggester查询来实现自动补全功能。这个查询会匹配以用户输入内容开头的词条返回。为了提高补全查询的效率，对于文档中字段的类型有一些约束：
- 参与补全查询的字段必须是completion类型
- 字段的内容一般是用来补全的多个词条形成的数组。
# 创建索引库：自动补全的索引库 |
查询语法如下：
# 自动补全查询 |
自动补全对字段的要求：
- 类型是completion类型
- 字段值是多词条的数组
代码应用：
PUT /hotel |
RestAPI实现自动补全
/** |
10.8. 数据同步
elasticsearch中的酒店数据来自于mysql数据库，因此mysql数据发生改变时，elasticsearch也必须跟着改变，这个就是elasticsearch与mysgl之间的数据同步。
数据同步方式：
- 方案一：同步调用
- MySQL和ES中同时更新
- 优点：实现简单，粗暴
- 缺点：业务耦合度高
- 方案二：异步通知
- 通过MQ去监听修改消息，异步更新MySQL和ES
- 优点：低耦合，实现难度一般
- 缺点：依赖mq的可靠性
- 方案三：监听binlog
- MySQL中启动binlog，在MySQL中数据被修改会存入binlog中，使用canal中间件监听binlog，通知Consumer更新ES
- 优点：完全解除服务间耦合
- 缺点：开启binlog增加数据库负担、实现复杂度高
实现：
① Consumer和Publisher导入mq依赖
<!--mq--> |
② 修改Consumer和Publisher的yml
spring: |
③ Consumer声明队列和交换机代码
常量【Publisher也放入一份】
package cn.itcast.hotel.constants;
/**
* @version v1.0
* @auther Bamboo
* @create 2023/8/3 12:14
*/
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";
}声明
package cn.itcast.hotel.config;
import cn.itcast.hotel.constants.MqConstants;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @version v1.0
* @auther Bamboo
* @create 2023/8/3 12:19
*/
public class MqConfig {
public TopicExchange topicExchange() {
return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
}
public Queue insertQueue() {
return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
}
public Queue deleteQueue() {
return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
}
public Binding insertQueueBinding() {
return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
}
public Binding deleteQueueBinding() {
return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
}
}
④ Publisher编写Controller，注入RabbitTemplate，实现消息发送
|
⑤ Consumer完成消息监听，并更新ES数据
消息监听组件
/**
* @version v1.0
* @auther Bamboo
* @create 2023/8/3 12:43
*/
public class HotelListener {
private IHotelService hotelService;
/**
* 监听酒店新增或修改的业务
* @param id
*/
public void listenHotelInsertOrUpdate(Long id) {
hotelService.insertById(id);
}
/**
* 监听酒店删除的业务
* @param id
*/
public void listenHotelDelete(Long id) {
hotelService.deleteById(id);
}
}Service层处理方法
public class HotelService extends ServiceImpl<HotelMapper, Hotel> implements IHotelService {
private RestHighLevelClient client;
public void insertById(Long id) {
try {
Hotel hotel = getById(id);
HotelDoc hotelDoc = new HotelDoc(hotel);
// 1.准备Request
IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
// 2.准备DSL
request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
// 3.发送请求
client.index(request, RequestOptions.DEFAULT);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
public void deleteById(Long id) {
try {
// 1.准备Request
DeleteRequest request = new DeleteRequest("hotel", id.toString());
// 2.发送请求
client.delete(request, RequestOptions.DEFAULT);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
// 其余业务代码...
}
10.9. elasticsearch集群
单机的elasticsearch做数据存储，必然面临两个问题:海量数据存储问题、单点故障问题
- 海量数据存储问题: 将索引库从逻辑上拆分为N个分片 (shard)，存储到多个节点
- 单点故障问题:将分片数据在不同节点备份 (replica )
10.9.1 创建ES集群
① 上传docker-compose.yml至/root
version: '2.2' |
② 修改linux配置
修改
/etc/sysctl.conf文件：vi /etc/sysctl.conf
添加以下内容：
vm.max_map_count=262144
执行命令使其配置生效：
sysctl -p
10.9.2 集群状态监控【cerebro】
kibana可以监控es集群，不过新版本需要依赖es的x-pack功能，配置复杂
推荐使用【cerebro】来监控es集群状态，官方网址：https://github.com/lmenezes/cerebro
① windows下解压cerebro，运行bin/cerebro.bat即可
② 输入地址：http://localhost:9000进行访问
③ 选择地址http://192.168.49.10:9200进行es集群连接
④ 创建分片【索引库】：
- 选择more -> create index
- 修改名称：name : bamboo
- 设置分片数：number of shards : 3
- 设置副本数量：number of replicas : 1
- 创建：create
ES集群的节点角色
elasticsearch中集群节点有不同的职责划分：
| 节点类型 | 配置参数 | 默认值 | 节点职责 |
|---|---|---|---|
| master eligible | node.master | true | 备选主节点：主节点可以管理和记录集群状态、决定分片在哪个节点、处理创建和删除索引库的请求 |
| data | node.data | true | 数据节点：存储数据、搜索、聚合、CRUD |
| ingest | node.ingest | true | 数据存储之前的预处理 |
| coordinating | 上面3个参数都为false时，则为coordinating节点 | 无 | 路由请求到其他节点，合并其他节点处理的结果，返回给用户 |
ES集群的分布式查询
elasticsearch中的每个节点角色都有自己不同的职责，因此建议集群部署时，每个节点都有独立的角色。
User
- LB
- coordinating
- coordinating
- coordinating
- data
- data
- data
- data
- data
- ※master-eligible
- master-eligible
- master-eligible
elasticsearch的查询分成两个阶段：
- scatter phase：分散阶段，coordinating node会把请求分发到每一个分片
- gather phase：聚集阶段，coordinating node汇总data node的搜索结果，并处理为最终结果集返回给用户
ES集群的脑裂
默认情况下，每个节点都是mastereligible节点，因此一旦master节点宕机，其它候选节点会选举一个成为主节点。当主节点与其他节点网
络故障时，可能发生脑裂问题。
为了避免脑裂，需要要求选票超过（eligible节点数量+1）/2才能当选为主，因此eligible节点数量最好是奇数。
对应配置项是discovery.zen.minimum_master_nodes，在es7.0以后，已经成为默认配置，因此一般不会发生脑裂问题。
ES集群的分布式存储
当新增文档时，应该保存到不同分片，保证数据均衡，那么coordinating的elasticsearch会通过hash算法来计算文档应该存储到哪个分
片：shard = hash(_routing) % number_of_shards
说明：
- _routing默认是文档的id
- 算法与分片数量有关，因此索引库一旦创建，分片数量不能修改
ES集群的故障转移
集群的master节点会监控集群中的节点状态，如果发现有节点岩机，会立即将宕机节点的分片数据迁移到其它节点，确保数据安全，这个
叫做故障转移。
- master宕机后，EligibleMaster选举为新的主节点
- master节点监控分片、节点状态，将故障节点上的分片转移到正常节点，确保数据安全
10.9.3 Insomnia工具使用
使用HttpRequest进行发送请求
① 添加数据
方式：POST
JSON数值：
{
"title": "试着插入一条数据 id = 3"
}
② 查询数据
方式：GET
JSON数值：
{
"explain": true, # 开启后可查看数据所在分片
"query": {
"match_all": {}
}
}返回值
{
"took": 1095,
"timed_out": false,
"_shards": {
"total": 3,
"successful": 3,
"skipped": 0,
"failed": 0
},
"hits": {
"total": {
"value": 4,
"relation": "eq"
},
"max_score": 1.0,
"hits": [
{
"_shard": "[bamboo][0]", # 数据所在分片
"_node": "5mLs1IqST3GCewoZEZku9g",
"_index": "bamboo",
"_type": "_doc",
"_id": "5",
"_score": 1.0,
"_source": {
"title": "试着插入一条数据 id = 5"
},
"_explanation": {
"value": 1.0,
"description": "*:*",
"details": []
}
},
{
"_shard": "[bamboo][1]",
"_node": "nJUAL_EqSbOdm6lpFx_Y6A",
"_index": "bamboo",
"_type": "_doc",
"_id": "3",
"_score": 1.0,
"_source": {
"title": "试着插入一条数据 id = 3"
},
"_explanation": {
"value": 1.0,
"description": "*:*",
"details": []
}
},
{
"_shard": "[bamboo][2]",
"_node": "nJUAL_EqSbOdm6lpFx_Y6A",
"_index": "bamboo",
"_type": "_doc",
"_id": "1",
"_score": 1.0,
"_source": {
"title": "试着插入一条数据 id = 1"
},
"_explanation": {
"value": 1.0,
"description": "*:*",
"details": []
}
},
{
"_shard": "[bamboo][2]",
"_node": "nJUAL_EqSbOdm6lpFx_Y6A",
"_index": "bamboo",
"_type": "_doc",
"_id": "6",
"_score": 1.0,
"_source": {
"title": "试着插入一条数据 id = 6"
},
"_explanation": {
"value": 1.0,
"description": "*:*",
"details": []
}
}
]
}
}
十一、Sentinel【微服务保护】
雪崩问题 微服务调用链路中的某个服务故障，引起整个链路中的所有微服务都不可用，这就是雪崩。
解决雪崩问题的常见方式有四种：
超时处理【服务故障而引发的雪崩】：设定超时时间，请求超过一定时间没有响应就返回错误信息，不会无休止等待。
舱壁模式【服务故障而引发的雪崩】：限定每个业务能使用的线程数，避免耗尽整个tomcat的资源，因此也叫线程隔离。
熔断降级【服务故障而引发的雪崩】：由断路器统计业务执行的异常比例，如果超出阈值则会熔断该业务，拦截访问该业务的一切请求。
流量控制【避免因瞬间高并发流量而导致服务故障】：限制业务访问的QPS，避免服务因流量的突增而故障。
服务保护技术对比
| Sentinel | Hystrix | |
|---|---|---|
| 隔离策略 | 信号量隔离 | 线程池隔离/信号量隔离 |
| 熔断降级策略 | 基于慢调用比例或异常比例 | 基于失败比例 |
| 实施指标实现 | 滑动窗口 | 滑动窗口（基于 RxJava） |
| 规则配置 | 支持多种数据源 | 支持多种数据源 |
| 扩展性 | 多个扩展点 | 插件的形式 |
| 基于注解的支持 | 支持 | 支持 |
| 限流 | 基于QPS，支持基于调用关系的限流 | 有限的支持 |
| 流量整形 | 支持慢启动、匀速排队模式 | 不支持 |
| 系统自适应保护 | 支持 | 不支持 |
| 控制台 | 开箱即用，可配置规则、查看秒级监控、机器发现等 | 不完善 |
| 常见框架的适配 | Servlet、Spring Cloud、Dubbo、gRPC等 | Servlet、Spring Cloud Netflix |
认识Sentinel Sentinel是阿里巴巴开源的一款微服务流量控制组件。官网地址: https://sentinelguard.io/zh-cn/index.html
Sentinel具有以下特征:
- 丰富的应用场景：Sentinel承接了阿里巴巴近10年的双十一大促流量的核心场景，例如秒杀(即突发流量控制在系统容量可以承受的范围)、消息削峰填谷、集群流量控制、实时熔断下游不可用应用等
- 完备的实时监控：Sentinel同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据，甚至500台以下规模的集群的汇总运行情况。
- 广泛的开源生态：Sentinel提供开箱即用的与其它开源框架/库的整合模块，例如与 Spring Cloud、Dubbo、gRPC的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。
- 完善的 SPI 扩展点：Sentinel提供简单易用、完善的 SPI 扩展接口。您可以通过实现扩展接口来快速地定制逻辑。例如定制规则管理、适配动态数据源等。
安装Sentinel控制台 sentinel官方提供了UI控制台，方便我们对系统做限流设置。可以在GitHub下载。【sentinel-dashboard-1.8.1.jar】
- 将其拷贝到一个非中文目录，然后运行命令:
java -jar sentinel-dashboard-1.8.1.jar - 然后访问：
localhost:8080即可看到控制台页面，默认的账户和密码都是sentinel
| 配置项 | 默认值 | 说明 |
|---|---|---|
| server.port | 8080 | 服务端口 |
| sentinel.dashboard.auth.username | sentinel | 默认用户名 |
| sentinel.dashboard.auth.password | sentinel | 默认密码 |
举例：java -jar sentinel-dashboard-1.8.1.jar -Dserver.port=8090
应用sentinel
结合微服务使用Sentinel，使用一个SpringCloud工厂
项目结构：
- gateway
- user-uservice【Publisher】：用户服务，包含用户CRUD
- order-service【Consumer】：订单服务，调用user-service
- feign-api：用户服务对外暴露的feign客户端、实体类
① Consumer引入依赖
<!--引入sentinel依赖--> |
② Consumer配置yml
spring: |
③ 访问微服务任意端点，触发sentinel监控
11.1. 限流规则
11.1.1 簇点链路
簇点链路：就是项目内的调用链路，链路中被监控的每个接口就是一个资源。默认情况下sentinel会监控SpringMVC的每一个端点 (Endpoint)，因此SpringMVC的每一个端点 (Endpoint)就是调用链路中的一个资源。
流控、熔断等都是针对簇点链路中的资源来设置的，因此我们可以点击对应资源后面的按钮来设置规则:
- 簇点链路 =》接口=》流控
11.1.2 JMeter测试
利用jmeter测试服务器压力
11.1.3 流控模式
在添加限流规则时，点击高级选项，可以选择三种流控模式：
- 直接：统计当前资源的请求，触发阈值时对当前资源直接限流，也是默认的模式
- 关联：统计与当前资源相关的另一个资源，触发阈值时，对当前资源限流
- 链路：统计从指定链路访问到本资源的请求，触发阈值时，对指定链路限流
① 流控模式——关联
关联模式：统计与当前资源相关的另一个资源，触发阈值时，对当前资源限流 使用场景：比如用户支付时需要修改订单状态，同时用户要查询订单。查询和修改操作会争抢数据库锁，产生竞争。业务需求是有限支付和更新订单的业务，因此当修改订单业务触发闻值时，需要对查询订单业务限流。
例：配置资源名/read，关联资源/write，则当/write访问量触发阈值时，就会对/read资源限流
案例：
- 在OrderController中设置
/order/query和/order/update - 配置流控规则，当
/order/update资源被访问的QPS超过5时，对/order/query请求限流
② 流控模式——链路
链路模式：只针对从指定链路访问到本资源的请求做统计，判断是否超过阈值。例如有两条请求链路。
- /test1 =》/common
- /test2 =》/common
如果只希望统计从/test2进入到/common的请求，则可以这样配置：
例：配置流控规则，设置资源名为/common，流控模式为链路，入口资源为/test2，这样就只会拦截test2
案例：
Sentinel默认只标记Controller中的方法为资源，如果要标记其它方法，需要利用@SentinelResource注解，示例：
// service层代码
public void queryGoods(){
System.err.println("查询商品");
}Sentinel默认会将Controller方法做context整合，导致链路模式的流控失效，需要修改application.yml，添加配置：
spring:
cloud:
sentinel:
web-context-unify: false # 关闭context整合Controller中的映射直接调用被标注@SentinelResource注解的方法即可
11.1.4 流控效果
流控效果是指请求达到流控阈值时应该采取的措施，包括三种：
- 快速失败：达到阈值后，新的请求会被立即拒绝并抛出FIowException异常。是默认的处理方式。
- warm up：预热模式，对超出阈值的请求同样是拒绝并抛出异常。但这种模式阈值会动态变化，从一个较小值逐渐增加到最大阈值。
- 排队等待：让所有的请求按照先后次序排队执行，两个请求的间隔不能小于指定时长
① 流控效果-warm up
warmup也叫预热模式，是应对服务冷启动的一种方案。请求阈值初始值是 threshold/coldFactor，持续指定时长后，逐渐提高到threshold值。而coldFactor的默认值是3。
例如，我设置QPS的threshold为10，预热时间为5秒，那么初始阈值就是 10/3，也就是3，然后在5秒后逐渐增长到10。
② 流控效果-排队等待
当请求超过QPS阈值时，快速失败和warm up 会拒绝新的请求并抛出异常。而排队等待则是让所有请求进入一个队列中，然后按照阈值允许的时间间隔依次执行。后来的请求必须等待前面执行完成，如果请求预期的等待时间超出最大时长，则会被拒绝。
例如:QPS=5，意味着每200ms处理一个队列中的请求，timeout= 2000，意味着预期等待超过2000ms的请求会被拒绝并抛出异常。
11.1.5 热点参数限流
之前的限流是统计访问某个资源的所有请求，判断是否超过QPS阈值。而热点参数限流是分别统计参数值相同的请求判断是否超过QPS阈值。
例：
- 请求参数：
id = 1id = 1id = 1id = 1
- QPS
- 3 :
id = 1 - 1：
id = 2
- 3 :
配置：参数索引设置0，单机阈值设置5，统计窗口时长设置1
高级选项：可以对部分参数设置例外配置
- 参数类型
- 参数值
- 限流阈值
注：热点参数限流对默认SpringMVC资源无效，需要在controller层的映射中增加@SentinelResource(“hot”)注解
应用：
|
配置Sentinel：
- 热点规则 = 》
- 资源名hot，参数索引0，单机阈值2，统计窗口时长1
- 参数例外项：参数类型long，参数值：102/4和103/10
11.2. 隔离和降级
虽然限流可以尽量避免因高并发而引起的服务故障，但服务还会因为其它原因而故障。而要将这些故障控制在一定范围避免雪崩，就要靠线程隔离(舱壁模式)和熔断降级手段了。
不管是线程隔离还是熔断降级，都是对**客户端 (调用方)**的保护。
11.2.1 Feign整合Sentinel
SpringCloud中，微服务调用都是通过Feign来实现的，因此做客户端保护必须整合Feign和Sentinel。
修改OrderService【Consumer】的application.yml文件，开启Feign的Sentinel功能
feign:
sentinel:
enabled: true # 开启feign对sentinel的支持给FeignClient编写失败后的降级逻辑
方式一：FallbackClass，无法对远程调用的异常做处理
方式二：FallbackFactory，可以对远程调用的异常做处理【推荐】
① 在feign-api中定义类，实现
FallBackFactory接口：
public class UserClientFallbackFactory implements FallbackFactory<UserClient> {
public UserClient create(Throwable throwable) {
return new UserClient() {
public User findById(Long id) {
log.error("查询用户异常", throwable);
return new User();
}
};
}
}② 在feign-api项目中的DefaultFeignConfiguration类中将UserClientFallbackFactory注册为一个Bean：
public class DefaultFeignConfiguration {
public Logger.Level logLevel(){
return Logger.Level.BASIC;
}
public UserClientFallbackFactory userClientFallbackFactory(){
return new UserClientFallbackFactory();
}
}③ 在feign-api中的UserClient接口中使用UserClientFallbackFactory：
public interface UserClient {
User findById( Long id);
}
11.2.2 线程隔离
线程隔离有两种方式实现：
- 线程池隔离
- 优点：
- 支持主动超时
- 支持异步调用
- 缺点：线程的额外开销比较大
- 场景：低扇出
- 优点：
- 信号量隔离（Sentinel默认采用）
- 优点：轻量级，无额外开销
- 缺点：
- 不支持主动超时
- 不支持异步调用
- 场景
- 高频调用
- 高扇出
① 线程隔离 (舱壁模式)
在添加限流规则时，可以选择两种阈值类型：
- OPS：就是每秒的请求数。
- 线程数：是该资源能使用的tomcat线程数的最大值。也就是通过限制线程数量，实现舱壁模式。
11.2.3 熔断降级
熔断降级是解决雪崩问题的重要手段。其思路是由断路器统计服务调用的异常比例、慢请求比例，如果超出阈值则会熔断该服务。即拦截访问该服务的一切请求；而当服务恢复时，断路器会放行访问该服务的请求。
① 熔断策略——慢调用
慢调用：业务的响应时长(RT response time)大于指定时长的请求认定为慢调用请求。在指定时间内，如果请求数量超过设定的最小数量，慢调用比例大于设定的阈值，则触发熔断。例如：
- 熔断策略：慢调用比例
- 最大RT：500ms
- 比例阈值：0.5
- 熔断时长：5s
- 最小请求数：10
- 统计时长：10000ms
解读：RT超过500ms的调用是慢调用，统计最近10000ms内的请求，如果请求量超过10次，并且慢调用比例不低于0.5则触发熔断，熔断时长为5秒。然后进入half-open状态，放行一次请求做测试。
② 熔断策略——异常比例、异常数
异常比例或异常数：统计指定时间内的调用，如果调用次数超过指定请求数，并且出现异常的比例达到设定的比例阈值(或超过指定异常数)，则触发熔断。例如:
- 异常比例策略
- 熔断策略：异常比例
- 比例阈值：0.4
- 熔断时长：5s
- 最小请求数：10
- 统计时长1000s
- 异常数策略
- 熔断策略：异常数
- 异常数：2
- 熔断时长：5s
- 最小请求数：10
- 统计时长1000s
解读：统计最近1000ms内的请求，如果请求量超过10次，并且异常比例不低于0.5，则触发熔断，熔断时长为5秒。然后进入half-open状态，放行一次请求做测试.
11.3. 授权规则
授权规则可以对调用方的来源做控制，有白名单和黑名单两种方式。
- 白名单：来源 (origin)在白名单内的调用者允许访问
- 黑名单：来源(origin)在黑名单内的调用者不允许访问
主要是配合网关gateway进行实现。
11.3.1 网关授权
Sentinel是通过RequestOriginParser这个接口的parseOrigin来获取请求的来源的。
public interface RequestOriginParser { |
例：从var1中获取一个名为origin的请求头，作为origin的值
① 在Consumer中添加此类
package cn.itcast.order.sentinel; |
② 在gateway服务中，利用网关的过滤器添加名为gateway的origin头
spring: |
③ 在sentinel给/order/{orderId}配置授权规则
- 资源名：
/order/{orderId} - 流控应用：gateway
- 授权类型：白名单
11.3.2 自定义异常结果
默认情况下，发生限流、降级、授权拦截时，都会抛出异常到调用放。自定义异常时返回结果，需要实现BlockExceptionHandler接口
public interface BlockExceptionHandler { |
| 异常 | 说明 |
|---|---|
| FlowException | 限流异常 |
| ParamFlowException | 热点参数限流的异常 |
| DegradeException | 降级异常 |
| AuthorityException | 授权规则异常 |
| SystemBlockException | 系统规则异常 |
实现：在Consumer服务者进行定义
package cn.itcast.order.sentinel; |
11.4. 规则持久化
11.4.1 规则管理模式
Sentinel的控制台规则管理有三种模式：
- 原始模式：Sentinel的默认模式，将规则保存在内存，重启服务会丢失
- pull模式：控制台将配置的规则推送到Sentinel客户端，而客户端会将配置规则保存在本地文件或数据库中。以后会定时去本地文件或数据库中查询，更新本地规则。
- push模式【推荐】：控制台将配置规则推送到远程配置中心，例如Nacos。Sentinel客户端监听Nacos，获取配置变更的推送消息，完成本地配置更新。
11.4.2 push模式应用
（1）修改OrderService，让其监听Nacos中的sentinel规则配置。
① 引入依赖
在order-service中引入sentinel监听nacos的依赖：
<dependency> |
② 配置nacos地址
在order-service中的application.yml文件配置nacos地址及监听的配置信息：
spring: |
（2）修改sentinel-dashboard源码
① IDEA打开sentinel.zip解压后的项目
② 修改nacos依赖
在sentinel-dashboard源码的pom文件中，nacos的依赖默认的scope是test，只能在测试时使用，这里要去除
<dependency> |
③ 添加nacos依赖
在sentinel-dashboard的test包下，已经编写了对nacos的支持，我们需要将其拷贝到main下。
- 将maven中
src\main\test下的com/alibaba/csp/sentinel/dashboard/rule/nacos包内所有文件移入src\main\java中
④ 修改nacos地址
修改测试代码中的NacosConfig类，修改其中的nacos地址，让其读取application.properties中的配置
package com.alibaba.csp.sentinel.dashboard.rule.nacos;
/**
* @author Eric Zhao
* @since 1.4.0
*/
public class NacosConfig {
// nacos地址
private String addr;
public Converter<List<FlowRuleEntity>, String> flowRuleEntityEncoder() {
return JSON::toJSONString;
}
public Converter<String, List<FlowRuleEntity>> flowRuleEntityDecoder() {
return s -> JSON.parseArray(s, FlowRuleEntity.class);
}
public ConfigService nacosConfigService() throws Exception {
return ConfigFactory.createConfigService(addr);
}
public String getAddr() {
return addr;
}
public void setAddr(String addr) {
this.addr = addr;
}
}在sentinel-dashboard的application.properties中添加nacos地址配置
nacos.addr=localhost:8848
⑤ 配置nacos数据源
修改com.alibaba.csp.sentinel.dashboard.controller.v2包下的FlowControllerV2类
|
⑥ 修改前端页面
修改前端页面，添加一个支持nacos的菜单
修改src/main/webapp/resources/app/scripts/directives/sidebar/目录下的sidebar.html文件
<!--将以下注释部分打开并修改--> |
⑦ 重新编译打包Sentinel-Dashboard模块
- 选择Maven中Toggle ‘Skip Tests’ Mode，去掉测试模块
- 运行package打包
⑧ 启动
启动方式跟官方一样：
java -jar sentinel-dashboard.jar |
如果要修改nacos地址，需要添加参数：
java -jar -Dnacos.addr=localhost:8848 sentinel-dashboard.jar |
十二、Seata【分布式事务】
事务的ACID原则：
- 原子性：事务中的所有操作，要么全部成功，要么全部失败
- 一致性：要保证数据库内部完全性约束、声明性约束
- 隔离性：对同一资源操作的事务不能同时发生
- 持久性：对数据库做的一切修改将永久保存，不管是否出现故障
分布式服务的事务问题
在分布式系统下，一个业务跨越多个服务或数据源，每个服务都是一个分支事务，要保证所有分支事务最终状态一致，这样的事务就是分布式事务。
12.1. 理论基础
12.1.1 CAP定理
1998年，加州大学的计算机科学家 Eric Brewer 提出，分布式系统有三个指标：
- Consistency(一致性)：用户访问分布式系统中的任意节点，得到的数据必须一致。
- Availability (可用性)：用户访问集群中的任意健康节点，必须能得到响应，而不是超时或拒绝。
- Partition tolerance (分区容错性)
- Partition（分区）：因为网络故障或其他原因导致分布式系统中的部分节点与其他节点失去连接，形成独立分区。
- Tolerance（容错）：在集群出现分区时，整个系统也要持续对外提供服务。
Eric Brewer 说，分布式系统无法同时满足这三个指标。
这个结论就叫做 CAP 定理。
结论
- 分布式系统节点通过网络连接，一定会出现分区问题（P）
- 当分区出现时，系统的一致性（C）和可用性（A）就无法同时满足
- ES集群出现分区时，故障节点会被剔除集群，数据分片会重新分配到其他节点，保证数据一致。因此是低可用性，高一致性，属于CP
12.1.2 BASE理论
BASE理论是对CAP的一种解决思路，包含三个思想：
- Basically Available (基本可用)：分布式系统在出现故障时，允许损失部分可用性，即保证核心可用。
- Soft state (软状态)：在一定时间内，允许出现中间状态，比如临时的不一致状态。
- Eventually Consistent(最终一致性)：虽然无法保证强一致性，但是在软状态结束后，最终达到数据一致。
而分布式事务最大的问题是各个子事务的一致性问题，因此可以借鉴CAP定理和BASE理论：
- AP模式：各子事务分别执行和提交，允许出现结果不一致，然后采用弥补措施恢复数据即可，实现最终一致。
- CP模式：各个子事务执行后互相等待，同时提交，同时回滚，达成强一致。但事务等待过程中，处于弱可用状态。
分布式事务模型：
解决分布式事务，各个子系统之间必须能感知到彼此的事务状态，才能保证状态一致，因此需要一个事务协调者来协调每一个事务的参与者(子系统事务)。 这里的子系统事务，称为分支事务；有关联的各个分支事务在一起称为全局事务
12.2. Seata搭建
Seata是 2019年1月份蚂蚁金服和阿里巴巴共同开源的分布式事务解决方案。致力于提供高性能和简单易用的分布式事务服务，为用户打造一站式的分布式解决方案。
官网地址:http://seata.io/，其中的文档、播客中提供了大量的使用说明、源码分析。
seata架构 Seata事务管理中有三个重要的角色：
- TC(Transaction Coordinator) - 事务协调者：维护全局和分支事务的状态，协调全局事务提交或回滚。
- TM(Transaction Manaer) - 事务管理器：定义全局事务的范围、开始全局事务、提交或回滚全局事务。
- RM(Resource Manager) - 资源管理器：管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。
Seata提供了四种不同的分布式事务解决方案
- XA模式：强一致性分阶段事务模式，牺牲了一定的可用性，无业务侵入。
- TCC模式：最终一致的分阶段事务模式，有业务侵入。
- AT模式：最终一致的分阶段事务模式，无业务侵入，也是Seata的默认模式。
- SAGA模式：长事务模式，有业务侵入。
12.2.1 部署TC服务
① 下载seata-server包
官方地址：http://seata.io/zh-cn/blog/download.html
② 解压
在非中文目录下解压，目录结构：
- bin：运行脚本
- conf：配置文件
- lib：依赖库
③ 修改配置
修改conf目录下的registry.conf
registry { |
④ 在nacos中添加配置
Data ID：seataServer.properties
Group：DEFAULT_GROUP
配置格式：Properties
配置内容：
# 数据存储方式，db代表数据库
store.mode=db
store.db.datasource=druid
store.db.dbType=mysql
store.db.driverClassName=com.mysql.jdbc.Driver
store.db.url=jdbc:mysql://127.0.0.1:3306/seata?useUnicode=true&rewriteBatchedStatements=true
store.db.user=root
store.db.password=root
store.db.minConn=5
store.db.maxConn=30
store.db.globalTable=global_table
store.db.branchTable=branch_table
store.db.queryLimit=100
store.db.lockTable=lock_table
store.db.maxWait=5000
# 事务、日志等配置
server.recovery.committingRetryPeriod=1000
server.recovery.asynCommittingRetryPeriod=1000
server.recovery.rollbackingRetryPeriod=1000
server.recovery.timeoutRetryPeriod=1000
server.maxCommitRetryTimeout=-1
server.maxRollbackRetryTimeout=-1
server.rollbackRetryTimeoutUnlockEnable=false
server.undo.logSaveDays=7
server.undo.logDeletePeriod=86400000
# 客户端与服务端传输方式
transport.serialization=seata
transport.compressor=none
# 关闭metrics功能，提高性能
metrics.enabled=false
metrics.registryType=compact
metrics.exporterList=prometheus
metrics.exporterPrometheusPort=9898
⑤ 创建数据库表
特别注意：tc服务在管理分布式事务时，需要记录事务相关数据到数据库中，你需要提前创建好这些表。
新建一个名为seata的数据库
SET NAMES utf8mb4; |
⑥ 启动TC服务
进入bin目录，运行其中的seata-server.bat即可
打开浏览器，访问nacos地址：http://localhost:8848，然后进入服务列表页面，可以看到seata-tc-server的信息。
12.2.2 集成seata
① 在微服务中引入seata依赖：
<dependency> |
② 修改yml配置文件
seata: |
总结
nacos服务名称组成包括：
- namespace
- group
- serviceName
- cluster
seata客户端获取tc的cluster名称方式
- 以tx-group-service的值为key到vgroupMapping中查找
12.3. Seata解决方案
| XA | AT | TCC | SAGA | |
|---|---|---|---|---|
| 一致性 | 强一致 | 弱一致 | 弱一致 | 最终一致 |
| 隔离性 | 完全隔离 | 基于全局锁隔离 | 基于资源预留隔离 | 无隔离 |
| 代码侵入 | 无 | 无 | 有，需要编写三个接口 | 有，需要编写状态机和补偿业务 |
| 性能 | 差 | 好 | 非常好 | 非常好 |
| 场景 | 对一致性、隔离性有高要求的业务 | 基于关系型数据库的大多数分布式事务场景都可以 | 对性能要求较高的事务；有非关系型数据库要参与的事务 | 业务流程长、业务流程多；参与者包含其他公司或遗留系统服务，无法提供TCC模式要求的三个接口 |
12.3.1 XA模式
原理：XA规范是X/Open 组织定义的分布式事务处理 (DTP，Distributed Transaction Processing)标准，XA 规范描述了全局的TM与局部的RM之间的接口，几平所有主流的数据库都对XA规范提供了支持。
① XA模式共分为两阶段运行
- 第一阶段：事务协调者与RM进行交互
- 事务协调者对RM进行prepare准备
- RM返回给事务协调者ready就绪或fail失败
- 第二阶段：事务协调者再次与RM交互
- 提交或回滚事务
② seata的XA模式做了一些调整
- RM一阶段的工作：
- 注册分支事务到TC
- 执行分支业务SQL但不提交
- 报告执行状态到TC
- TC二阶段的工作【TC检测各分支事务执行状态】：
- 都成功，通知所有RM提交事务
- 有失败，通知所有RM回滚事务
- RM二阶段的工作：
- 接收TC指令，提交或回滚事务
③ XA模式优缺点
XA模式的优点：
- 事务的强一致性，满足ACID原则。
- 常用数据库都支持，实现简单，并且没有代码侵入。
XA模式的缺点：
- 因为一阶段需要锁定数据库资源，等待二阶段结束才释放，性能较差。
- 依赖关系型数据库实现事务。
④ 实现XA模式
修改yml配置文件（每个参与事务的微服务），开启XA模式：
seata:
data-source-proxy-mode: XA # 开启数据源代理的XA模式给发起全局事务的入口方法添加@GlobalTransactional注解
public class OrderServiceImpl implements OrderService {
private final AccountClient accountClient;
private final StorageClient storageClient;
private final OrderMapper orderMapper;
public OrderServiceImpl(AccountClient accountClient, StorageClient storageClient, OrderMapper orderMapper) {
this.accountClient = accountClient;
this.storageClient = storageClient;
this.orderMapper = orderMapper;
}
// 开启XA模式事务
public Long create(Order order) {
// 创建订单
orderMapper.insert(order);
try {
// 扣用户余额
accountClient.deduct(order.getUserId(), order.getMoney());
// 扣库存
storageClient.deduct(order.getCommodityCode(), order.getCount());
} catch (FeignException e) {
log.error("下单失败，原因:{}", e.contentUTF8(), e);
throw new RuntimeException(e.contentUTF8(), e);
}
return order.getId();
}
}
12.3.2 AT模式
原理：AT模式同样是分阶段提交的事务模型，不过缺弥补了XA模型中资源锁定周期过长的缺陷。
阶段一RM的工作：
- 注册分支事务
- 记录undo-log (数据快照)
- 执行业务sql并提交
- 报告事务状态
阶段二提交时RM的工作：
- 删除undo-log即可
阶段二回滚时RM的工作：
- 根据undo-log恢复数据到更新前
① 简述AT模式与XA模式最大的区别
- XA模式一阶段不提交事务，锁定资源；AT模式一阶段直接提交，不锁定资源。
- XA模式依赖数据库机制实现回滚；AT模式利用数据快照实现数据回滚。
- XA模式强一致；AT模式最终一致。
② AT模式脏写问题
多个事务执行时触发的快照问题，无法回滚到正确的快照。
AT模式的写隔离：使用全局锁解决脏写问题
- 全局锁：由TC记录当前正在操作某行数据的事务，该事务持有全局锁，具备执行权。
③ AT模式优缺点
AT模式的优点：
- 一阶段完成直接提交事务，释放数据库资源，性能比较好利用全局锁实现读写隔离。
- 没有代码侵入，框架自动完成回滚和提交。
AT模式的缺点：
- 两阶段之间属于软状态，属于最终一致。
- 框架的快照功能会影响性能，但比XA模式要好很多。
④ 实现AT模式
AT模式中的快照生成、回滚动作都是由框架自动完成，没有任何代码侵入
导入sql文件
lock_table导入到TC服务关联的数据库
-- ----------------------------
-- Table structure for lock_table
-- ----------------------------
DROP TABLE IF EXISTS `lock_table`;
CREATE TABLE `lock_table` (
`row_key` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
`xid` varchar(96) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`transaction_id` bigint(20) NULL DEFAULT NULL,
`branch_id` bigint(20) NOT NULL,
`resource_id` varchar(256) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`table_name` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`pk` varchar(36) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
`gmt_create` datetime NULL DEFAULT NULL,
`gmt_modified` datetime NULL DEFAULT NULL,
PRIMARY KEY (`row_key`) USING BTREE,
INDEX `idx_branch_id`(`branch_id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;undo_log表导入到微服务关联的数据库
-- ----------------------------
-- Table structure for undo_log
-- ----------------------------
DROP TABLE IF EXISTS `undo_log`;
CREATE TABLE `undo_log` (
`branch_id` bigint(20) NOT NULL COMMENT 'branch transaction id',
`xid` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'global transaction id',
`context` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'undo_log context,such as serialization',
`rollback_info` longblob NOT NULL COMMENT 'rollback info',
`log_status` int(11) NOT NULL COMMENT '0:normal status,1:defense status',
`log_created` datetime(6) NOT NULL COMMENT 'create datetime',
`log_modified` datetime(6) NOT NULL COMMENT 'modify datetime',
UNIQUE INDEX `ux_undo_log`(`xid`, `branch_id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = 'AT transaction mode undo table' ROW_FORMAT = Compact;
-- ----------------------------
-- Records of undo_log
-- ----------------------------
修改yml配置文件，将事务模式修改为AT模式
seata:
data-source-proxy-mode: AT # 开启数据源代理的AT模式
12.3.3 TCC模式
TCC模式与AT模式非常相似，每阶段都是独立事务，不同的是TCC通过人工编码来实现数据恢复。需要实现三个方法：
- Try：资源的检测和预留。
- Confirm：完成资源操作业务;要求Try 成功 Confirm一定要能成功。
- Cancel：预留资源释放，可以理解为try的反向操作。
① 作用及其优缺点
TCC模式的每个阶段的作用
- Try：资源检查和预留
- Confirm：业务执行和提交
- Cancel：预留资源的释放
TCC的优点
- 一阶段完成直接提交事务，释放数据库资源，性能好
- 相比AT模型，无需生成快照，无需使用全局锁，性能最强
- 不依赖数据库事务，而是依赖补偿操作，可以用于非事务型数据库
TCC的缺点
- 有代码侵入，需要人为编写try、Confirm和Cancel接口，太麻烦
- 软状态，事务是最终一致
- 需要考虑Confirm和Cancel的失败情况，做好幂等处理
② 应用案例
需求：改造account-service服务，利用TCC实现分布式事务
- 修改account-service，编写try、confirm、cancel逻辑
- try业务：添加冻结金额，扣减可用金额
- confirm业务：删除冻结金额
- cancel业务：删除冻结金额，恢复可用金额
- 保证confirm、cancel接口的幂等性
- 允许空回滚
- 拒绝业务悬挂
业务分析：为了实现空回滚、防止业务悬挂，以及幂等性要求。我们必须在数据库记录冻结金额的同时，记录当前事务id和执行状态，为此设计了一张表，将此表插入微服务数据库中：
-- ---------------------------- |
- 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业务
实现代码：
cn.itcast.account.service.AccountTCCService
public interface AccountTCCService {
void deduct( String userId,
int money);
boolean confirm(BusinessActionContext ctx);
boolean cancel(BusinessActionContext ctx);
}cn.itcast.account.service.impl.AccountTCCServiceImpl/**
* @version v1.0
* @auther Bamboo
* @create 2023/8/7 8:29
*/
public class AccountTCCServiceImpl implements AccountTCCService {
private AccountMapper accountMapper;
private AccountFreezeMapper freezeMapper;
public void deduct(String userId, int money) {
// 0.获取事务id
String xid = RootContext.getXID();
// 1.业务悬挂判断：判断freeze中是否由冻结记录，如果有，一定是CANCEL执行过，我要拒绝业务
AccountFreeze oldFreeze = freezeMapper.selectById(xid);
if (oldFreeze != null) {
// CANCEL执行过，拒绝业务
return;
}
// 2.扣减可用金额
accountMapper.deduct(userId, money);
// 3.记录冻结金额，事务状态
AccountFreeze freeze = new AccountFreeze();
freeze.setUserId(userId);
freeze.setFreezeMoney(money);
freeze.setState(AccountFreeze.State.TRY);
freeze.setXid(xid);
freezeMapper.insert(freeze);
}
public boolean confirm(BusinessActionContext ctx) {
// 1.获取事务id
String xid = ctx.getXid();
// 2.根据id删除冻结记录
int count = freezeMapper.deleteById(xid);
return count == 1;
}
public boolean cancel(BusinessActionContext ctx) {
// 0.查询冻结金额
String xid = ctx.getXid();
String userId = ctx.getActionContext("userId").toString();
AccountFreeze freeze = freezeMapper.selectById(xid);
// 2.空回滚判断，判断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);
}
// 3. 幂等判断处理
if (freeze.getState() == AccountFreeze.State.CANCEL) {
// 已经处理过一次CANCEL了，无需重复处理
return true;
}
// 4.恢复可用余额
accountMapper.refund(freeze.getUserId(), freeze.getFreezeMoney());
// 5.将冻结金额清零，状态改为CANCEL
freeze.setFreezeMoney(0);
freeze.setFreezeMoney(AccountFreeze.State.CANCEL);
int count = freezeMapper.updateById(freeze);
return count == 1;
}
}替换controller层注入
public class AccountController {
private AccountTCCService accountService;
public ResponseEntity<Void> deduct( String userId, Integer money){
accountService.deduct(userId, money);
return ResponseEntity.noContent().build();
}
}依赖mapper
AccountFreezeMapper.java
public interface AccountFreezeMapper extends BaseMapper<AccountFreeze> {
}AccountMapper.java
public interface AccountMapper extends BaseMapper<Account> {
int deduct( String userId, int money);
int refund( String userId, int money);
}
依赖pojo
public class AccountFreeze {
private String xid;
private String userId;
private Integer freezeMoney;
private Integer state;
public static abstract class State {
public final static int TRY = 0;
public final static int CONFIRM = 1;
public final static int CANCEL = 2;
}
}
③ TCC的空回滚和业务悬挂
当某分支事务的try阶段阻塞时，可能导致全局事务超时而触发二阶段的cancel操作。在未执行try操作时先执行了cancel操作，这时cancel不能做回滚，就是空回滚。
对于已经空回滚的业务，如果以后继续执行try，就永远不可能confirm或cancel，这就是业务悬挂。应当阻止执行空回滚后的try操作，避免悬挂。
④ 声明TCC接口
TCC的Try、Confirm、Cancel方法都需要在接口中基于注解来声明，语法如下：
|
3.4 SAGA模式
Saga模式是SEATA提供的长事务解决方案。也分为两个阶段：
- 一阶段：直接提交本地事务
- 二阶段：成功则什么都不做；失败则通过编写补偿业务来回滚
优点：
- 事务参与者可以基于事件驱动实现异步调用，吞吐高
- 一阶段直接提交事务，无锁，性能好
- 不用编写TCC中的三个阶段，实现简单
缺点：
- 软状态持续时间不确定，时效性差
- 没有锁，没有事务隔离，会有脏写
12.4. TC服务的高可用和异地容灾
12.4.1 模拟异地容灾的TC集群
计划启动两台seata的tc服务节点：
| 节点名称 | ip地址 | 端口号 | 集群名称 |
|---|---|---|---|
| seata | 127.0.0.1 | 8091 | SH |
| seata2 | 127.0.0.1 | 8092 | HZ |
之前我们已经启动了一台seata服务，端口是8091，集群名为SH。
现在，将seata目录复制一份，起名为seata2
修改seata2/conf/registry.conf内容如下：
registry { |
进入seata2/bin目录，然后运行命令：
seata-server.bat -p 8092 |
12.4.2 将事务组映射配置到nacos
接下来，我们需要将tx-service-group与cluster的映射关系都配置到nacos配置中心。
新建一个配置：
Data ID：client.properties
Group：SEATA_GROUP
配置格式：properties
配置内容：
# 事务组映射关系
service.vgroupMapping.seata-demo=SH
service.enableDegrade=false
service.disableGlobalTransaction=false
# 与TC服务的通信配置
transport.type=TCP
transport.server=NIO
transport.heartbeat=true
transport.enableClientBatchSendRequest=false
transport.threadFactory.bossThreadPrefix=NettyBoss
transport.threadFactory.workerThreadPrefix=NettyServerNIOWorker
transport.threadFactory.serverExecutorThreadPrefix=NettyServerBizHandler
transport.threadFactory.shareBossWorker=false
transport.threadFactory.clientSelectorThreadPrefix=NettyClientSelector
transport.threadFactory.clientSelectorThreadSize=1
transport.threadFactory.clientWorkerThreadPrefix=NettyClientWorkerThread
transport.threadFactory.bossThreadSize=1
transport.threadFactory.workerThreadSize=default
transport.shutdown.wait=3
# RM配置
client.rm.asyncCommitBufferLimit=10000
client.rm.lock.retryInterval=10
client.rm.lock.retryTimes=30
client.rm.lock.retryPolicyBranchRollbackOnConflict=true
client.rm.reportRetryCount=5
client.rm.tableMetaCheckEnable=false
client.rm.tableMetaCheckerInterval=60000
client.rm.sqlParserType=druid
client.rm.reportSuccessEnable=false
client.rm.sagaBranchRegisterEnable=false
# TM配置
client.tm.commitRetryCount=5
client.tm.rollbackRetryCount=5
client.tm.defaultGlobalTransactionTimeout=60000
client.tm.degradeCheck=false
client.tm.degradeCheckAllowTimes=10
client.tm.degradeCheckPeriod=2000
# undo日志配置
client.undo.dataValidation=true
client.undo.logSerialization=jackson
client.undo.onlyCareUpdateColumns=true
client.undo.logTable=undo_log
client.undo.compress.enable=true
client.undo.compress.type=zip
client.undo.compress.threshold=64k
client.log.exceptionRate=100
12.4.3 微服务读取nacos配置
接下来，需要修改每一个微服务的application.yml文件，让微服务读取nacos中的client.properties文件：
seata: |
重启微服务，现在微服务到底是连接tc的SH集群，还是tc的HZ集群，都统一由nacos的client.properties来决定了。
十三、Redis缓存【分布式缓存】
基于Redis集群解决单机Redis存在四大的问题
- 数据丢失问题
- 实现Redis数据持久化
- 并发能力问题
- 搭建主从集群，实现读写分离
- 故障回复问题
- 利用Redis哨兵，实现健康检测和自动恢复
- 存储能力问题
- 搭建分片集群，利用插槽机制实现动态扩容
13.1. 安装Redis
① 首先需要安装Redis所需要的依赖：
yum install -y gcc tcl |
② 下载redis.tar.gz到/tmp目录下
③ 解压缩
tar -xvf redis-6.2.4.tar.gz |
④ 进入redis目录：
cd redis-6.2.4 |
⑤ 运行编译命令：
make && make install |
⑥ 修改redis.conf文件中的一些配置：
# 绑定地址，默认是127.0.0.1，会导致只能在本地访问。修改为0.0.0.0则可以在任意IP访问 |
⑦ 启动与停止
启动Redis：
redis-server redis.conf |
停止redis服务：
redis-cli shutdown |
13.2. Redis持久化
Redis有两种持久化方案：
- RDB持久化
- AOF持久化
RDB全称Redis Database Backup file（Redis数据备份文件），也被叫做Redis数据快照。简单来说就是把内存中的所有数据都记录到磁盘中。当Redis实例故障重启后，从磁盘读取快照文件，恢复数据。快照文件称为RDB文件，默认是保存在当前运行目录。
13.2.1 RDB持久化
① 执行时机
RDB持久化在四种情况下会执行：
- 执行save命令
- 执行bgsave命令
- Redis停机时
- 触发RDB条件时
save命令
执行下面的命令，可以立即执行一次RDB：
redis-cli # 进入redis |
save命令会导致主进程执行RDB，这个过程中其它所有命令都会被阻塞。只有在数据迁移时可能用到。
bgsave命令
下面的命令可以异步执行RDB：
redis-cli # 进入redis |
这个命令执行后会开启独立进程完成RDB，主进程可以持续处理用户请求，不受影响。
停机时
Redis停机时会执行一次save命令，实现RDB持久化。
触发RDB条件
Redis内部有触发RDB的机制，可以在redis.conf文件中找到，格式如下：
# 900秒内，如果至少有1个key被修改，则执行bgsave ， 如果是save "" 则表示禁用RDB |
注：禁用RDB则添加save ""
RDB的其它配置也可以在redis.conf文件中设置：
# 是否压缩 ,建议不开启，压缩也会消耗cpu，磁盘的话不值钱 |
② RDB原理
bgsave开始时会fork主进程得到子进程，子进程共享主进程的内存数据。完成fork后读取内存数据并写入 RDB 文件。
fork采用的是copy-on-write技术：
- 当主进程执行读操作时，访问共享内存；
- 当主进程执行写操作时，则会拷贝一份数据，执行写操作。
③ 小结
RDB方式bgsave的基本流程？
- fork主进程得到一个子进程，共享内存空间
- 子进程读取内存数据并写入新的RDB文件
- 用新RDB文件替换旧的RDB文件
RDB会在什么时候执行？save 60 1000代表什么含义？
- 默认是服务停止时
- 代表60秒内至少执行1000次修改则触发RDB
RDB的缺点？
- RDB执行间隔时间长，两次RDB之间写入数据有丢失的风险
- fork子进程、压缩、写出RDB文件都比较耗时
13.2.2 AOF持久化
① AOF原理
AOF全称为Append Only File（追加文件）。Redis处理的每一个写命令都会记录在AOF文件，可以看做是命令日志文件。
② AOF配置
AOF默认是关闭的，需要修改redis.conf配置文件来开启AOF：
# 关闭RDB |
AOF的命令记录的频率也可以通过redis.conf文件来配：
# 表示每执行一次写命令，立即记录到AOF文件 |
三种策略对比：
| 配置项 | 刷盘时机 | 优点 | 缺点 |
|---|---|---|---|
| Always | 同步刷盘 | 可靠性高，几乎不丢数据 | 性能影响大 |
| everysec | 每秒刷盘 | 性能始终 | 最多丢失1秒数据 |
| no | 操作系统控制 | 性能最好 | 可靠性较差，可能丢失大量数据 |
③ AOF文件重写
因为是记录命令，AOF文件会比RDB文件大的多。而且AOF会记录对同一个key的多次写操作，但只有最后一次写操作才有意义。通过执行bgrewriteaof命令，可以让AOF文件执行重写功能，用最少的命令达到相同效果。
例如，AOF原本有三个命令set num 123;set name jack;set num 666，但是set num 123 和 set num 666都是对num的操作，第二次会覆盖第一次的值，因此第一个命令记录下来没有意义。
所以重写命令后，AOF文件内容就是：mset name jack num 666
Redis也会在触发阈值时自动去重写AOF文件。阈值也可以在redis.conf中配置：
# AOF文件比上次文件 增长超过多少百分比则触发重写 |
13.2.3 RDB与AOF对比
RDB和AOF各有自己的优缺点，如果对数据安全性要求较高，在实际开发中往往会结合两者来使用。
| RDB | AOF | |
|---|---|---|
| 持久化方式 | 定时对整个内存做快照 | 记录每一次执行的命令 |
| 数据完整性 | 不完整，两次备份之间会丢失 | 相对完整，取决于刷盘策略 |
| 文件大小 | 会有压缩，文件体积小 | 记录命令，文件体积很大 |
| 宕机恢复速度 | 很快 | 慢 |
| 数据恢复优先级 | 低，因为数据完整性不如AOF | 高，因为数据完整性更高 |
| 系统资源占用 | 高，大量CPU和内存消耗 | 低，主要是磁盘IO资源，但AOF重写时会占用大量CPU和内存资源 |
| 使用场景 | 可以容忍分钟的数据丢失，追求更快的启动速度 | 对数据安全性要求较高时常见 |
13.3. Redis主从
13.3.1 搭建主从架构
单节点Redis的并发能力是有上限的，要进一步提高Redis的并发能力，就需要搭建主从集群，实现读写分离。
① 集群架构
共包含三个节点，一个主节点，两个从节点。
这里我们会在同一台虚拟机中开启3个redis实例，模拟主从集群，信息如下：
| IP | PORT | 角色 |
|---|---|---|
| 192.168.150.101 | 7001 | master |
| 192.168.150.101 | 7002 | slave |
| 192.168.150.101 | 7003 | slave |
② 准备实例和配置
要在同一台虚拟机开启3个实例，必须准备三份不同的配置文件和目录，配置文件所在目录也就是工作目录。
1）创建目录
我们创建三个文件夹，名字分别叫7001、7002、7003：
# 进入/tmp目录 |
2）恢复原始配置
修改redis-6.2.4/redis.conf文件，将其中的持久化模式改为默认的RDB模式，AOF保持关闭状态。
# 开启RDB |
3）拷贝配置文件到每个实例目录
然后将redis-6.2.4/redis.conf文件拷贝到三个目录中（在/tmp目录执行下列命令）：
# 方式一：逐个拷贝 |
4）修改每个实例的端口、工作目录
修改每个文件夹内的配置文件，将端口分别修改为7001、7002、7003，将rdb文件保存位置都修改为自己所在目录（在/tmp目录执行下列命令）：
sed -i -e 's/6379/7001/g' -e 's/dir .\//dir \/tmp\/7001\//g' 7001/redis.conf |
5）修改每个实例的声明IP
虚拟机本身有多个IP，为了避免将来混乱，我们需要在redis.conf文件中指定每一个实例的绑定ip信息，格式如下：
# redis实例的声明 IP |
每个目录都要改，我们一键完成修改（在/tmp目录执行下列命令）：
# 逐一执行 |
③ 启动
为了方便查看日志，我们打开3个ssh窗口，分别启动3个redis实例，启动命令：
# 第1个 |
如果要一键停止，可以运行下面命令：
printf '%s\n' 7001 7002 7003 | xargs -I{} -t redis-cli -p {} shutdown |
④ 开启主从关系
现在三个实例还没有任何关系，要配置主从可以使用replicaof 或者slaveof（5.0以前）命令。
有临时和永久两种模式：
修改配置文件（永久生效）
- 在redis.conf中添加一行配置：
slaveof <masterip> <masterport>
- 在redis.conf中添加一行配置：
使用redis-cli客户端连接到redis服务，执行slaveof命令（重启后失效）：
slaveof <masterip> <masterport>
注意：在5.0以后新增命令replicaof，与salveof效果一致。
这里我们为了演示方便，使用方式二。
通过redis-cli命令连接7002，执行下面命令：
# 连接 7002 |
通过redis-cli命令连接7003，执行下面命令：
# 连接 7003 |
然后连接 7001节点，查看集群状态：
# 连接 7001 |
⑤ 测试
执行下列操作以测试：
- 利用redis-cli连接7001，执行
set num 123 - 利用redis-cli连接7002，执行
get num，再执行set num 666 - 利用redis-cli连接7003，执行
get num，再执行set num 888
可以发现，只有在7001这个master节点上可以执行写操作，7002和7003这两个slave节点只能执行读操作。
13.3.2 主从数据同步原理
① 全量同步
主从第一次建立连接时，会执行全量同步，将master节点的所有数据都拷贝给slave节点
这里有一个问题，master如何得知salve是第一次来连接呢？？
有几个概念，可以作为判断依据：
- Replication Id：简称replid，是数据集的标记，id一致则说明是同一数据集。每一个master都有唯一的replid，slave则会继承master节点的replid
- offset：偏移量，随着记录在repl_baklog中的数据增多而逐渐增大。slave完成同步时也会记录当前同步的offset。如果slave的offset小于master的offset，说明slave数据落后于master，需要更新。
因此slave做数据同步，必须向master声明自己的replication id 和offset，master才可以判断到底需要同步哪些数据。
因为slave原本也是一个master，有自己的replid和offset，当第一次变成slave，与master建立连接时，发送的replid和offset是自己的replid和offset。
master判断发现slave发送来的replid与自己的不一致，说明这是一个全新的slave，就知道要做全量同步了。
master会将自己的replid和offset都发送给这个slave，slave保存这些信息。以后slave的replid就与master一致了。
因此，master判断一个节点是否是第一次同步的依据，就是看replid是否一致。
完整流程描述：
- slave节点请求增量同步
- master节点判断replid，发现不一致，拒绝增量同步
- master将完整内存数据生成RDB，发送RDB到slave
- slave清空本地数据，加载master的RDB
- master将RDB期间的命令记录在repl_baklog，并持续将log中的命令发送给slave
- slave执行接收到的命令，保持与master之间的同步
② 增量同步
全量同步需要先做RDB，然后将RDB文件通过网络传输个slave，成本太高了。因此除了第一次做全量同步，其它大多数时候slave与master都是做增量同步。
什么是增量同步？就是只更新slave与master存在差异的部分数据。
③ repl_backlog原理
master怎么知道slave与自己的数据差异在哪里呢?
这就要说到全量同步时的repl_baklog文件了。
这个文件是一个固定大小的数组，只不过数组是环形，也就是说角标到达数组末尾后，会再次从0开始读写，这样数组头部的数据就会被覆盖。
注：repl_baklog大小有上限，写满后会覆盖最早的数据。如果slave断开时间过久，导致尚未备份的数据被覆盖，则无法基于log做增量同步，只能再次全量同步。
13.3.3 主从同步优化
主从同步可以保证主从数据的一致性，非常重要。
可以从以下几个方面来优化Redis主从就集群：
- 在master中配置
repl-diskless-sync yes启用无磁盘复制，避免全量同步时的磁盘IO。 - Redis单节点上的内存占用不要太大，减少RDB导致的过多磁盘IO
- 适当提高repl_baklog的大小，发现slave宕机时尽快实现故障恢复，尽可能避免全量同步
- 限制一个master上的slave节点数量，如果实在是太多slave，则可以采用主-从-从链式结构，减少master压力
13.3.4 小结
简述全量同步和增量同步区别？
- 全量同步：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时
13.4. Redis哨兵
Redis提供了哨兵（Sentinel）机制来实现主从集群的自动故障恢复。
13.4.1 哨兵原理
① 集群结构和作用
- RedisClient
- Sentinel若干，给RedisClient发送服务状态变更通知，同时监控redis集群状态
- Redis【master单个，slave若干】，master将数据同步给slave
哨兵的作用如下：
- 监控：Sentinel 会不断检查您的master和slave是否按预期工作
- 自动故障恢复：如果master故障，Sentinel会将一个slave提升为master。当故障实例恢复后也以新的master为主
- 通知：Sentinel充当Redis客户端的服务发现来源，当集群发生故障转移时，会将最新信息推送给Redis的客户端
② 集群监控原理
Sentinel基于心跳机制监测服务状态，每隔1秒向集群的每个实例发送ping命令：
•主观下线：如果某sentinel节点发现某实例未在规定时间响应，则认为该实例主观下线。
•客观下线：若超过指定数量（quorum）的sentinel都认为该实例主观下线，则该实例客观下线。quorum值最好超过Sentinel实例数量的一半。
③ 集群故障恢复原理
一旦发现master故障，sentinel需要在salve中选择一个作为新的master，选择依据是这样的：
- 首先会判断slave节点与master节点断开时间长短，如果超过指定值（down-after-milliseconds * 10）则会排除该slave节点
- 然后判断slave节点的slave-priority值，越小优先级越高，如果是0则永不参与选举
- 如果slave-prority一样，则判断slave节点的offset值，越大说明数据越新，优先级越高
- 最后是判断slave节点的运行id大小，越小优先级越高。
当选出一个新的master后，该如何实现切换呢？
流程如下：
- sentinel给备选的slave1节点发送slaveof no one命令，让该节点成为master
- sentinel给所有其它slave发送slaveof 192.168.150.101 7002 命令，让这些slave成为新master的从节点，开始从新的master上同步数据。
- 最后，sentinel将故障节点标记为slave，当故障节点恢复后会自动成为新的master的slave节点
④ 小结
Sentinel的三个作用是什么？
- 监控
- 故障转移
- 通知
Sentinel如何判断一个redis实例是否健康？
- 每隔1秒发送一次ping命令，如果超过一定时间没有相向则认为是主观下线
- 如果大多数sentinel都认为实例主观下线，则判定服务下线
故障转移步骤有哪些？
- 首先选定一个slave作为新的master，执行slaveof no one
- 然后让所有节点都执行slaveof 新master
- 修改故障节点配置，添加slaveof 新master
13.4.2 搭建哨兵集群
① 集群结构
这里我们搭建一个三节点形成的Sentinel集群，来监管之前的Redis【7001,7002,7003】主从集群。
三个sentinel实例信息如下：
| 节点 | IP | PORT |
|---|---|---|
| s1 | 192.168.150.101 | 27001 |
| s2 | 192.168.150.101 | 27002 |
| s3 | 192.168.150.101 | 27003 |
② 准备实例和配置
要在同一台虚拟机开启3个实例，必须准备三份不同的配置文件和目录，配置文件所在目录也就是工作目录。
我们创建三个文件夹，名字分别叫s1、s2、s3：
# 进入/tmp目录 |
在s1目录创建一个sentinel.conf文件，添加下面的内容：
port 27001 |
解读：
port 27001：是当前sentinel实例的端口sentinel monitor mymaster 192.168.150.101 7001 2：指定主节点信息mymaster：主节点名称，自定义，任意写192.168.150.101 7001：主节点的ip和端口2：选举master时的quorum值
然后将s1/sentinel.conf文件拷贝到s2、s3两个目录中（在/tmp目录执行下列命令）：
# 方式一：逐个拷贝 |
修改s2、s3两个文件夹内的配置文件，将端口分别修改为27002、27003：
sed -i -e 's/27001/27002/g' -e 's/s1/s2/g' s2/sentinel.conf |
③ 启动
为了方便查看日志，我们打开3个ssh窗口，分别启动3个redis实例，启动命令：
# 第1个 |
④ 测试
- 尝试让master节点7001宕机，查看sentinel日志
13.4.3 RedisTemplate
在Sentinel集群监管下的Redis主从集群，其节点会因为自动故障转移而发生变化，Redis的客户端必须感知这种变化，及时更新连接信息。Spring的RedisTemplate底层利用lettuce实现了节点的感知和自动切换。
下面，我们通过一个测试来实现RedisTemplate集成哨兵机制。
① 引入依赖
在项目的pom文件中引入依赖：
<dependency> |
② 配置Redis地址
在配置文件application.yml中指定redis的sentinel相关信息：
spring: |
③ 配置读写分离
在项目的启动类中，添加一个新的bean：
|
这个bean中配置的就是读写策略，包括四种：
- MASTER：从主节点读取
- MASTER_PREFERRED：优先从master节点读取，master不可用才读取replica
- REPLICA：从slave（replica）节点读取
- REPLICA _PREFERRED：优先从slave（replica）节点读取，所有的slave都不可用才读取master
13.5. Redis分片集群
13.5.1 搭建分片集群
主从和哨兵可以解决高可用、高并发读的问题。但是依然有两个问题没有解决：
- 海量数据存储问题
- 高并发写的问题
使用分片集群可以解决上述问题。
分片集群特征：
- 集群中有多个master，每个master保存不同数据
- 每个master都可以有多个slave节点
- master之间通过ping监测彼此健康状态
- 客户端请求可以访问集群任意节点，最终都会被转发到正确节点
① 集群结构
分片集群需要的节点数量较多，这里我们搭建一个最小的分片集群，包含3个master节点，每个master包含一个slave节点
在同一台虚拟机中开启6个redis实例，模拟分片集群，信息如下：
| IP | PORT | 角色 |
|---|---|---|
| 192.168.150.101 | 7001 | master |
| 192.168.150.101 | 7002 | master |
| 192.168.150.101 | 7003 | master |
| 192.168.150.101 | 8001 | slave |
| 192.168.150.101 | 8002 | slave |
| 192.168.150.101 | 8003 | slave |
② 准备实例和配置
删除之前的7001、7002、7003这几个目录，重新创建出7001、7002、7003、8001、8002、8003目录：
# 进入/tmp目录 |
在/tmp下准备一个新的redis.conf文件，内容如下：
port 6379 |
将这个文件拷贝到每个目录下：
# 进入/tmp目录 |
修改每个目录下的redis.conf，将其中的6379修改为与所在目录一致：
# 进入/tmp目录 |
③ 启动
因为已经配置了后台启动模式，所以可以直接启动服务：
# 进入/tmp目录 |
通过ps查看状态：
ps -ef | grep redis |
如果要关闭所有进程，可以执行命令：
ps -ef | grep redis | awk '{print $2}' | xargs kill |
或者（推荐这种方式）：
printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t redis-cli -p {} shutdown |
④ 创建集群
虽然服务启动了，但是目前每个服务之间都是独立的，没有任何关联。
我们需要执行命令来创建集群，在Redis5.0之前创建集群比较麻烦，5.0之后集群管理命令都集成到了redis-cli中。
1）Redis5.0之前
Redis5.0之前集群命令都是用redis安装包下的src/redis-trib.rb来实现的。因为redis-trib.rb是有ruby语言编写的所以需要安装ruby环境。
# 安装依赖 |
然后通过命令来管理集群：
# 进入redis的src目录 |
2）Redis5.0以后
我们使用的是Redis6.2.4版本，集群管理以及集成到了redis-cli中，格式如下：
redis-cli --cluster create --cluster-replicas 1 192.168.150.101:7001 192.168.150.101:7002 192.168.150.101:7003 192.168.150.101:8001 192.168.150.101:8002 192.168.150.101:8003 |
命令说明：
redis-cli --cluster或者./redis-trib.rb：代表集群操作命令create：代表是创建集群--replicas 1或者--cluster-replicas 1：指定集群中每个master的副本个数为1，此时节点总数 ÷ (replicas + 1)得到的就是master的数量。因此节点列表中的前n个就是master，其它节点都是slave节点，随机分配到不同master
运行后：输入yes，开始创建集群
通过命令可以查看集群状态：
redis-cli -p 7001 cluster nodes |
⑤ 测试
尝试连接7001节点，存储一个数据：
# 连接 |
结果悲剧了，(error) MOVED 15495 192.168.150.101:7003
集群操作时，需要给redis-cli加上-c参数才可以：
redis-cli -c -p 7001 |
13.5.2 散列插槽
① 插槽原理
Redis会把每一个master节点映射到0~16383共16384个插槽（hash slot）上，查看集群信息时就能看到：slots:[0-5460] (5461 slots) master
数据key不是与节点绑定，而是与插槽绑定。redis会根据key的有效部分计算插槽值，分两种情况：
- key中包含”{}”，且“{}”中至少包含1个字符，“{}”中的部分是有效部分
- key中不包含“{}”，整个key都是有效部分
例如：key是num，那么就根据num计算，如果是{itcast}num，则根据itcast计算。计算方式是利用CRC16算法得到一个hash值，然后对16384取余，得到的结果就是slot值。
set a 1 |
如上述代码，在7001这个节点执行set a 1时，对a做hash运算，对16384取余，得到的结果是15495，因此要存储到103节点。
到了7003后，执行get num时，对num做hash运算，对16384取余，得到的结果是2765，因此需要切换到7001节点
② 小结
Redis如何判断某个key应该在哪个实例？
- 将16384个插槽分配到不同的实例
- 根据key的有效部分计算哈希值，对16384取余
- 余数作为插槽，寻找插槽所在实例即可
如何将同一类数据固定的保存在同一个Redis实例？
- 这一类数据使用相同的有效部分，例如key都以{typeId}为前缀
13.5.3 集群伸缩
redis-cli –cluster提供了很多操作集群的命令，可以通过下面方式查看：
redis-cli --cluster help |
比如，添加节点的命令：
reids-cli --cluster add-node new_host:new_port existing_host:existing_port |
① 需求分析
需求：向集群中添加一个新的master节点，并向其中存储 num = 10
- 启动一个新的redis实例，端口为7004
- 添加7004到之前的集群，并作为一个master节点
- 给7004节点分配插槽，使得num这个key可以存储到7004实例
这里需要两个新的功能：
- 添加一个节点到集群中
- 将部分插槽分配到新插槽
② 创建新的redis实例
创建一个文件夹：
mkdir 7004 |
拷贝配置文件：
cp redis.conf /7004 |
修改配置文件：
sed /s/6379/7004/g 7004/redis.conf |
启动
redis-server 7004/redis.conf |
③ 添加新的节点到redis
执行命令：
redis-cli --cluster add-node 192.168.150.101:7004 192.168.150.101:7001 |
通过命令查看集群状态：
redis-cli -p 7001 cluster nodes |
7004加入了集群，并且默认是一个master节点，但是，可以看到7004节点的插槽数量为0，因此没有任何数据可以存储到7004上。
④ 转移插槽
我们要将num存储到7004节点，因此需要先看看num的插槽是多少：
get num |
查到num的插槽为2765，将0~3000的插槽从7001转移到7004，命令格式如下：
redis-cli --cluster reshard 192.168.150.101:7001 |
得到反馈：How many slots do you want to move (from 1 to 16384)?
询问要移动多少个插槽，我们计划是3000个
得到反馈：What is the receiving node ID?
使用哪个Node来接收这些插槽？
显然是7004，那么7004节点的id是多少呢，复制代码中id，然后拷贝到刚才的控制台后，这里询问，你的插槽是从哪里移动过来的？
- all：代表全部，也就是三个节点各转移一部分
- 具体的id：目标节点的id
- done：没有了
这里我们要从7001获取，因此填写7001的id
填完后，点击done，这样插槽转移就准备好了
确认要转移吗？输入yes
然后，通过命令查看结果：
redis-cli -p 7001 cluster node |
13.5.4 故障转移
① 自动故障转移
当集群中有一个master宕机会发生什么呢？
直接停止一个redis实例，例如7002：
redis-cli -p 7002 shutdown |
1）首先是该实例与其它实例失去连接
2）然后是疑似宕机
3）最后是确定下线，自动提升一个slave为新的master
4）当7002再次启动，就会变为一个slave节点了
② 手动故障转移
利用cluster failover命令可以手动让集群中的某个master宕机，切换到执行cluster failover命令的这个slave节点，实现无感知的数据迁移。
这种failover命令可以指定三种模式：
- 缺省：默认的流程，如图1~6歩
- force：省略了对offset的一致性校验
- takeover：直接执行第5歩，忽略数据一致性、忽略master状态和其它master的意见
案例需求：在7002这个slave节点执行手动故障转移，重新夺回master地位
步骤如下：
1）利用redis-cli连接7002这个节点
2）执行cluster failover命令
redis-cli -p 7002 |
13.5.5 RedisTemplate访问分片集群
RedisTemplate底层同样基于lettuce实现了分片集群的支持，而使用的步骤与哨兵模式基本一致：
1）引入redis的starter依赖
2）配置分片集群地址
3）配置读写分离
与哨兵模式相比，其中只有分片集群的配置方式略有差异，如下：
spring: |
十四、多级缓存
传统的缓存策略一般是请求到达Tomcat后，先查询Redis，如果未命中则查询数据库
存在下面的问题：
请求要经过Tomcat处理，Tomcat的性能成为整个系统的瓶颈
Redis缓存失效时，会对数据库产生冲击
多级缓存就是充分利用请求处理的每个环节，分别添加缓存，减轻Tomcat压力，提升服务性能：
- 浏览器访问静态资源时，优先读取浏览器本地缓存
- 访问非静态资源（ajax查询数据）时，访问服务端
- 请求到达Nginx后，优先读取Nginx本地缓存
- 如果Nginx本地缓存未命中，则去直接查询Redis（不经过Tomcat）
- 如果Redis查询未命中，则查询Tomcat
- 请求进入Tomcat后，优先查询JVM进程缓存
- 如果JVM进程缓存未命中，则查询数据库
在多级缓存架构中，Nginx内部需要编写本地缓存查询、Redis查询、Tomcat查询的业务逻辑，因此这样的nginx服务不再是一个反向代理服务器，而是一个编写业务的Web服务器了。
因此这样的业务Nginx服务也需要搭建集群来提高并发，再有专门的nginx服务来做反向代理。
另外，我们的Tomcat服务将来也会部署为集群模式。
可见，多级缓存的关键有两个：
- 一个是在nginx中编写业务，实现nginx本地缓存、Redis、Tomcat的查询
- 另一个就是在Tomcat中实现JVM进程缓存
其中Nginx编程则会用到OpenResty框架结合Lua这样的语言。
14.1 JVM进程缓存
14.1.1 案例引入
① 安装MySQL
后期做数据同步需要用到MySQL的主从功能，所以需要在虚拟机中，利用Docker来运行一个MySQL容器。
1）为了方便后期配置MySQL，我们先准备两个目录，用于挂载容器的数据和配置文件目录：
# 进入/tmp目录 |
2）运行命令
进入mysql目录后，执行下面的Docker命令：
docker run \ |
3）修改配置
在/tmp/mysql/conf目录添加一个my.cnf文件，作为mysql的配置文件：
# 创建文件 |
文件的内容如下：
[mysqld] |
4）重启
配置修改后，必须重启容器：
docker restart mysql |
② 导入SQL
接下来，利用Navicat客户端连接MySQL，然后导入sql文件：
其中包含两张表：
- tb_item：商品表，包含商品的基本信息
- tb_item_stock：商品库存表，包含商品的库存信息
之所以将库存分离出来，是因为库存是更新比较频繁的信息，写操作较多。而其他信息修改的频率非常低。
③ 导入Demo工程
其中的业务包括：
- 分页查询商品
- 新增商品
- 修改商品
- 修改库存
- 删除商品
- 根据id查询商品
- 根据id查询库存
业务全部使用mybatis-plus来实现，如有需要请自行修改业务逻辑。
14.1.2 初识Caffeine
缓存在日常开发中启动至关重要的作用，由于是存储在内存中，数据的读取速度是非常快的，能大量减少对数据库的访问，减少数据库的压力。缓存分为两类：
- 分布式缓存，例如Redis：
- 优点：存储容量更大、可靠性更好、可以在集群间共享
- 缺点：访问缓存有网络开销
- 场景：缓存数据量较大、可靠性要求较高、需要在集群间共享
- 进程本地缓存，例如HashMap、GuavaCache：
- 优点：读取本地内存，没有网络开销，速度更快
- 缺点：存储容量有限、可靠性较低、无法共享
- 场景：性能要求较高，缓存数据量较小
利用Caffeine框架来实现JVM进程缓存。
Caffeine是一个基于Java8开发的，提供了近乎最佳命中率的高性能的本地缓存库。目前Spring内部的缓存使用的就是Caffeine。GitHub地址：https://github.com/ben-manes/caffeine
pom依赖
<dependency> |
缓存使用的基本API：
|
Caffeine既然是缓存的一种，肯定需要有缓存的清除策略，不然的话内存总会有耗尽的时候。
Caffeine提供了三种缓存驱逐策略：
基于容量：设置缓存的数量上限
// 创建缓存对象
Cache<String, String> cache = Caffeine.newBuilder()
.maximumSize(1) // 设置缓存大小上限为 1
.build();基于时间：设置缓存的有效时间
// 创建缓存对象
Cache<String, String> cache = Caffeine.newBuilder()
// 设置缓存有效期为 10 秒，从最后一次写入开始计时
.expireAfterWrite(Duration.ofSeconds(10))
.build();基于引用：设置缓存为软引用或弱引用，利用GC来回收缓存数据。性能较差，不建议使用。
注意：在默认情况下，当一个缓存元素过期的时候，Caffeine不会自动立即将其清理和驱逐。而是在一次读或写操作后，或者在空闲时间完成对失效数据的驱逐。
14.1.3 实现JVM进程缓存
① 需求
利用Caffeine实现下列需求：
- 给根据id查询商品的业务添加缓存，缓存未命中时查询数据库
- 给根据id查询商品库存的业务添加缓存，缓存未命中时查询数据库
- 缓存初始大小为100
- 缓存上限为10000
② 实现
首先，我们需要定义两个Caffeine的缓存对象，分别保存商品、库存的缓存数据。
在item-service的com.heima.item.config包下定义CaffeineConfig类：
package com.heima.item.config; |
然后，修改item-service中的com.heima.item.web包下的ItemController类，添加缓存逻辑：
|
14.2 Lua语法入门
Nginx编程需要用到Lua语言，因此我们必须先入门Lua的基本语法。
14.2.1 初识Lua
Lua 是一种轻量小巧的脚本语言，用标准C语言编写并以源代码形式开放， 其设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。官网：https://www.lua.org/
Lua经常嵌入到C语言开发的程序中，例如游戏开发、游戏插件等。
Nginx本身也是C语言开发，因此也允许基于Lua做拓展。
14.2.2 HelloWorld
CentOS7默认已经安装了Lua语言环境，所以可以直接运行Lua代码。
1）在Linux虚拟机的任意目录下，新建一个hello.lua文件
touch hello.lua |
2）添加下面的内容
print("Hello World!") |
3）运行
lua hello.lua |
14.2.3 量和循环
学习任何语言必然离不开变量，而变量的声明必须先知道数据的类型。
14.2.3 变量和循环
① Lua的数据类型
Lua中支持的常见数据类型包括：
| 数据类型 | 描述 |
|---|---|
| nil | 只有值nil属于该类，表示一个无效值（在条件表达式中相当于false） |
| boolean | 包含两个值：false和true |
| number | 表示双精度类型的实浮点数 |
| string | 字符串由一对双引号或单引号来表示 |
| function | 由 C 或 Lua 编写的函数 |
| table | Lua中表（table）其实是一个”关联数组”(associative arrays)，数组的索引可以是数字、字符串或表类型。在Lua中，table的创建是通过”构造表达式”来完成，最简单的构造表达式是{ }，用来创建一个空表。 |
另外，Lua提供了type()函数来判断一个变量的数据类型：
print(type(10*3.4)) |
② 声明变量
Lua声明变量的时候无需指定数据类型，而是用local来声明变量为局部变量：
-- 声明字符串，可以用单引号或双引号， |
Lua中的table类型既可以作为数组，又可以作为Java中的map来使用。数组就是特殊的table，key是数组角标而已：
-- 声明数组 ，key为角标的 table |
Lua中的数组角标是从1开始，访问的时候与Java中类似：
-- 访问数组，lua数组的角标从1开始 |
Lua中的table可以用key来访问：
-- 访问table |
③ 循环
对于table，我们可以利用for循环来遍历。不过数组和普通table遍历略有差异。
遍历数组：
-- 声明数组 key为索引的 table |
遍历普通table
-- 声明map，也就是table |
14.2.4 条件控制、函数
Lua中的条件控制和函数声明与Java类似。
① 函数
定义函数的语法：
function 函数名( argument1, argument2..., argumentn) |
例如，定义一个函数，用来打印数组：
function printArr(arr) |
② 条件控制
类似Java的条件控制，例如if、else语法：
if(布尔表达式) |
与java不同，布尔表达式中的逻辑运算是基于英文单词：
| and | or | not |
|---|---|---|
③ 实例
需求：自定义一个函数，可以打印table，当参数为nil时，打印错误信息
function printArr(arr) |
14.3 实现多级缓存
多级缓存的实现离不开Nginx编程，而Nginx编程又离不开OpenResty。
14.3.1 安装OpenResty
OpenResty® 是一个基于 Nginx的高性能 Web 平台，用于方便地搭建能够处理超高并发、扩展性极高的动态 Web 应用、Web 服务和动态网关。具备下列特点：
- 具备Nginx的完整功能
- 基于Lua语言进行扩展，集成了大量精良的 Lua 库、第三方模块
- 允许使用Lua自定义业务逻辑、自定义库
官方网站： https://openresty.org/cn/
① 安装开发库
首先要安装OpenResty的依赖开发库，执行命令：
yum install -y pcre-devel openssl-devel gcc --skip-broken |
② 安装OpenResty仓库
可以在 CentOS 系统中添加 openresty 仓库，这样就可以便于未来安装或更新的软件包（通过 yum check-update 命令）。运行下面的命令就可以添加仓库：
yum-config-manager --add-repo https://openresty.org/package/centos/openresty.repo |
如果提示说命令不存在，则运行：
yum install -y yum-utils |
然后再重复上面的命令
③ 安装OpenResty
安装软件包，比如 openresty：
yum install -y openresty |
④ 安装opm工具
opm是OpenResty的一个管理工具，可以帮助我们安装一个第三方的Lua模块。
如果你想安装命令行工具 opm，那么可以像下面这样安装 openresty-opm 包：
yum install -y openresty-opm |
⑤ 配置nginx的环境变量
默认情况下，OpenResty安装的目录是：/usr/local/openresty，OpenResty就是在Nginx基础上集成了一些Lua模块。
配置后可在任意目录执行
打开配置文件：
vi /etc/profile |
在最下面加入两行：
export NGINX_HOME=/usr/local/openresty/nginx |
NGINX_HOME：后面是OpenResty安装目录下的nginx的目录
然后让配置生效：
source /etc/profile |
⑥ 启动和运行
OpenResty底层是基于Nginx的，查看OpenResty目录的nginx目录，结构与windows中安装的nginx基本一致：
# 启动nginx |
nginx的默认配置文件注释太多，影响后续我们的编辑，这里将nginx.conf中的注释部分删除，保留有效部分。
修改/usr/local/openresty/nginx/conf/nginx.conf文件，内容如下：
#user nobody; |
在Linux的控制台输入命令以启动nginx：
nginx |
然后访问页面：http://192.168.150.101:8081，注意ip地址替换为你自己的虚拟机IP
⑦ 备注
加载OpenResty的lua模块：
#lua 模块 |
common.lua
-- 封装函数，发送http请求，并解析响应 |
释放Redis连接API：
-- 关闭redis连接的工具方法，其实是放入连接池 |
读取Redis数据的API：
-- 查询redis的方法 ip和port是redis地址，key是查询的key |
开启共享词典：
# 共享字典，也就是本地缓存，名称叫做：item_cache，大小150m |
14.3.2 OpenResty快速入门
① 反向代理流程
- windows上的nginx用来做反向代理服务，将前端的查询商品的ajax请求代理到OpenResty集群
- OpenResty集群用来编写多级缓存业务
页面有发起ajax请求查询真实商品数据
- Method：GET
- URL：http://localhost/api/item/1001
请求地址是localhost，端口是80，就被windows上安装的Nginx服务给接收到了。然后代理给了OpenResty集群
# OpenResty集群，在虚拟机中，实现多级缓存业务 |
我们需要在OpenResty中编写业务，查询商品数据并返回到浏览器。
但是这次，我们先在OpenResty接收请求，返回假的商品数据。
② OpenResty监听请求【模拟数据，静态】
OpenResty的很多功能都依赖于其目录下的Lua库，需要在nginx.conf中指定依赖库的目录，并导入依赖：
1）添加对OpenResty的Lua模块的加载
修改/usr/local/openresty/nginx/conf/nginx.conf文件，在其中的http下面，添加下面代码：
#lua 模块 |
2）监听/api/item路径
修改/usr/local/openresty/nginx/conf/nginx.conf文件，在nginx.conf的server下面，添加对/api/item这个路径的监听：
location /api/item { |
这个监听，就类似于SpringMVC中的@GetMapping("/api/item")做路径映射。
而content_by_lua_file lua/item.lua则相当于调用item.lua这个文件，执行其中的业务，把结果返回给用户。相当于java中调用service。
③ 编写item.lua
1）在/usr/loca/openresty/nginx目录创建文件夹：lua
mkdir lua |
2）在/usr/loca/openresty/nginx/lua文件夹下，新建文件：item.lua
touch item.lua |
3）编写item.lua，返回假数据
item.lua中，利用ngx.say()函数返回数据到Response中
ngx.say('{"id":10001,"name":"SALSA AIR","title":"RIMOWA 21寸托运箱拉杆箱 SALSA AIR系列果绿色 820.70.36.4","price":17900,"image":"https://m.360buyimg.com/mobilecms/s720x720_jfs/t6934/364/1195375010/84676/e9f2c55f/597ece38N0ddcbc77.jpg!q70.jpg.webp","category":"拉杆箱","brand":"RIMOWA","spec":"","status":1,"createTime":"2019-04-30T16:00:00.000+00:00","updateTime":"2019-04-30T16:00:00.000+00:00","stock":2999,"sold":31290}') |
4）重新加载配置
nginx -s reload |
刷新商品页面：http://localhost/item.html?id=10001，即可看到效果
14.3.3 请求参数处理
上一节中，我们在OpenResty接收前端请求，但是返回的是假数据。
要返回真实数据，必须根据前端传递来的商品id，查询商品信息才可以。
那么如何获取前端传递的商品参数呢？
① 获取参数的API
OpenResty中提供了一些API用来获取不同类型的前端请求参数：
| 参数格式 | 参数示例 | 参数解析代码示例 |
|---|---|---|
| 路径占位符 | /item/1001 | # 1.正则表达式匹配： location ~ /item/(\d+) { content_by_lua_file lua/item.lua } # 2.匹配到的参数会存入ngx.var数组中，可以使用角标获取 local id = ngx.var[1] |
| 请求头 | id:1001 | # 获取请求头，返回值是table类型 local headers = ngx.req.get_headers() |
| Get请求参数 | ?id=1001 | # 获取Get请求参数，返回值是table类型 local getParams = ngx.req.get_uri_args() |
| Post表单参数 | id=1001 | # 读取请求体 ngx.req.read_body() # 获取post表单参数，返回值是table类型 local postParams = ngx.req.get_post_args() |
| JSON参数 | {“id”:1001} | # 读取请求体 ngx.req.read_body() # 获取body中的json参数，返回值是string类型 local jsonBody = ngx.req.get_body_data() |
② 获取参数并返回
在前端发起的ajax请求：GET：http://localhost/api/item/10001
商品id是以路径占位符方式传递的，因此可以利用正则表达式匹配的方式来获取ID
1）获取商品id
修改/usr/loca/openresty/nginx/nginx.conf文件中监听/api/item的代码，利用正则表达式获取ID：
location ~ /api/item/(\d+) { |
2）拼接ID并返回
修改/usr/loca/openresty/nginx/lua/item.lua文件，获取id并拼接到结果中返回：
-- 获取商品id |
3）重新加载并测试
运行命令以重新加载OpenResty配置：
nginx -s reload |
刷新页面可以看到结果中已经带上了ID
14.3.4 查询Tomcat
拿到商品ID后，本应去缓存中查询商品信息，不过目前我们还未建立nginx、redis缓存。因此，这里我们先根据商品id去tomcat查询商品信息。
需要注意的是，我们的OpenResty是在虚拟机，Tomcat是在Windows电脑上。两者IP一定不要搞错了。
- OpenResty的ip是192.168.49.10
- 本机ip地址取前三位：192.168.150.1即可，需要关闭防火墙
① 发送http请求的API
nginx提供了内部API用以发送http请求：
local resp = ngx.location.capture("/path",{ |
返回的响应内容包括：
- resp.status：响应状态码
- resp.header：响应头，是一个table
- resp.body：响应体，就是响应数据
注意：这里的path是路径，并不包含IP和端口。这个请求会被nginx内部的server监听并处理。
但是我们希望这个请求发送到Tomcat服务器，所以还需要编写一个server来对这个路径做反向代理：
location /path { |
② 封装http工具
下面，我们封装一个发送Http请求的工具，基于ngx.location.capture来实现查询tomcat。
1）添加反向代理，到windows的Java服务
因为item-service中的接口都是/item开头，所以我们监听/item路径，代理到windows上的tomcat服务。
修改 /usr/local/openresty/nginx/conf/nginx.conf文件，添加一个location：
location /item { |
以后，只要我们调用ngx.location.capture("/item")，就一定能发送请求到windows的tomcat服务。
2）封装工具类
之前我们说过，OpenResty启动时会加载以下两个目录中的工具文件：
#lua 模块 |
所以，自定义的http工具也需要放到这个目录下。
在/usr/local/openresty/lualib目录下，新建一个common.lua文件：
vi /usr/local/openresty/lualib/common.lua |
内容如下:
-- 封装函数，发送http请求，并解析响应 |
这个工具将read_http函数封装到_M这个table类型的变量中，并且返回，这类似于导出。
使用的时候，可以利用require('common')来导入该函数库，这里的common是函数库的文件名。
3）实现商品查询
最后，我们修改/usr/local/openresty/lua/item.lua文件，利用刚刚封装的函数库实现对tomcat的查询：
-- 引入自定义common工具模块，返回值是common中返回的 _M |
这里查询到的结果是json字符串，并且包含商品、库存两个json字符串，页面最终需要的是把两个json拼接为一个json，这就需要我们先把JSON变为lua的table，完成数据整合后，再转为JSON。
③ CJSON工具类
OpenResty提供了一个cjson的模块用来处理JSON的序列化和反序列化。
官方地址： https://github.com/openresty/lua-cjson/
1）引入cjson模块：
local cjson = require "cjson" |
2）序列化：
local obj = { |
3）反序列化：
local json = '{"name": "jack", "age": 21}' |
④ 实现Tomcat查询
下面，我们修改之前的item.lua中的业务，添加json处理功能：
-- 导入common函数库 |
⑤ 基于ID负载均衡
刚才的代码中，我们的tomcat是单机部署。而实际开发中，tomcat一定是集群模式：
因此，OpenResty需要对tomcat集群做负载均衡。
而默认的负载均衡规则是轮询模式，当我们查询/item/10001时：
- 第一次会访问8081端口的tomcat服务，在该服务内部就形成了JVM进程缓存
- 第二次会访问8082端口的tomcat服务，该服务内部没有JVM缓存（因为JVM缓存无法共享），会查询数据库
- …
你看，因为轮询的原因，第一次查询8081形成的JVM缓存并未生效，直到下一次再次访问到8081时才可以生效，缓存命中率太低了。
怎么办？
如果能让同一个商品，每次查询时都访问同一个tomcat服务，那么JVM缓存就一定能生效了。
也就是说，我们需要根据商品id做负载均衡，而不是轮询。
1）原理
nginx提供了基于请求路径做负载均衡的算法：
nginx根据请求路径做hash运算，把得到的数值对tomcat服务的数量取余，余数是几，就访问第几个服务，实现负载均衡。
例如：
- 我们的请求路径是 /item/10001
- tomcat总数为2台（8081、8082）
- 对请求路径/item/1001做hash运算求余的结果为1
- 则访问第一个tomcat服务，也就是8081
只要id不变，每次hash运算结果也不会变，那就可以保证同一个商品，一直访问同一个tomcat服务，确保JVM缓存生效。
2）实现
修改/usr/local/openresty/nginx/conf/nginx.conf文件，实现基于ID做负载均衡。
首先，定义tomcat集群，并设置基于路径做负载均衡：
upstream tomcat-cluster { |
然后，修改对tomcat服务的反向代理，目标指向tomcat集群：
location /item { |
重新加载OpenResty
nginx -s reload |
3）测试
启动两台tomcat服务：8081,8082
14.3.5 Redis缓存预热
Redis缓存会面临冷启动问题：
冷启动：服务刚刚启动时，Redis中并没有缓存，如果所有商品数据都在第一次查询时添加缓存，可能会给数据库带来较大压力。
缓存预热：在实际开发中，我们可以利用大数据统计用户访问的热点数据，在项目启动时将这些热点数据提前查询并保存到Redis中。
我们数据量较少，并且没有数据统计相关功能，目前可以在启动时将所有数据都放入缓存中。
1）利用Docker安装Redis
docker run --name redis -p 6379:6379 -d redis redis-server --appendonly yes |
2）在item-service服务中引入Redis依赖
<dependency> |
3）配置Redis地址
spring: |
4）编写初始化类
缓存预热需要在项目启动时完成，并且必须是拿到RedisTemplate之后。
这里我们利用InitializingBean接口来实现，因为InitializingBean可以在对象被Spring创建并且成员变量全部注入后执行。
package com.heima.item.config; |
14.3.6 查询Redis缓存
现在，Redis缓存已经准备就绪，我们可以再OpenResty中实现查询Redis的逻辑了。
当请求进入OpenResty之后：
- 优先查询Redis缓存
- 如果Redis缓存未命中，再查询Tomcat
① 封装Redis工具
OpenResty提供了操作Redis的模块，我们只要引入该模块就能直接使用。但是为了方便，我们将Redis操作封装到之前的common.lua工具库中。
修改/usr/local/openresty/lualib/common.lua文件：
1）引入Redis模块，并初始化Redis对象
-- 导入redis |
2）封装函数，用来释放Redis连接，其实是放入连接池
-- 关闭redis连接的工具方法，其实是放入连接池 |
3）封装函数，根据key查询Redis数据
-- 查询redis的方法 ip和port是redis地址，key是查询的key |
4）导出
-- 将方法导出 |
完整的common.lua：
-- 导入redis |
② 实现Redis查询
接下来，我们就可以去修改item.lua文件，实现对Redis的查询了。
查询逻辑是：
- 根据id查询Redis
- 如果查询失败则继续查询Tomcat
- 将查询结果返回
1）修改/usr/local/openresty/lua/item.lua文件，添加一个查询函数：
-- 导入common函数库 |
2）而后修改商品查询、库存查询的业务
-- 获取路径参数 |
3）完整的item.lua代码：
-- 导入common函数库 |
14.3.7 Nginx本地缓存
现在，整个多级缓存中只差最后一环，也就是nginx的本地缓存了。
① 本地缓存API
OpenResty为Nginx提供了shard dict的功能，可以在nginx的多个worker之间共享数据，实现缓存功能。
1）开启共享字典，在nginx.conf的http下添加配置：
# 共享字典，也就是本地缓存，名称叫做：item_cache，大小150m |
2）操作共享字典：
-- 获取本地缓存对象 |
② 实现本地缓存查询
1）修改/usr/local/openresty/lua/item.lua文件，修改read_data查询函数，添加本地缓存逻辑：
-- 导入共享词典，本地缓存 |
2）修改item.lua中查询商品和库存的业务，实现最新的read_data函数：
-- 查询商品信息 |
其实就是多了缓存时间参数，过期后nginx缓存会自动删除，下次访问即可更新缓存。
这里给商品基本信息设置超时时间为30分钟，库存为1分钟。
因为库存更新频率较高，如果缓存时间过长，可能与数据库差异较大。
3）完整的item.lua文件：
-- 导入common函数库 |
14.4 缓存同步【bug，待解决】
大多数情况下，浏览器查询到的都是缓存数据，如果缓存数据与数据库数据存在较大差异，可能会产生比较严重的后果。
所以我们必须保证数据库数据、缓存数据的一致性，这就是缓存与数据库的同步。
14.4.1 数据同步策略
缓存数据同步的常见方式有三种：
设置有效期：给缓存设置有效期，到期后自动删除。再次查询时更新
- 优势：简单、方便
- 缺点：时效性差，缓存过期之前可能不一致
- 场景：更新频率较低，时效性要求低的业务
同步双写：在修改数据库的同时，直接修改缓存
- 优势：时效性强，缓存与数据库强一致
- 缺点：有代码侵入，耦合度高；
- 场景：对一致性、时效性要求较高的缓存数据
异步通知：修改数据库时发送事件通知，相关服务监听到通知后修改缓存数据
- 优势：低耦合，可以同时通知多个缓存服务
- 缺点：时效性一般，可能存在中间不一致状态
- 场景：时效性要求一般，有多个服务需要同步
而异步实现又可以基于MQ或者Canal来实现：
1）基于MQ的异步通知：
- 商品服务完成对数据的修改后，只需要发送一条消息到MQ中。
- 缓存服务监听MQ消息，然后完成对缓存的更新
依然有少量的代码侵入。
2）基于Canal的通知：
- 商品服务完成商品修改后，业务直接结束，没有任何代码侵入
- Canal监听MySQL变化，当发现变化后，立即通知缓存服务
- 缓存服务接收到canal通知，更新缓存
代码零侵入
14.4.2 Canal
① 认识Canal
**Canal [kə’næl]**，译意为水道/管道/沟渠，canal是阿里巴巴旗下的一款开源项目，基于Java开发。基于数据库增量日志解析，提供增量数据订阅&消费。GitHub的地址：https://github.com/alibaba/canal
Canal是基于mysql的主从同步来实现的
- 1）MySQL master 将数据变更写入二进制日志( binary log），其中记录的数据叫做binary log events
- 2）MySQL slave 将 master 的 binary log events拷贝到它的中继日志(relay log)
- 3）MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据
而Canal就是把自己伪装成MySQL的一个slave节点，从而监听master的binary log变化。再把得到的变化信息通知给Canal的客户端，进而完成对其它数据库的同步。
② 安装Canal
1）开启MySQL主从
Canal是基于MySQL的主从同步功能，因此必须先开启MySQL的主从功能才可以。
这里以之前用Docker运行的mysql为例：
开启binlog
打开mysql容器挂载的日志文件，我的在
/tmp/mysql/conf目录修改文件：
vi /tmp/mysql/conf/my.cnf
添加内容：
log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=heima配置解读：
log-bin=/var/lib/mysql/mysql-bin：设置binary log文件的存放地址和文件名，叫做mysql-binbinlog-do-db=heima：指定对哪个database记录binary log events，这里记录heima这个库
最终效果：
[mysqld]
skip-name-resolve
character_set_server=utf8
datadir=/var/lib/mysql
server-id=1000
log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=heima设置用户权限
接下来添加一个仅用于数据同步的账户，出于安全考虑，这里仅提供对heima这个库的操作权限。
create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%' identified by 'canal';
FLUSH PRIVILEGES;重启mysql容器即可
docker restart mysql
测试设置是否成功：在mysql控制台，或者Navicat中，输入命令：
show master status;
2）安装Canal
创建网络
我们需要创建一个网络，将MySQL、Canal、MQ放到同一个Docker网络中：
docker network create heima
让mysql加入这个网络：
docker network connect heima mysql
安装
上传到虚拟机，然后通过命令导入：
docker load -i canal.tar
然后运行命令创建Canal容器：
docker run -p 11111:11111 --name canal \
-e canal.destinations=heima \
-e canal.instance.master.address=mysql:3306 \
-e canal.instance.dbUsername=canal \
-e canal.instance.dbPassword=canal \
-e canal.instance.connectionCharset=UTF-8 \
-e canal.instance.tsdb.enable=true \
-e canal.instance.gtidon=false \
-e canal.instance.filter.regex=heima\\..* \
--network heima \
-d canal/canal-server:v1.1.5说明:
-p 11111:11111：这是canal的默认监听端口-e canal.instance.master.address=mysql:3306：数据库地址和端口，如果不知道mysql容器地址，可以通过docker inspect 容器id来查看-e canal.instance.dbUsername=canal：数据库用户名-e canal.instance.dbPassword=canal：数据库密码-e canal.instance.filter.regex=：要监听的表名称
表名称监听支持的语法：
mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\)
常见例子：
1. 所有表：.* or .*\\..*
2. canal schema下所有表： canal\\..*
3. canal下的以canal打头的表：canal\\.canal.*
4. canal schema下的一张表：canal.test1
5. 多个规则组合使用然后以逗号隔开：canal\\..*,mysql.test1,mysql.test2
③ 监听Canal
Canal提供了各种语言的客户端，当Canal监听到binlog变化时，会通知Canal的客户端。
利用Canal提供的Java客户端，监听Canal通知消息。当收到变化的消息时，完成对缓存的更新。
不过这里我们会使用GitHub上的第三方开源的canal-starter客户端。地址：https://github.com/NormanGyllenhaal/canal-client
与SpringBoot完美整合，自动装配，比官方客户端要简单好用很多。
1）引入依赖：
<dependency> |
2）编写配置：
canal: |
3）修改Item实体类
通过@Id、@Column、等注解完成Item与数据库表字段的映射：
package com.heima.item.pojo; |
4）编写监听器
通过实现EntryHandler<T>接口编写监听器，监听Canal消息。注意两点：
- 实现类通过
@CanalTable("tb_item")指定监听的表信息 - EntryHandler的泛型是与表对应的实体类
package com.heima.item.canal; |
在这里对Redis的操作都封装到了RedisHandler这个对象中，是我们之前做缓存预热时编写的一个类，内容如下：
package com.heima.item.config; |
十五、服务异步通信【MQ相关】
消息队列在使用过程中，面临着很多实际问题需要思考：
- 消息可靠性问题：如何确保发送的消息至少被消费一次
- 延迟消息问题：如何实现消息的延迟投递
- 消息堆积问题：如何解决是百万消息的总堆积，无法及时消费的问题
- 高可用问题：如何避免单点MQ故障导致的不可用问题
15.1 消息可靠性
消息从发送，到消费者接收，会经历多个过程：
- publisher => exchange => 多个queue => 每个queue对应的consumer
其中的每一步都可能导致消息丢失，常见的丢失原因包括：
- 发送时丢失：
- 生产者发送的消息未送达exchange
- 消息到达exchange后未到达queue
- MQ宕机，queue将消息丢失
- consumer接收到消息后未消费就宕机
针对这些问题，RabbitMQ分别给出了解决方案：
- 生产者确认机制
- mq持久化
- 消费者确认机制
- 失败重试机制
15.1.1 生产者消息确认
RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功。
返回结果有两种方式：
- publisher-confirm，发送者确认
- 消息成功投递到交换机，返回ack
- 消息未投递到交换机，返回nack
- publisher-return，发送者回执
- 消息投递到交换机了，但是没有路由到队列。返回ACK，及路由失败原因。
注：确认机制发送消息时，需要给每个消息设置一个全局唯一id，以区分不同消息，避免ack冲突
① 修改配置
首先，修改publisher服务中的application.yml文件，添加下面的内容：
spring: |
说明：
publish-confirm-type：开启publisher-confirm，这里支持两种类型：simple：同步等待confirm结果，直到超时correlated：异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback
publish-returns：开启publish-return功能，同样是基于callback机制，不过是定义ReturnCallbacktemplate.mandatory：定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息
② 定义Return回调
每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目加载时配置：
修改publisher服务，添加一个：
package cn.itcast.mq.config; |
③ 定义ConfirmCallback
ConfirmCallback可以在发送消息时指定，因为每个业务处理confirm成功或失败的逻辑不一定相同。
在publisher服务的cn.itcast.mq.spring.SpringAmqpTest类中，定义一个单元测试方法：
public void testSendMessage2SimpleQueue() throws InterruptedException { |
15.1.2 消息持久化
生产者确认可以确保消息投递到RabbitMQ的队列中，但是消息发送到RabbitMQ以后，如果突然宕机，也可能导致消息丢失。
要想确保消息在RabbitMQ中安全保存，必须开启消息持久化机制。
- 交换机持久化
- 队列持久化
- 消息持久化
① 交换机持久化
RabbitMQ中交换机默认是非持久化的，mq重启后就丢失。
SpringAMQP中可以通过代码指定交换机持久化：
|
事实上，默认情况下，由SpringAMQP声明的交换机都是持久化的。
可以在RabbitMQ控制台看到持久化的交换机都会带上D的标示
② 队列持久化
RabbitMQ中队列默认是非持久化的，mq重启后就丢失。
SpringAMQP中可以通过代码指定交换机持久化：
|
事实上，默认情况下，由SpringAMQP声明的队列都是持久化的。
可以在RabbitMQ控制台看到持久化的队列都会带上D的标示
③ 消息持久化
利用SpringAMQP发送消息时，可以设置消息的属性（MessageProperties），指定delivery-mode：
- 1：非持久化
- 2：持久化
用java代码指定：
|
默认情况下，SpringAMQP发出的任何消息都是持久化的，不用特意指定。
15.1.3 消费者消息确认
RabbitMQ是阅后即焚机制，RabbitMQ确认消息被消费者消费后会立刻删除。
而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的：消费者获取消息后，应该向RabbitMQ发送ACK回执，表明自己已经处理消息。
设想这样的场景：
- 1）RabbitMQ投递消息给消费者
- 2）消费者获取消息后，返回ACK给RabbitMQ
- 3）RabbitMQ删除消息
- 4）消费者宕机，消息尚未处理
这样，消息就丢失了。因此消费者返回ACK的时机非常重要。
而SpringAMQP则允许配置三种确认模式：
- manual：手动ack，需要在业务代码结束后，调用api发送ack。
- auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack
- none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除
由此可知：
- none模式下，消息投递是不可靠的，可能丢失
- auto模式类似事务机制，出现异常时返回nack，消息回滚到mq；没有异常，返回ack
- manual：自己根据业务情况，判断什么时候该ack
一般，我们都是使用默认的auto即可。
① 演示none模式
修改consumer服务的application.yml文件，添加下面内容：
spring: |
修改consumer服务的SpringRabbitListener类中的方法，模拟一个消息处理异常：
|
测试可以发现，当消息处理抛异常时，消息依然被RabbitMQ删除了。
② 演示auto模式
再次把确认机制修改为auto:
spring: |
在异常位置打断点，再次发送消息，程序卡在断点时，可以发现此时消息状态为unack（未确定状态）
15.1.4 消费失败重试机制
当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，带来不必要的压力
① 本地重试
我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列。
修改consumer服务的application.yml文件，添加内容：
spring: |
重启consumer服务，重复之前的测试。可以发现：
- 在重试3次后，SpringAMQP会抛出异常AmqpRejectAndDontRequeueException，说明本地重试触发了
- 查看RabbitMQ控制台，发现消息被删除了，说明最后SpringAMQP返回的是ack，mq删除消息了
结论：
- 开启本地重试时，消息处理过程中抛出异常，不会requeue到队列，而是在消费者本地重试
- 重试达到最大次数后，Spring会返回ack，消息会被丢弃
② 失败策略
在之前的测试中，达到最大重试次数后，消息会被丢弃，这是由Spring内部机制决定的。
在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecovery接口来处理，它包含三种不同的实现：
- RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式
- ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队
- RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机
比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。
1）在consumer服务中定义处理失败消息的交换机和队列
|
2）定义一个RepublishMessageRecoverer，关联队列和交换机
|
完整代码：
package cn.itcast.mq.config; |
15.1.5 总结
如何确保RabbitMQ消息的可靠性？
- 开启生产者确认机制，确保生产者的消息能到达队列
- 开启持久化功能，确保消息未消费前在队列中不会丢失
- 开启消费者确认机制为auto，由spring确认消息处理成功后完成ack
- 开启消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理
15.2 死信交换机
15.2.1初识死信交换机
① 什么是死信交换机
什么是死信？
当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：
- 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
- 消息是一个过期消息，超时无人消费
- 要投递的队列消息满了，无法投递
如果这个包含死信的队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（Dead Letter Exchange，检查DLX）。
- 一个消息被消费者拒绝了，变成了死信，即simple.queue
- 因为simple.queue绑定了死信交换机dl.direct，因此死信会投递给这个交换机
- 如果这个死信交换机也绑定了一个队列，则消息最终会进入这个存放死信的队列dl.queue
另外，队列将死信投递给死信交换机时，必须知道两个信息：
- 死信交换机名称
- 死信交换机与死信队列绑定的RoutingKey
这样才能确保投递的消息能到达死信交换机，并且正确的路由到死信队列。
② 利用死信交换机接收死信（拓展）
在失败重试策略中，默认的RejectAndDontRequeueRecoverer会在本地重试次数耗尽后，发送reject给RabbitMQ，消息变成死信，被丢弃。
我们可以给simple.queue添加一个死信交换机，给死信交换机绑定一个队列。这样消息变成死信后也不会丢弃，而是最终投递到死信交换机，路由到与死信交换机绑定的队列。
我们在consumer服务中，定义一组死信交换机、死信队列：
// 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct |
也可以一键注解：
package cn.itcast.mq.listener; |
③ 总结
什么样的消息会成为死信？
- 消息被消费者reject或者返回nack
- 消息超时未消费
- 队列满了
死信交换机的使用场景是什么？
- 如果队列绑定了死信交换机，死信会投递到死信交换机；
- 可以利用死信交换机收集所有消费者处理失败的消息（死信），交由人工处理，进一步提高消息队列的可靠性。
15.2.2 TTL
一个队列中的消息如果超时未消费，则会变为死信，超时分为两种情况：
- 消息所在的队列设置了超时时间
- 消息本身设置了超时时间
① 接收超时死信的死信交换机
在consumer服务的SpringRabbitListener中，定义一个新的消费者，并且声明 死信交换机、死信队列：
|
② 声明一个队列，并且指定TTL
要给队列设置超时时间，需要在声明队列时配置x-message-ttl属性：
|
注意，这个队列设定了死信交换机为dl.ttl.direct
声明交换机，将ttl与交换机绑定：
|
发送消息，但是不要指定TTL：
|
因为队列的TTL值是10000ms，也就是10秒。所以消息发送与接收之间的时差刚好是10秒。
③ 发送消息时，设定TTL
在发送消息时，也可以指定TTL：
|
发送与接收的延迟只有5秒。说明当队列、消息都设置了TTL时，任意一个到期就会成为死信。
④ 总结
消息超时的两种方式是？
- 给队列设置ttl属性，进入队列后超过ttl时间的消息变为死信
- 给消息设置ttl属性，队列接收到消息超过ttl时间后变为死信
如何实现发送一个消息20秒后消费者才收到消息？
- 给消息的目标队列指定死信交换机
- 将消费者监听的队列绑定到死信交换机
- 发送消息时给消息设置超时时间为20秒
15.2.3 延迟队列
利用TTL结合死信交换机，我们实现了消息发出后，消费者延迟收到消息的效果。这种消息模式就称为延迟队列（Delay Queue）模式。
延迟队列的使用场景包括：
- 延迟发送短信
- 用户下单，如果用户在15 分钟内未支付，则自动取消
- 预约工作会议，20分钟后自动通知所有参会人员
因为延迟队列的需求非常多，所以RabbitMQ的官方也推出了一个插件，原生支持延迟队列效果。
这个插件就是DelayExchange插件。参考RabbitMQ的插件列表页面：https://www.rabbitmq.com/community-plugins.html
使用方式可以参考官网地址：https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq
① 安装DelayExchange插件
1）安装MQ
注：部署此插件需要把plugins挂载到磁盘中
docker run \ |
2）安装DelayExchange插件
官方的安装指南地址为：https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq
上述文档是基于linux原生安装RabbitMQ，然后安装插件。
因为我们之前是基于Docker安装RabbitMQ，所以下面我们会讲解基于Docker来安装RabbitMQ插件。
- 下载插件：RabbitMQ有一个官方的插件社区，地址为：https://www.rabbitmq.com/community-plugins.html
- 去对应的GitHub页面下载3.8.9版本的插件，地址为https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/3.8.9这个对应RabbitMQ的3.8.5以上版本。
3）上传插件
因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。如果不是基于Docker的同学，请参考第一章部分，重新创建Docker容器。
我们之前设定的RabbitMQ的数据卷名称为mq-plugins，所以我们使用下面命令查看数据卷：
docker volume inspect mq-plugins |
将插件上传到这个目录即可
4）安装插件
最后就是安装了，需要进入MQ容器内部来执行安装。我的容器名为mq，所以执行下面命令：
docker exec -it mq bash |
执行时，请将其中的 -it 后面的mq替换为你自己的容器名.
进入容器内部后，执行下面命令开启插件：
rabbitmq-plugins enable rabbitmq_delayed_message_exchange |
② DelayExchange原理
DelayExchange需要将一个交换机声明为delayed类型。当我们发送消息到delayExchange时，流程如下：
- 接收消息
- 判断消息是否具备x-delay属性
- 如果有x-delay属性，说明是延迟消息，持久化到硬盘，读取x-delay值，作为延迟时间
- 返回routing not found结果给消息发送者
- x-delay时间到期后，重新投递消息到指定队列
③ 使用DelayExchange
插件的使用也非常简单：声明一个交换机，交换机的类型可以是任意类型，只需要设定delayed属性为true即可，然后声明队列与其绑定即可。
1）声明DelayExchange交换机
基于注解方式（推荐）：
|
也可以基于@Bean的方式：
|
2）发送消息
发送消息时，一定要携带x-delay属性，指定延迟的时间：
|
④ 总结
延迟队列插件的使用步骤包括哪些？
- 声明一个交换机，添加delayed属性为true
- 发送消息时，添加x-delay头，值为超时时间
15.3 惰性队列
15.3.1 消息堆积问题
当生产者发送消息的速度超过了消费者处理消息的速度，就会导致队列中的消息堆积，直到队列存储消息达到上限。之后发送的消息就会成为死信，可能会被丢弃，这就是消息堆积问题。
解决消息堆积有两种思路：
- 增加更多消费者，提高消费速度。也就是我们之前说的work queue模式
- 扩大队列容积，提高堆积上限
要提升队列容积，把消息保存在内存中显然是不行的。
15.3.2 惰性队列
从RabbitMQ的3.6.0版本开始，就增加了Lazy Queues的概念，也就是惰性队列。惰性队列的特征如下：
- 接收到消息后直接存入磁盘而非内存
- 消费者要消费消息时才会从磁盘中读取并加载到内存
- 支持数百万条的消息存储
① 基于命令行设置lazy-queue
而要设置一个队列为惰性队列，只需要在声明队列时，指定x-queue-mode属性为lazy即可。可以通过命令行将一个运行中的队列修改为惰性队列：
rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues |
命令解读：
rabbitmqctl：RabbitMQ的命令行工具set_policy：添加一个策略Lazy：策略名称，可以自定义"^lazy-queue$"：用正则表达式匹配队列的名字'{"queue-mode":"lazy"}'：设置队列模式为lazy模式--apply-to queues：策略的作用对象，是所有的队列
② 基于@Bean声明lazy-queue
|
③ 基于@RabbitListener声明LazyQueue
|
15.3.3 总结
消息堆积问题的解决方案？
- 队列上绑定多个消费者，提高消费速度
- 使用惰性队列，可以再mq中保存更多消息
惰性队列的优点有哪些？
- 基于磁盘存储，消息上限高
- 没有间歇性的page-out，性能比较稳定
惰性队列的缺点有哪些？
- 基于磁盘存储，消息时效性会降低
- 性能受限于磁盘的IO
15.4 MQ集群
15.4.1 集群分类
RabbitMQ的是基于Erlang语言编写，而Erlang又是一个面向并发的语言，天然支持集群模式。RabbitMQ的集群有两种模式：
•普通集群：是一种分布式集群，将队列分散到集群的各个节点，从而提高整个集群的并发能力。
•镜像集群：是一种主从集群，普通集群的基础上，添加了主从备份功能，提高集群的数据可用性。
镜像集群虽然支持主从，但主从同步并不是强一致的，某些情况下可能有数据丢失的风险。因此在RabbitMQ的3.8版本以后，推出了新的功能：仲裁队列来代替镜像集群，底层采用Raft协议确保主从的数据一致性。
15.4.2 普通集群
① 集群结构和特征
普通集群，或者叫标准集群（classic cluster），具备下列特征：
- 会在集群的各个节点间共享部分数据，包括：交换机、队列元信息。不包含队列中的消息。
- 当访问集群某节点时，如果队列不在该节点，会从数据所在节点传递到当前节点并返回
- 队列所在节点宕机，队列中的消息就会丢失
② 部署
集群分类
在RabbitMQ的官方文档中，讲述了两种集群的配置方式：
- 普通模式：普通模式集群不进行数据同步，每个MQ都有自己的队列、数据信息（其它元数据信息如交换机等会同步）。例如我们有2个MQ：mq1，和mq2，如果你的消息在mq1，而你连接到了mq2，那么mq2会去mq1拉取消息，然后返回给你。如果mq1宕机，消息就会丢失。
- 镜像模式：与普通模式不同，队列会在各个mq的镜像节点之间同步，因此你连接到任何一个镜像节点，均可获取到消息。而且如果一个节点宕机，并不会导致数据丢失。不过，这种方式增加了数据同步的带宽消耗。
我们先来看普通模式集群，我们的计划部署3节点的mq集群：
| 主机名 | 控制台端口 | amqp通信端口 |
|---|---|---|
| mq1 | 8081 —> 15672 | 8071 —> 5672 |
| mq2 | 8082 —> 15672 | 8072 —> 5672 |
| mq3 | 8083 —> 15672 | 8073 —> 5672 |
集群中的节点标示默认都是：rabbit@[hostname]，因此以上三个节点的名称分别为：
- rabbit@mq1
- rabbit@mq2
- rabbit@mq3
获取cookie
RabbitMQ底层依赖于Erlang，而Erlang虚拟机就是一个面向分布式的语言，默认就支持集群模式。集群模式中的每个RabbitMQ 节点使用 cookie 来确定它们是否被允许相互通信。
要使两个节点能够通信，它们必须具有相同的共享秘密，称为Erlang cookie。cookie 只是一串最多 255 个字符的字母数字字符。
每个集群节点必须具有相同的 cookie。实例之间也需要它来相互通信。
我们先在之前启动的mq容器中获取一个cookie值，作为集群的cookie。执行下面的命令：
docker exec -it mq cat /var/lib/rabbitmq/.erlang.cookie |
可以看到cookie值如下：
FXZMCVGLBIXZCDEMMVZQ |
接下来，停止并删除当前的mq容器，我们重新搭建集群。
docker rm -f mq |
准备集群配置
在/tmp目录新建一个配置文件 rabbitmq.conf：
cd /tmp |
文件内容如下：
loopback_users.guest = false |
再创建一个文件，记录cookie
cd /tmp |
准备三个目录,mq1、mq2、mq3：
cd /tmp |
然后拷贝rabbitmq.conf、cookie文件到mq1、mq2、mq3：
# 进入/tmp |
启动集群
创建一个网络：
docker network create mq-net |
docker volume create
运行命令
docker run -d --net mq-net \ |
docker run -d --net mq-net \ |
docker run -d --net mq-net \ |
测试
在mq1这个节点上添加一个队列：
- Name : simple.queue
- Durability : Durable
- Node : rabbit@mq1
15.4.3 镜像集群
① 集群结构和特征
镜像集群：本质是主从模式，具备下面的特征：
- 交换机、队列、队列中的消息会在各个mq的镜像节点之间同步备份。
- 创建队列的节点被称为该队列的主节点，备份到的其它节点叫做该队列的镜像节点。
- 一个队列的主节点可能是另一个队列的镜像节点
- 所有操作都是主节点完成，然后同步给镜像节点
- 主宕机后，镜像节点会替代成新的主
② 部署
官方文档地址：https://www.rabbitmq.com/ha.html
镜像模式的特征
默认情况下，队列只保存在创建该队列的节点上。而镜像模式下，创建队列的节点被称为该队列的主节点，队列还会拷贝到集群中的其它节点，也叫做该队列的镜像节点。
但是，不同队列可以在集群中的任意节点上创建，因此不同队列的主节点可以不同。甚至，一个队列的主节点可能是另一个队列的镜像节点。
用户发送给队列的一切请求，例如发送消息、消息回执默认都会在主节点完成，如果是从节点接收到请求，也会路由到主节点去完成。镜像节点仅仅起到备份数据作用。
当主节点接收到消费者的ACK时，所有镜像都会删除节点中的数据。
总结如下：
- 镜像队列结构是一主多从（从就是镜像）
- 所有操作都是主节点完成，然后同步给镜像节点
- 主宕机后，镜像节点会替代成新的主（如果在主从同步完成前，主就已经宕机，可能出现数据丢失）
- 不具备负载均衡功能，因为所有操作都会有主节点完成（但是不同队列，其主节点可以不同，可以利用这个提高吞吐量）
镜像模式的配置
镜像模式的配置有3种模式：
| ha-mode | ha-params | 效果 |
|---|---|---|
| 准确模式exactly | 队列的副本量count | 集群中队列副本（主服务器和镜像服务器之和）的数量。count如果为1意味着单个副本：即队列主节点。count值为2表示2个副本：1个队列主和1个队列镜像。换句话说：count = 镜像数量 + 1。如果群集中的节点数少于count，则该队列将镜像到所有节点。如果有集群总数大于count+1，并且包含镜像的节点出现故障，则将在另一个节点上创建一个新的镜像。 |
| all | (none) | 队列在群集中的所有节点之间进行镜像。队列将镜像到任何新加入的节点。镜像到所有节点将对所有群集节点施加额外的压力，包括网络I / O，磁盘I / O和磁盘空间使用情况。推荐使用exactly，设置副本数为（N / 2 +1）。 |
| nodes | node names | 指定队列创建到哪些节点，如果指定的节点全部不存在，则会出现异常。如果指定的节点在集群中存在，但是暂时不可用，会创建节点到当前客户端连接到的节点。 |
这里我们以rabbitmqctl命令作为案例来讲解配置语法。
语法示例：
exactly模式
rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}' |
rabbitmqctl set_policy：固定写法ha-two：策略名称，自定义"^two\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以two.开头的队列名称'{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}': 策略内容"ha-mode":"exactly"：策略模式，此处是exactly模式，指定副本数量"ha-params":2：策略参数，这里是2，就是副本数量为2，1主1镜像"ha-sync-mode":"automatic"：同步策略，默认是manual，即新加入的镜像节点不会同步旧的消息。如果设置为automatic，则新加入的镜像节点会把主节点中所有消息都同步，会带来额外的网络开销
all模式
rabbitmqctl set_policy ha-all "^all\." '{"ha-mode":"all"}' |
ha-all：策略名称，自定义"^all\."：匹配所有以all.开头的队列名'{"ha-mode":"all"}'：策略内容"ha-mode":"all"：策略模式，此处是all模式，即所有节点都会称为镜像节点
nodes模式
rabbitmqctl set_policy ha-nodes "^nodes\." '{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}' |
rabbitmqctl set_policy：固定写法ha-nodes：策略名称，自定义"^nodes\."：匹配队列的正则表达式，符合命名规则的队列才生效，这里是任何以nodes.开头的队列名称'{"ha-mode":"nodes","ha-params":["rabbit@nodeA", "rabbit@nodeB"]}': 策略内容"ha-mode":"nodes"：策略模式，此处是nodes模式"ha-params":["rabbit@mq1", "rabbit@mq2"]：策略参数，这里指定副本所在节点名称
测试
我们使用exactly模式的镜像，因为集群节点数量为3，因此镜像数量就设置为2.
运行下面的命令：
docker exec -it mq1 rabbitmqctl set_policy ha-two "^two\." '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}' |
③ Java代码创建仲裁队列
|
15.4.4 SpringAMQP连接MQ集群
注意，这里用address来代替host、port方式
spring: |
15.4.5 集群扩容
① 加入集群
1）启动一个新的MQ容器：
docker run -d --net mq-net \ |
2）进入容器控制台：
docker exec -it mq4 bash |
3）停止mq进程
rabbitmqctl stop_app |
4）重置RabbitMQ中的数据：
rabbitmqctl reset |
5）加入mq1：
rabbitmqctl join_cluster rabbit@mq1 |
6）再次启动mq进程
rabbitmqctl start_app |
② 增加仲裁队列副本
我们先查看下quorum.queue这个队列目前的副本情况，进入mq1容器：
docker exec -it mq1 bash |
执行命令：
rabbitmq-queues quorum_status "quorum.queue" |
现在，我们让mq4也加入进来：
rabbitmq-queues add_member "quorum.queue" "rabbit@mq4" |
再次查看：
rabbitmq-queues quorum_status "quorum.queue" |
查看控制台，发现quorum.queue的镜像数量也从原来的 +2 变成了 +3
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 竹间小客！