微服务技术栈思维导图模板_ProcessOn思维导图、流程图

微服务治理

服务注册和发现

Nacos

Nacos安装与启动

Windows安装(linux安装请参照docker统一部署)： 1、下载安装包 在Nacos的GitHub页面，提供有下载链接，可以下载编译好的Nacos服务端或者源代码： GitHub主页：https://github.com/alibaba/nacos GitHub的Release下载页：https://github.com/alibaba/nacos/releases 2、解压 将这个包解压到任意非中文目录下 3、端口配置 Nacos的默认端口是8848，如果要修改Nacos的默认端口是可以进入nacos的conf目录，修改application.properties配置文件中的端口 4、启动 windows单机命令： startup.cmd -m standalone 5、访问 在浏览器输入地址：http://127.0.0.1:8848/nacos即可 默认的账号和密码都是nacos

Nacos服务的注册和发现

1）引入依赖 在cloud-demo父工程的pom文件中的<dependencyManagement>中引入SpringCloudAlibaba的依赖： <dependency>     <groupId>com.alibaba.cloud</groupId>     <artifactId>spring-cloud-alibaba-dependencies</artifactId>     <version>2.2.6.RELEASE</version>     <type>pom</type>     <scope>import</scope> </dependency> 然后在user-service和order-service中的pom文件中引入nacos-discovery依赖： <dependency>     <groupId>com.alibaba.cloud</groupId>     <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> </dependency> 2）配置nacos地址 在user-service和order-service的application.yml中添加nacos地址： application.yml spring:  cloud:   nacos:    server-addr: localhost:8848 3）重启 重启微服务后，登录nacos管理页面，可以看到微服务信息

Nacos服务的分级存储

Nacos服务的分级存储模型

1、一级是服务，例如userservice 2、二级是集群,例如，杭州，上海等 3、三级是实例，例如杭州机房的某台部署了userservice的服务器

为什么要实现服务的分级存储？

一个服务可以有多个实例，例如我们的user-service，可以有: 127.0.0.1:8081 127.0.0.1:8082 127.0.0.1:8083 假如这些实例分布于全国各地的不同机房，例如： 127.0.0.1:8081，在上海机房 127.0.0.1:8082，在上海机房 127.0.0.1:8083，在杭州机房 Nacos就将同一机房内的实例 划分为一个集群。 也就是说，user-service是服务，一个服务可以包含多个集群，如杭州、上海，每个集群下可以有多个实例，形成分级模型 当微服务互相访问时，应该尽可能访问同集群实例，因为本地访问速度更快。当本集群内不可用时，才访问其它集群。 杭州机房内的order-service应该优先访问同机房的user-service。

如何实现服务的分级存储？

给user-service配置集群 修改user-service的application.yml文件，添加集群配置： spring:   cloud:     nacos:       server-addr: localhost:8848       discovery:         cluster-name: HZ # 集群名称 重启两个user-service实例后，我们可以在nacos控制台查看结果

NacosRule负载均衡

根据权重负载均衡

在实际的部署中，会出现这样的场景： 服务设备性能有差异，部分实例所在机器性能较好，另一些较差，我们希望性能较好的机器承担更多的用户请求 Nacos提供了权重配置来控制访问频率，权重越大访问频率越高。 在Nacos控制台可以设置实例的权重值，首先选中实例后面的编辑按钮，点击编辑，然后设置权重(设置的是提供者的权重)。

配置负载均衡规则

修改order-service的application.yml文件，修改负载均衡规则： userservice:   ribbon:     NFLoadBalancerRuleClassName: com.alibaba.cloud.nacos.ribbon.NacosRule # 负载均衡规则 NacosRule默认优先选择同集群服务实例，当同集群服务实例找不到时，才会跨集群寻找，然后随机轮询服务实例

Nacos环境隔离

通过设置命名空间，可以将相同服务下的不同实例进行隔离，不同nameSpacoe下的服务之间是互不可见的 如何配置命名空间： 1、登陆Nacos控制台，点击命名空间菜单， 点击新建命名空间按钮 2、填写命名空间名和描述，ID可不填，会自动生成，点击确定。 3、修改order-service的application.yml ，添加 namespace spring:  cloud:   nacos:    discovery:     namespace:492a7d5d-237b-46a1-a99a-fa8e98e4b0f9    #命名空间，填ID

eureka

配置管理

什么是配置管理

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

Nacos

nacos统一配置原理

Nacos实现配置管理

控制台新增配置文件

1、登陆Nacos控制台 2、Nacos点击配置管理，选择配置列表，点击+号按钮新增 3、首先填写 DataID：userservice-dev.yaml DataID ：[服务名称]-[profile].[后缀名] 4、填写分组，默认值就行 5、配置文件格式，选择yaml 6、配置内容 7、保存

微服务拉取配置文件

1、引入Nacos的配置管理客户端依赖 <dependency>     <groupId>com.alibaba.cloud</groupId>     <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId> </dependency> 2、配置bootstrap.yml bootStrap.yml spring:   application:     name: userservice   profiles:     active: test # 环境   cloud:     nacos:       server-addr: nacos:8848 # nacos地址       discovery:         cluster-name: HZ       config:         file-extension: yaml # 文件后缀名

配置热更新

方式1：通过 Spring Cloud 原生注解 @RefreshScope 实现配置自动更新 @RestController @RequestMapping("/config") @RefreshScope public class ConfigController {     @Value("${useLocalCache:false}")     private boolean useLocalCache;     @RequestMapping("/get")     public boolean get() {         return useLocalCache;     } } 方式2：通过ConfigurationProperties注入自动刷新 @Data @Component @ConfigurationProperties(prefix = "pattern") public class PatternProperties {     private String dateformat;     private String envSharedValue;     private String name; } 注意：不是所有的配置都适合放在配置中心，维护起来比较麻烦，建议将一些关键参数， 需要运行时调整的参数放到nacos配置中心，一般都是自定义配置

多环境配置共享

微服务会从nacos配置中心读取两个配置文件： 1、[服务名]-[spring.profile.active].yaml，环境配置 2、[服务名].yaml，默认配置，多环境共享 优先级： [服务名]-[spring.profile.active].yaml > [服务名].yaml > 本地配置 优先级最高的生效

Nacos集群搭建

方案1：http://ip1:port/openAPI 直连ip模式： ip+端口进行部署，客户端直接连接Nacos的ip

这三种方案都是为了尽可能实现高可用，后两种方案除了基本的部署流程外更多侧重于实现高可用的工作上。 这里以第一种ip+端口的方式为大家介绍集群部署方式，当然ip+端口也有多种部署方式。 1ip+n端口+Nginx：普通玩家部署方式，没有过多服务器，单台服务器启动多个nacos实例，仅适合测试使用 nip+n端口+Nginx：RMB玩家部署方式，服务器资源充足，组建完美集群，实现更好的容灾与隔离 无论怎么部署，部署方式都是一样的，这里我以1ip+3端口+Nginx的方式进行集群搭建 1、搭建mysql集群，并初始化数据库表     nacos数据库表结构在nacos-server-1.4.1\nacos\conf下，nacos-mysql.sql中 2、下载解压naocs 3、修改集群配置、数据库配置     1）、修改Nacos-server目录conf/下的application.properties文件，添加mysql数据源     2）、修改集群配置     修改conf/下的cluster.conf.example文件，将其命名为cluster.conf，内容如下     # ip:port     10.1.8.27:8849     10.1.8.27:8850     10.1.8.27:8851     注：一定要记得将配置文件重命名为cluster.conf, 最好用实际ip，而非127.0.0.1，否则会出现问题 4、分别启动多个Nacos服务     执行如下三条命令     bash startup.sh -p 8849     bash startup.sh -p 8850     bash startup.sh -p 8851     启动成功后会打印如下一句话：     nacos is starting，you can check the /usr/local/nacos/nacos/logs/start.out     查看该目录下的start.out文件，可以看到如下日志     完成上面的配置后，已经基本完成集群搭建的90%了     这里我们可以通过Nginx配置，为Nacos提供统一的入口，来实现一个简单的负载均衡     Nginx配置如下执行命令 sudo nginx启动nginx     通过8848端口访问Nacos后台，此时Nginx会将请求分发至nacos-server下的地址中，这里默认的分发策略是线性轮询 5、nginx反向代理     upstream nacos-server {       server 127.0.0.1:8849;       server 127.0.0.1:8850;       server 127.0.0.1:8851;     }     server {       listen 8848;       server_name  localhost;       location /nacos/ {         proxy_pass http://nacos-server/nacos/;       }     }

方案2：http://Vip:port/openAPI 挂载虚拟IP模式： 配合KeepAlive，Nacos真实ip都挂载虚拟Ip下客户端访问Vip发起请求 当主Nacos宕机后，备用Nacos接管，实现高可用，

方案3：http://www.nacostest.com:port/openAPI 挂载虚拟IP+域名模式： 为虚拟ip绑定一个域名，当Nacos集群迁移时，客户端配置无需修改。

SpringCloudConifg

服务远程调用

OpenFeign

基于feign的远程调用

1使用 Feign的步骤如下： 1.引入依赖： <dependency>   <groupId>org.springframework.cloud</groupId>   <artifactId>spring-cloud-starter-openfeign</artifactId> </dependency> 2.在order-service的启动类添加注解开启Feign的功能： @EnableFeignClients 3、编写Feign客户端 @FeignClient("userservice") public interface UserClient{     GetMapper("/user/{id}")     User findById(PathVariable("id") Long id); } 主要是基于SpringMVC的注解来声明远程调用的信息，比如： •服务名称：userservice •请求方式：GET •请求路径：/user/{id} •请求参数：Long id •返回值类型：User 4、在service类中注入feign客户端对象，通过feign客户端对象发送请求 @Service public class OrderService {     @Autowired     private OrderMapper orderMapper;     @Autowired     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;     }

自定义Feign的配置

feign可以自定义的配置

修改feign配置的两种方式，这里以修改日志配置为例 方式一：配置文件方式 ①全局生效： feign:   client:     config:      default:  #这里用default就是全局配置，如果是写服务名称，则是针对某个微服务的配置        loggerLevel: FULL  #日志级别 ②局部生效： feign:   client:     config:       userservice:  #这里用default就是全局配置，如果是写服务名称，则是针对某个微服务的配置         loggerLevel:FULL#日志级别 配置Feign日志的方式二： java代码方式，需要先声明一个Bean： public class FeignClientConfiguration {   @Bean   public Logger.Level feignLogLevel(){     return Logger.Level.BASIC;   } } ①而后如果是全局配置，则把它放到@EnableFeignClients这个注解中： @EnableFeignClients(defaultConfiguration = FeignClientConfiguration.class) ②如果是局部配置，则把它放到@FeignClient这个注解中： @FeignClient(value ="userservice",configuration = FeignClientConfiguration.class）

feign的性能优化

1、使用连接池替代UrlConnection

feign底层默认实现的是UrlConnection，而urlConnection是不支持连接池的， 这就造成feign客户端每次请求是都要进行三次握手四次挥手，造成资源的浪费， 所以用httpClient来替代urlConnectionFeign添加HttpClient的支持： ① 引入依赖：  <dependency> <groupId>io.github.openfeign</groupId> <artifactId>feign-httpclient</artifactId> </dependency> ② 配置连接池： feign:  client:   config:    default:     loggerLevel: BASIC  #日志级别，BASIC就是基本的请求和响应信息  httpclient:   enabled: true  #开启feign对HttpClient的支持   max-connections: 200  #最大的连接数   max-connections-per-route: 50  #每个路径的最大连接数

2、日志级别，最好用basic或none

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

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

实现最佳实践方式二的步骤如下： 1. 首先创建一个module，命名为feign-api，然后引feign的starter依赖 2. 将order-service中编写的UserClient、User、DefaultFeignConfiguration都复制到feign-api项目中 3. 在order-service中引入feign-api的依赖 4. 修改order-service中的所有与上述三个组件有关的import部分，改成导入feign-api中的包 5. 重启测试 当定义的FeignClient不在SpringBootApplication的扫描包范围时，这些FeignClient无法使用。有两种方式解决： 方式一： 指定FeignClient所在包 @EnableFeignClients(basePackages ="cn.itcast.feign.clients") 方式二： 指定feignClient字节码 @EnableFeignClients(clients = {UserClient.class})

什么是FeignClient

FeignClient简化了请求的编写，且通过动态负载（Feign集成了Ribbon）进行选择要使用哪个服务进行消费，而这一切都由Spring动态配置实现，我们不用关心这些，只管使用方法即可。

Dubbo

负载均衡

ribbon负载均衡

ribbon负载均衡原理

ribbon负载均衡策略

RoundRobinRule： 简单轮询服务列表来选择服务器。它是Ribbon默认的负载均衡规则。

AvailabilityFilteringRule： 对以下两种服务器进行忽略： （1）在默认情况下， 这台服务器如果3次连接失败，这台服务就会被设置为“短路”状态。短路状态将持续30秒， 如果再次连接失败，短路的持续时间就会几何级地增加。 （2）并发数过高的服务器。如果一个服务器的并发连接数过高， 配置了AvailabilityFilteringRule规则的客户端也会将其忽略。并发连接数的上限，可以由客户端的 <clientName><clientConfigNameSpace>.ActiveConnectionsLimit属性进行配置。

WeightedResponseTimeRule： 为每一个服务器赋予一个权重值。服务器响应时间越长，这个服务器的权重就越小。 这个规则会随机选择服务器，这个权重值会影响服务器的选择。

ZoneAvoidanceRule： 以区域可用的服务器为基础进行服务器的 选择。使用Zone对服务器进行分类，这个Zone可以理解为一个机房、一个机架等。 而后再对Zone内的多个服务做轮询。

BestAvailableRule： 忽略那些短路的服务器，并选择并发数较低的服务器。

RandomRule： 随机选择一个可用的服务器。

RetryRule： 重试机制的选择逻辑

配置负载均衡规则

通过定义IRule实现可以修改负载均衡规则 ，有两种方式： 1.代码方式：在消费者服务中的OrderApplication类中，定义一个新的IRule： @Bean public IRule randomRule(){   return new RandomRule(); } 2.配置文件方式： 在消费者服务的application.yml文件中，添加新的配置也可以修改规则： userservice:  ribbon:   NFLoadBalancerRuleClassName:    com.netflix.loadbalancer.RandomRule  # 负载均衡规则

NacosRule负载均衡

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

网关路由

什么是网关路由

1、分布式服务架构、微服务架构与 API 网关     在微服务架构里，服务的粒度被进一步细分，各个业务服务可以被独立的设计、开发、测试、部署和管理。 这时，各个独立部署单元可以用不同的开发测试团队维护，可以使用不同的编程语言和技术平台进行设计，这就要求必须使用一种语言和平台无关的服务协议作为各个单元间的通讯方式。 2、API 网关的定义     网关的角色是作为一个 API 架构，用来保护、增强和控制对于 API 服务的访问。 API 网关是一个处于应用程序或服务（提供 REST API 接口服务）之前的系统，用来管理授权、访问控制和流量限制等，这样 REST API 接口服务就被 API 网关保护起来，对所有的调用者透明。因此，隐藏在 API 网关后面的业务系统就可以专注于创建和管理服务，而不用去处理这些策略性的基础设施。 3、API网关职能 ① 作为所有API接口服务请求的接入点 ② 作为所有后端服务的聚合点 ③ 实现安全、验证、路由、过滤、流控等策略 ④ 对所有API服务和策略进行统一管理 4、API 网关的分类与功能 ① 流量网关，关注稳定与安全 全局流控，日志统计，防止sql注入，防止web攻击，屏蔽工具扫描，黑白IP名单，证书/加密处理 ② 业务网关，提供更好的服务 服务级别流控，服务降级与熔断，路由和负载均衡、灰度策略，服务过滤、聚合与发现，权限验证和用户等级策略，业务规则与参数校验，多级缓存策略

gateway

搭建网关服务

搭建网关服务的步骤： 1.创建新的module，引入SpringCloudGateway的依赖和nacos的服务发现依赖：  <dependency>   <groupId>org.springframework.cloud</groupId>   <artifactId>spring-cloud-starter-gateway</artifactId> </dependency>  <dependency>   <groupId>com.alibaba.cloud</groupId>   <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId> </dependency> 2.编写路由配置及nacos地址 server:   port: 10010  #网关端口 spring:   application:     name: gateway  #服务名称   cloud:     nacos:       server-addr: localhost:8848  # nacos地址   gateway:     routes:  #网关路由配置       - id: user-service  #路由id，自定义，只要唯一即可            # uri: http://127.0.0.1:8081 #路由的目标地址http就是固定地址         uri: lb://userservice  #路由的目标地址lb就是负载均衡，后面跟服务名称           predicates: #路由断言，也就是判断请求是否符合路由规则的条件           - Path=/user/**  #这个是按照路径匹配，只要以/user/开头就符合要求                filters:                     - AddRequestHeader=Truth,Itcast is freaking awesome #过滤器     default-filters:         - AddRequestHeader=Truth,Itcast is freaking awesome!!  #全局过滤器

断言工厂

网关路由可以配置的内容包括： 1、路由id：路由唯一标示 2、uri：路由目的地，支持lb和http两种 3、predicates：路由断言，判断请求是否符合要求，符合则转发到路由目的地 4、filters：路由过滤器，处理请求或响应 我们在配置文件中写的断言规则只是字符串，这些字符串会被Predicate Factory读取并处理，转变为路由判断的条件 例如Path=/user/**是按照路径匹配，这个规则是由 org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory 类来处理的像这样的断言工厂在SpringCloudGateway还有十几个

过滤器

路由过滤器和defaultfilters

GatewayFilter是网关中提供的一种过滤器。 过滤器的作用是什么？ ① 对路由的请求或响应做加工处理，比如添加请求头 ② 配置在路由下的过滤器只对当前路由的请求生效 defaultFilters的作用是什么？ ① 对所有路由都生效的过滤器 Spring提供了31种不同的路由过滤器工厂。例如： AddRequestHeader：给当前请求添加一个请求头 RemoveRequestHeader：移除请求中的一个请求头 AddResponseHeader：给响应结果中添加一个响应头 RemoveResponseHeader：从响应结果中移除有一个响应头 RequestRateLimiter：限制请求的流量 ......

全局过滤器

全局过滤器的作用也是处理一切进入网关的请求和微服务响应，与GatewayFilter的作用一样。 区别在于GatewayFilter通过配置定义，处理逻辑是固定的。而GlobalFilter的逻辑需要自己写代码实现。 定义方式是实现GlobalFilter接口。 public interface GlobalFilter { /** *  处理当前请求，有必要的话通过{ @linkGatewayFilterChain }将请求交给下一个过滤器处理 * * @param exchange请求上下文，里面可以获取Request、Response等信息 * @param chain用来把请求委托给下一个过滤器 * @return{@codeMono<Void>}返回标示当前过滤器业务结束 */     Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain); }

自定义类，实现GlobalFilter接口，添加@Order注解： @Order(-1)  //注解@Order的作用是定义Spring IOC容器中Bean的执行顺序的优先级,值越小优先级越高 @Component public class AuthorizeFilter implements GlobalFilter {     @Override     public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {         // 1.获取请求参数         MultiValueMap<String, String> params = exchange.getRequest().getQueryParams();         // 2.获取authorization参数        String auth = params.getFirst("authorization");         // 3.校验         if("admin".equals(auth)) {             //放行             return chain.filter(exchange);         }         // 4.拦截         // 4.1.禁止访问         exchange.getResponse().setStatusCode(HttpStatus.FORBIDDEN);         // 4.2.结束处理         return exchange.getResponse().setComplete();     } }

过滤器执行顺序

请求进入网关会碰到三类过滤器： 当前路由的过滤器、DefaultFilter、GlobalFilter请求路由后，会将当前路由过滤器和DefaultFilter、GlobalFilter，合并到一个过滤器链（集合）中，排序后依次执行每个过滤器。 每一个过滤器都必须指定一个int类型的order值，order值越小，优先级越高，执行顺序越靠前。 GlobalFilter通过实现Ordered接口，或者添加@Order注解来指定order值，由我们自己指定路由过滤器和defaultFilter的order由Spring指定，默认是按照声明顺序从1递增。 当过滤器的order值一样时，会按照defaultFilter > 路由过滤器 > GlobalFilter的顺序执行。

Zuul

DevOps

Docker原理

大型项目组件较多，运行环境也较为复杂，部署时会碰到一些问题： • 依赖关系复杂，容易出现兼容性问题 • 开发、测试、生产环境有差异 Docker如何解决依赖的兼容问题的？ • 将应用的Libs（函数库）、Deps（依赖）、配置与应用一起打包 • 将每个应用放到一个隔离容器去运行，避免互相干扰 不同环境的操作系统不同，Docker如何解决？ 我们先来了解下操作系统结构，内核与硬件交互，提供操作硬件的指令系统应用封装内核指令为函数， 便于程序员调用用户程序基于系统函数库实现功能。 Docker如何解决不同系统环境的问题？ • Docker将用户程序与所需要调用的系统(比如Ubuntu)函数库一起打包 • Docker运行到不同操作系统时，直接基于打包的库函数，借助于操作系统的Linux内核来运行 Docker如何解决大型项目依赖关系复杂，不同组件依赖的兼容性问题？ • Docker允许开发中将应用、依赖、函数库、配置一起打包，形成可移植镜像 • Docker应用运行在容器中，使用沙箱机制，相互隔离Docker如何解决开发、测试、生产环境有差异的问题 • Docker镜像中包含完整运行环境，包括系统函数库，仅依赖系统的Linux内核，因此可以在任意Linux操作系统上运行

Docker架构

Docker和DockerHub

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

镜像和容器

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

docker架构

Docker是一个CS架构的程序，由两部分组成： 服务端(server)：Docker守护进程，负责处理Docker指令，管理镜像、容器等 客户端(client)：通过命令或RestAPI向Docker服务端发送指令。可以在本地或远程向服务端发送指

Docker使用

安装Docker

1.1.卸载

如果之前安装过旧版本的Docker，可以使用下面命令卸载： yum remove docker \                   docker-client \                   docker-client-latest \                   docker-common \                   docker-latest \                   docker-latest-logrotate \                   docker-logrotate \                   docker-selinux \                   docker-engine-selinux \                   docker-engine \                   docker-ce

1.2.安装docker

首先需要大家虚拟机联网，安装yum工具 yum install -y yum-utils \            device-mapper-persistent-data \            lvm2 --skip-broken 然后更新本地镜像源： # 设置docker镜像源 yum-config-manager \     --add-repo \     https://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo      sed -i 's/download.docker.com/mirrors.aliyun.com\/docker-ce/g' /etc/yum.repos.d/docker-ce.repo yum makecache fast 然后输入命令： yum install -y docker-ce docker-ce为社区免费版本。稍等片刻，docker即可安装成功。

1.3.启动docker

Docker应用需要用到各种端口，逐一去修改防火墙设置。非常麻烦，因此建议大家直接关闭防火墙！ 启动docker前，一定要关闭防火墙后！！ # 关闭 systemctl stop firewalld # 禁止开机启动防火墙 systemctl disable firewalld 通过命令启动docker： systemctl start docker  # 启动docker服务 systemctl stop docker  # 停止docker服务 systemctl restart docker  # 重启docker服务 然后输入命令，可以查看docker版本： docker -v

1.4.配置镜像加速

docker官方镜像仓库网速较差，我们需要设置国内镜像服务： 参考阿里云的镜像加速文档： https://cr.console.aliyun.com/cn-hangzhou/instances/mirrors

Docker基本操作

镜像操作命令

• 镜像名称[name]一般分两部分组成：[repository]:[tag]。 • 在没有指定tag时，默认是latest，代表最新版本的镜像 • docker images 查看所有镜像 • docker rmi [name] 删除指定镜像    例 docker rmi redis:latest • docker pull 拉取镜像    例 docker pull redis:latest • docker push 上传镜像 • docker save 导出镜像到本地   例 docker save -o redis.tar redis:latest • docker load 从本地导入镜像   例 docker load -i redis.tar  • docker xx help xx代表命令 通过help查看帮助文档

容器操作命令

• docker run 创建并运行容器  docker run --name containerName -p 80:80 -d nginx 命令解读： docker run：创建并运行一个容器 --name :给容器起一个名字，比如叫做mn -p：将宿主机端口与容器端口映射，冒号左侧是宿主机端口，右侧是容器端口 -d：后台运行容器 nginx：镜像名称，例如nginx  • docker exec进入容器执行命令  docker exec -it mn bash 进入容器。进入我们刚刚创建的nginx容器的命令为： 命令解读： docker exec：进入容器内部，执行一个命令 -it :给当前进入的容器创建一个标准输入、输出终端，允许我们与容器交互 mn：要进入的容器的名称 bash：进入容器后执行的命令，bash是一个linux终端交互命令  • docker logs 查看容器日志的命令  添加-f参数可以持续查看日志  • docker ps  查看运行中的容器状态 添加 -all参数可以查看所有容器包括未运行的 •docker rm 删除容器： 不能删除运行中的容器，除非添加-f参数

数据卷

为什么要使用数据卷

容器与数据耦合的问题 1、不便于修改 当我们要修改Nginx的html内容时，需要进入容器内部修改，很不方便。 2、数据不可复用 在容器内的修改对外是不可见的。所有修改对新创建的容器是不可复用的。 3、升级维护困难 数据在容器内，如果要升级容器必然删除旧容器，所有数据都跟着删除了 数据卷（volume）是一个虚拟目录，指向宿主机文件系统中的某个目录。

操作数据卷

数据卷操作的基本语法如下： docker volume命令是数据卷操作， 根据命令后跟随的command来确定下一步的操作： create  创建一个volume inspect  显示一个或多个volume的信息 ls  列出所有的volume prune  删除未使用的volume rm  删除一个或多个指定的volume 例： ①创建数据卷 docker volume create html ②查看所有数据 docker volume ls ③查看数据卷详细信息卷 docker volume inspect html

挂载数据卷

我们在创建容器时，可以通过-v参数来挂载一个数据卷到某个容器目录 举例说明： docker run --name mn -v html:/root/html -p 8080:80nginx  命令解释： docker run：就是创建并运行容器 -- name mn：给容器起个名字叫mn -v html:/root/htm：把html数据卷挂载到容器内的/root/html这个目录中 -p 8080:80：把宿主机的8080端口映射到容器内的80端口 nginx：镜像名称 之后我们就可以进入/root/目录用vim直接修改html文件里

自定义镜像

镜像结构

镜像是将应用程序及其需要的系统函数库、环境、配置、依赖打包而成。 镜像结构 入口（Entrypoint） 镜像运行入口，一般是程序启动的脚本和参数 基础镜像（BaseImage） 应用依赖的系统函数库、环境、配置、文件等 层（Layer） 在BaseImage基础上添加安装包、依赖、配置等，每次操作都形成新的一层。

Dockerfile

什么是Dockerfile

Dockerfile就是一个文本文件，其中包含一个个的指令(Instruction)，用指令来说明要执行什么操作来构建镜像。 每一个指令都会形成一层Layer。 更详细语法说明，请参考官网文档：https://docs.docker.com/engine/reference/builder

案例

步骤

步骤1：新建一个空文件夹docker-demo                mkdir /tmp/docker-demo  步骤2：将需要上传的jar包文件放到docker-demo这个目录              用xftp上传 步骤3：将需要上传的jdk8.tar.gz文件放到docker-demo这个目录             用xftp上传 步骤4：将写好的Dockerfile到docker-demo这个目录             用xftp上传 步骤5：进入docker-demo 运行命令：docker build -t javaweb:1.0 .             cd /tmp/docker-demo              docker build -t javaweb:1.0 . **注意点代表dockerfile所在目录

DockerFlie 文件内容

# 指定基础镜像 FROM ubuntu:16.04 # 配置环境变量，JDK的安装目录 ENV JAVA_DIR=/usr/local # 拷贝jdk和java项目的包 COPY ./jdk8.tar.gz $JAVA_DIR/ COPY ./docker-demo.jar /tmp/app.jar # 安装JDK RUN cd $JAVA_DIR \  && tar -xf ./jdk8.tar.gz \  && mv ./jdk1.8.0_144 ./java8 # 配置环境变量 ENV JAVA_HOME=$JAVA_DIR/java8 ENV PATH=$PATH:$JAVA_HOME/bin # 暴露端口 EXPOSE 8090 # 入口，java项目的启动命令 ENTRYPOINT java -jar /tmp/app.jar

Docker-Compose

下载安装

2.CentOS7安装DockerCompose 2.1.下载 Linux下需要通过命令下载： # 安装 curl -L https://github.com/docker/compose/releases/download/1.23.1/docker-compose-`uname -s`-`uname -m` > /usr/local/bin/docker-compose 如果下载速度较慢，或者下载失败，可以使用课前资料提供的docker-compose文件： image-20210417133020614 上传到/usr/local/bin/目录也可以。 2.2.修改文件权限 修改文件权限： # 修改权限 chmod +x /usr/local/bin/docker-compose 2.3.Base自动补全命令： # 补全命令 curl -L https://raw.githubusercontent.com/docker/compose/1.29.1/contrib/completion/bash/docker-compose > /etc/bash_completion.d/docker-compose 如果这里出现错误，需要修改自己的hosts文件： echo "199.232.68.133 raw.githubusercontent.com" >> /etc/hosts

Docker-Compose使用

将之前学习的cloud-demo微服务集群利用DockerCompose部署实现思路如下： ① 准备好要上传的cloud-demo文件夹，里面编写好docker-compose文件 ② 修改自己的cloud-demo项目，将数据库、nacos地址都命名为docker-compose中的服务名 ③ 使用maven打包工具，将项目中的每个微服务都打包为app.jar ④ 将打包好的app.jar拷贝到cloud-demo中的每一个对应的子目录中（每个子目录里是jar包和DockerFile文件） ⑤ 将cloud-demo上传至虚拟机，利用docker-compose up -d来部署

Compose文件内容

version: "3.2" services:   nacos:     image: nacos/nacos-server     environment:       MODE: standalone     ports:       - "8848:8848"   mysql:     image: mysql:8.0.11     environment:       MYSQL_ROOT_PASSWORD: 123456   userservice:     build: ./user-service   #各微服务的接口不会对外暴露，只暴露网关接口   userservice1:     build: ./user-service1   userservice3:     build: ./user-service3   orderservice:     build: ./order-service   gateway:     build: ./gateway     ports:       - "10010:10010"

什么是Docker-Compose

• Docker Compose可以基于Compose文件帮我们快速的部署分布式应用，而无需手动一个个创建和运行容器！ • Compose文件是一个文本文件，通过指令定义集群中的每个容器如何运行。

Docker镜像仓库

创建docker私有镜像仓库

3.1.简化版镜像仓库 Docker官方的Docker Registry是一个基础版本的Docker镜像仓库，具备仓库管理的完整功能，但是没有图形化界面。 搭建方式比较简单，命令如下： docker run -d \     --restart=always \     --name registry \     -p 5000:5000 \     -v registry-data:/var/lib/registry \     registry 命令中挂载了一个数据卷registry-data到容器内的/var/lib/registry 目录，这是私有镜像库存放数据的目录。 访问http://YourIp:5000/v2/_catalog 可以查看当前私有镜像服务中包含的镜像 3.2.带有图形化界面版本 使用DockerCompose部署带有图象界面的DockerRegistry，命令如下： version: '3.0' services:   registry:     image: registry     volumes:       - ./registry-data:/var/lib/registry   ui:     image: joxit/docker-registry-ui:static     ports:       - 8080:80     environment:       - REGISTRY_TITLE=传智教育私有仓库       - REGISTRY_URL=http://registry:5000     depends_on:       - registry 3.3.配置Docker信任地址 我们的私服采用的是http协议，默认不被Docker信任，所以需要做一个配置： # 打开要修改的文件 vi /etc/docker/daemon.json # 添加内容： "insecure-registries":["http://192.168.150.101:8080"] # 重加载 systemctl daemon-reload # 重启docker systemctl restart docker

私有镜像的推送或拉取

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

异步通信

同步异步

同步调用的优缺点

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

异步调用的优缺点

优点： • 吞吐量提升：无需等待订阅者处理完成，响应更快速 • 故障隔离：服务没有直接调用，不存在级联失败问题 • 调用间没有阻塞，不会造成无效的资源占用 • 耦合度极低，每个服务都可以灵活插拔，可替换 • 流量削峰：不管发布事件的流量波动多大，都由Broker接收，订阅者可以按照自己的速度去处理事件 缺点： • 依赖于Broker的可靠性、安全性、吞吐能力 • 架构复杂了，业务没有明显的流程线，不好追踪管理

几种常见的MQ

MQ（MessageQueue），中文是消息队列，字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。 追求可用性：Kafka、 RocketMQ 、RabbitMQ 追求可靠性：RabbitMQ、RocketMQ 追求吞吐能力：RocketMQ、Kafka 追求消息低延迟：RabbitMQ、Kafka

RabbitMQ入门

创建并运行RabbitMQ

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

AMQP和RabbitMQ

提到RabbitMQ，就不得不提AMQP协议。AMQP协议是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。 先了解一下AMQP协议中间的几个重要概念： Server：接收客户端的连接，实现AMQP实体服务。 Connection：连接，应用程序与Server的网络连接，TCP连接。 Channel：信道，消息读写等操作在信道中进行。客户端可以建立多个信道，每个信道代表一个会话任务。 Message：消息，应用程序和服务器之间传送的数据，消息可以非常简单，也可以很复杂。有Properties和Body组成。Properties为外包装，可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body就是消息体内容。 Virtual Host：虚拟主机，用于逻辑隔离。一个虚拟主机里面可以有若干个Exchange和Queue，同一个虚拟主机里面不能有相同名称的Exchange或Queue。 Exchange：交换器，接收消息，按照路由规则将消息路由到一个或者多个队列。如果路由不到，或者返回给生产者，或者直接丢弃。RabbitMQ常用的交换器常用类型有direct、topic、fanout、headers四种，后面详细介绍。 Binding：绑定，交换器和消息队列之间的虚拟连接，绑定中可以包含一个或者多个RoutingKey。 RoutingKey：路由键，生产者将消息发送给交换器的时候，会发送一个RoutingKey，用来指定路由规则，这样交换器就知道把消息发送到哪个队列。路由键通常为一个“.”分割的字符串，例如“com.rabbitmq”。 Queue：消息队列，用来保存消息，供消费者消费。 我们完全可以直接使用 Connection 就能完成信道的工作，为什么还要引入信道呢? 试想这样一个场景， 一个应用程序中有很多个线程需要从 RabbitMQ 中消费消息，或者生产消息，那么必然需要建立很多个 Connection，也就是许多个 TCP 连接。然而对于操作系统而言，建立和销毁 TCP 连接是非常昂贵的开销，如果遇到使用高峰，性能瓶颈也随之显现。 RabbitMQ 采用 TCP 连接复用的方式，不仅可以减少性能开销，同时也便于管理 。

MQ消息模型

RabbitMQ常用的交换器类型有direct、topic、fanout、headers四种。 Direct Exchange 该类型的交换器将所有发送到该交换器的消息被转发到RoutingKey指定的队列中，也就是说路由到BindingKey和RoutingKey完全匹配的队列中。 Topic Exchange 该类型的交换器将所有发送到Topic Exchange的消息被转发到所有RoutingKey中指定的Topic的队列上面。 Exchange将RoutingKey和某Topic进行模糊匹配，其中“”用来匹配一个词，“#”用于匹配一个或者多个词。例如“com.#”能匹配到“com.rabbitmq.oa”和“com.rabbitmq”；而"login."只能匹配到“com.rabbitmq”。 Fanout Exchange 该类型不处理路由键，会把所有发送到交换器的消息路由到所有绑定的队列中。优点是转发消息最快，性能最好。 Headers Exchange 该类型的交换器不依赖路由规则来路由消息，而是根据消息内容中的headers属性进行匹配。headers类型交换器性能差，在实际中并不常用。

RabbitMQ的基本使用

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

SpringAMQP

什么是AMQP

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

什么是Spring AMQP

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

如何使用Spring AMQP实现基础的消息队列功能

1、在父工程中引入 Spring AMQP依赖  <dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter-amqp</artifactId> </dependency> 2、在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列       ① 在publisher服务中编写application.yml，添加mq连接信息：        spring:           rabbitmq:           host: 192.168.150.101  #主机名           port: 5672  #端口           virtual-host:/  #虚拟主机           username: itcast#用户名           password:123321  #密码       ② 在publisher服务中新建一个测试类，编写测试方法：            public classSpringAmqpTest {                 @Autowired                  private RabbitTemplate;                  public void testSimpleQueue() {                      String queueName = "simple.queue";                      String message = "hello, spring amqp!";                      rabbitTemplate.convertAndSend(queueName, message);           } 3、在consumer服务中编写消费逻辑，绑定simple.queue这个队列       ① 在consumer服务中编写application.yml，添加mq连接信息：（同2①）       ② 在consumer服务中新建一个类，编写消费逻辑：            @Component             public class SpringRabbitListener {                 @RabbitListener(queues ="simple.queue")                  public void listenSimpleQueueMessage(String msg)throwsInterruptedException {                        System.out.println("spring消费者接收到消息：【"+ msg +"】");                  }            }

消息预取限制

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

Work Queue工作队列

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

消息发布订阅模式

FanoutExchange

SpringAMQP实现

利用SpringAMQP演示FanoutExchange的使用 实现思路如下： 1. 在consumer服务常见一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding，代码如下：     @Configuration     public class FanoutConfig {       //声明 FanoutExchange 交换机       @Bean       public FanoutExchange fanoutExchange(){         return new FanoutExchange("itcast.fanout");       }       //声明第1个队列       @Bean       public Queue fanoutQueue1(){         return newQueue("fanout.queue1");       }       //绑定队列1和交换机       @Bean       public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){         return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);       }       //...略，以相同方式声明第2个队列，并完成绑定     } 2. 在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2     @RabbitListener(queues ="fanout.queue1")     public void listenFanoutQueue1(String msg) {       System.out.println("消费者1接收到Fanout消息：【"+ msg +"】");     }     @RabbitListener(queues ="fanout.queue2")     public void listenFanoutQueue2(String msg) {       System.out.println("消费者2接收到Fanout消息：【"+ msg +"】");     } 3. 在publisher中编写测试方法，向itcast.fanout发送消息     @Test     public void testFanoutExchange() {       //队列名称       String exchangeName ="itcast.fanout";       //消息       String message ="hello, everyone!";       //发送消息，参数分别是：交互机名称、RoutingKey（暂时为空）、消息       rabbitTemplate.convertAndSend(exchangeName,"",message);     }

交换机(Exchange)的作用

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

声明队列、交换机、绑定关系的Bean是什么

• Queue • FanoutExchange • Binding

DirectExchange

SpringAMQ实现

利用SpringAMQP演示DirectExchange的使用实现思路如下：(下面这种写法，可以不用去config中声明queue、Exchange) 1. 利用@RabbitListener声明Exchange、Queue、RoutingKey 2. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2     @RabbitListener(bindings = @QueueBinding(             value = @Queue(name = "direct.queue1"),             exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),             key = {"red", "blue"}     ))     public void listenDirectQueue1(String msg){         System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");     }     @RabbitListener(bindings = @QueueBinding(             value = @Queue(name = "direct.queue2"),             exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),             key = {"red", "yellow"}     ))     public void listenDirectQueue2(String msg){         System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");     }     @RabbitListener(bindings = @QueueBinding(             value = @Queue(name = "topic.queue1"),             exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),             key = "china.#"     )) 3. 在publisher中编写测试方法，向itcast. direct发送消息     @Test     public void testSendDirectExchange() {         // 交换机名称         String exchangeName = "itcast.direct";         // 消息         String message = "hello, red!";         // 发送消息         rabbitTemplate.convertAndSend(exchangeName, "red", message);     }

Direct Exchange路由模式

Direct Exchange 会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。 每一个Queue都与Exchange设置一个BindingKey 发布者发送消息时，指定消息的RoutingKey Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

Direct交换机与Fanout交换机的差异

• Fanout交换机将消息路由给每一个与之绑定的队列 • Direct交换机根据RoutingKey判断路由给哪个队列 • 如果多个队列具有相同的RoutingKey，则与Fanout功能类似

基于@RabbitListener注解声明队列和交换机的常见注解

• @Queue • @Exchange

TopicExchange

SpringAMQ实现

1. 并利用@RabbitListener声明Exchange、Queue、RoutingKey 2. 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2     public void listenTopicQueue1(String msg){         System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");     }     @RabbitListener(bindings = @QueueBinding(             value = @Queue(name = "topic.queue2"),             exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),             key = "#.news"     ))     public void listenTopicQueue2(String msg){         System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");     } 3. 在publisher中编写测试方法，向itcast. topic发送消息

Direct交换机与Topic交换机的差异

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以.分割。 Queue与Exchange指定BindingKey时可以使用通配符： ＃：代指0个或多个单词 *：代指一个单词 例如： china.news 代表有中国的新闻消息；china.weather 代表中国的天气消息； japan.news 则代表日本新闻 japan.weather 代表日本的天气消息；

消息转换器

消息转换器是做什么的

在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息， SpringAMQP的消息转换器，会帮我们序列化为字节后发送。

SpringAMQ实现

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。 而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。 如果要修改只需要定义一个MessageConverter类型的Bean即可。推荐用JSON方式序列化，步骤如下： • 我们在publisher服务引入依赖     <dependency>         <groupId>com.fasterxml.jackson.core</groupId>         <artifactId>jackson-databind</artifactId>     </dependency> • 我们在publisher服务声明MessageConverter：    @Bean     public MessageConverter jsonMessageConverter(){         return new Jackson2JsonMessageConverter();     } • 我们在consumer服务引入Jackson依赖：    <dependency>        <groupId>com.fasterxml.jackson.core</groupId>        <artifactId>jackson-databind</artifactId>    </dependency> • 我们在consumer服务定义MessageConverter：     @Bean     public MessageConverter jsonMessageConverter(){         return new Jackson2JsonMessageConverter();     } • 然后定义一个消费者，监听object.queue队列并消费消息：     @RabbitListener(queues ="object.queue")     public void listenObjectQueue(Map<String, Object> msg) {         System.out.println("收到消息：【"+ msg +"】");     }

消费者限流

ElasticSearch（分布式搜索）

了解ElasticSearch

正向索引倒排索引

文档

索引

映射

安装ES

部署单点ES

子主题

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

部署kibana

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

IK分词器

安装IK分词器

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

IK分词器拓展词库

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

IK分词器停用词库

要禁用某些敏感词条，只需要修改一个ik分词器目录中的config目录中的IkAnalyzer.cfg.xml文件： <?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd"> <properties>         <comment>IK Analyzer 扩展配置</comment>                  <entry key="ext_dict">ext.dic</entry>                   <entry key="ext_stopwords">stopword.dic</entry> </properties> <?xml version="1.0" encoding="UTF-8"?> 然后在名为stopword.dic的文件中，添加想要拓展的词语即可： 习大大

索引库（库）操作

mapping属性

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

创建索引库

ES中通过Restful请求操作索引库、文档。请求内容用DSL语句来表示。 创建索引库和mapping的DSL语法如下： PUT /索引库名称 {   "mappings": {     "properties": {       "字段名":{         "type": "text",         "analyzer": "ik_smart"       },       "字段名2":{         "type": "keyword",         "index": "false"       },       "字段名3":{         "properties": {           "子字段":{             "type": "keyword"           }         }       },     // ...略     }   } } 例： PUT /heima {   "mappings": {     "properties": {       "info":{         "type": "text",         "analyzer": "ik_smart"       },       "email":{         "type": "keyword",         "index": "false"       },       "name":{         "properties": {           "firstName": {             "type": "keyword"           }         }       },     // ...略     }   } }

查看、删除索引库

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

修改索引库

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

文档（表）操作

添加文档

新增文档的DSL语法如下： POST /索引库名/_doc/文档id {     "字段1": "值1",     "字段2": "值2",     "字段3": {         "子属性1": "值3",         "子属性2": "值4"        },     // ... } 示例 POST /heima/_doc/1 {     "info": "黑马程序员Java讲师",     "email": "zy@itcast.cn",     "name": {         "firstName": "云",         "lastName": "赵"      } }

查看、删除文档

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

修改文档

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

DSL查询语法

DSL Query的分类

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

DSL Query基本语法

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

全文检索查询

match查询： 全文检索查询的一种，会对用户输入内容分词，然后去倒排索引库检索，语法： GET /indexName/_search {   "query": {     "match": {       "FIELD": "TEXT"     }   } } multi_match： 与match查询类似，只不过允许同时查询多个字段，语法： GET /indexName/_search {   "query": {     "multi_match": {       "query": "TEXT",       "fields": ["FIELD1", " FIELD12"]     }   } }

精确查询

精确查询一般是查找keyword、数值、日期、boolean等类型字段。所以不会对搜索条件分词。 常见的有： • term：根据词条精确值查询 • range：根据值的范围查询 精确查询常见的有term查询和range查询。语法如下： // term查询 GET /indexName/_search {   "query": {     "term": {       "FIELD": {         "value": "VALUE"       }     }   } } // range查询 GET /indexName/_search {   "query": {     "range": {       "FIELD": {         "gte": 10,         "lte": 20       }     }   } }

地理查询

根据经纬度查询，官方文档。例如： • geo_bounding_box：查询geo_point值落在某个矩形范围的所有文档 // geo_bounding_box查询 GET /indexName/_search {   "query": {     "geo_bounding_box": {       "FIELD": {         "top_left": {           "lat": 31.1,           "lon": 121.5         },         "bottom_right": {           "lat": 30.9,           "lon": 121.7         }       }     }   } } • geo_distance：查询到指定中心点小于某个距离值的所有文档 // geo_distance查询 GET /indexName/_search {   "query": {     "geo_distance": {       "distance": "15km",       "FIELD": "31.21,121.5"     }   } }

复合查询

Function Score Query

使用function score query，可以修改文档的相关性算分（query score），根据新得到的算分排序。 GET /hotel/_search {   "query": {     "function_score": {       "query": { "match": {"all": "外滩"} }, # 原始查询条件，搜索文档并根据相关性打分(queryscore)       "functions": [         {           "filter": {"term": {"id": "1"}}, # 过滤条件，符合条件的文档才会被重新算分           "weight": 10 # 算分函数，算分函数的结果称为function score，将来会与query score运算                                   ，得到新算分，常见的算分函数有：                                             • weight：给一个常量值，作为函数结果（function score）                                             • field_value_factor：用文档中的某个字段值作为函数结果                                             • random_score：随机生成一个值，作为函数结果                                             • script_score：自定义计算公式，公式结果作为函数结果         }       ],       "boost_mode": "multiply" # 加权模式，定义function score与query score的运算方式，包括：                                                       • multiply：两者相乘。默认就是这个                                                       • replace：用function score替换query score                                                       •其它：sum、avg、max、min     }   } } 案例： GET /hotel/_search {   "query": {     "function_score": {       "query": {// ...},       "functions": [ // 算分函数         {           "filter": { // 满足的条件，品牌必须是如家             "term": {               "brand": "如家"             }           },           "weight": 2// 算分权重为2         }       ]       ,"boost_mode": "sum"     }   } }

Boolean Query

案例： 搜索名字包含“如家”，价格不高于400，在坐标31.21,121.5周围10km范围内的酒店。 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}           }         }       ]     }   } }

搜索结果处理

分页

elasticsearch默认情况下只返回top10的数据。而如果要查询更多数据 就需要修改分页参数了。 elasticsearch中通过修改from、size参数来控制要返回的分页结果： GET /hotel/_search {   "query": {     "match_all": {}   },   "from": 990, // 分页开始的位置，默认为0   "size": 10, // 期望获取的文档总数   "sort": [     {"price": "asc"}   ] }

深度分页问题

ES是分布式的，所以会面临深度分页问题。例如按price排序后，获取from = 990，size =10的数据： 1. 首先在每个数据分片上都排序并查询前1000条文档。 2. 然后将所有节点的结果聚合，在内存中重新排序选出前1000条文档 3. 最后从这1000条中，选取从990开始的10条文档如果搜索页数过深，     或者结果集（from +size）越大，对内存和CPU的消耗也越高。因此ES设定结果集查询的上限是10000 解决方案： 针对深度分页，ES提供了两种解决方案， 官方文档： • search after：分页时需要排序，原理是从上一次的排序值开始，查询下一页数据。官方推荐使用的方式。 • scroll：原理将排序数据形成快照，保存在内存。官方已经不推荐使用。 from + size：         • 优点：支持随机翻页         • 缺点：深度分页问题，默认查询上限（from + size）是10000         • 场景：百度、京东、谷歌、淘宝这样的随机翻页搜索 after search：         • 优点：没有查询上限（单次查询的size不超过10000）         • 缺点：只能向后逐页查询，不支持随机翻页         • 场景：没有随机翻页需求的搜索，例如手机向下滚动翻页 scroll：         • 优点：没有查询上限（单次查询的size不超过10000）         • 缺点：会有额外内存消耗，并且搜索结果是非实时的         • 场景：海量数据的获取和迁移。从ES7.1开始不推荐，建议用after search方案。

排序

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

高亮

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

搜索结果处理整体语法： GET /hotel/_search {   "query": {     "match": {       "name": "如家"     }   },   "from": 0, // 分页开始的位置   "size": 20, // 期望获取的文档总数   "sort": [      {  "price": "asc" }, // 普通排序     {       "_geo_distance" : { // 距离排序           "location" : "31.040699,121.618075",            "order" : "asc",           "unit" : "km"       }     }   ],   "highlight": {     "fields": { // 高亮字段       "name": {         "pre_tags": "",  // 用来标记高亮字段的前置标签         "post_tags": ""// 用来标记高亮字段的后置标签       }     }   } }

RestClient操作索引库

ES官方提供了各种不同语言的客户端，用来操作ES。 这些客户端的本质就是组装DSL语句，通过http请求发送给ES

RestClient查询文档

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

微服务保护

雪崩问题及解决方案

什么是服务雪崩

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

解决方案

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

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

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

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

初识Sentinel

服务保护技术对比

Sentinel介绍和安装

认识Sentinel

Sentinel是阿里巴巴开源的一款微服务流量控制组件。官网地址：https://sentinelguard.io/zh-cn/index.html Sentinel 具有以下特征: 丰富的应用场景：Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景，例如秒杀（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、集群流量控制、实时熔断下游不可用应用等。 完备的实时监控：Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据，甚至 500 台以下规模的集群的汇总运行情况。 广泛的开源生态：Sentinel 提供开箱即用的与其它开源框架/库的整合模块，例如与 Spring Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。 完善的 SPI 扩展点：Sentinel 提供简单易用、完善的 SPI 扩展接口。您可以通过实现扩展接口来快速地定制逻辑。例如定制规则管理、适配动态数据源等。

Sentinel安装

在GitHub下载jar包：sentinel-dashboard-1.8.1.jar 启动命令 java -jar sentinel-dashboard-1.8.1.jar -Dserver.port=8090

微服务整合Sentinel

我们在order-service中整合Sentinel，并且连接Sentinel的控制台，步骤如下： 1、引入sentinel依赖：   <dependency>        <groupId>com.alibaba.cloud</groupId>    <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId> </dependency> 2、配置控制台地址： spring:      cloud:           sentinel:              transport:                  dashboard: localhost:8080 3、访问微服务的任意端点，触发sentinel监控

限流规则

簇点链路

就是项目内的调用链路，链路中被监控的每个接口就是一个资源。 默认情况下sentinel会监控SpringMVC的每一个端点（Endpoint(就是controller，service，mapper)）， 因此SpringMVC的每一个端点（Endpoint）就是调用链路中的一个资源。 流控、熔断等都是针对簇点链路中的资源来设置的，因此我们可以点击对应资源后面的按钮来设置规则

使用方法

点击资源/order/{orderId}后面的流控按钮，就可以弹出表单。表单中可以添加流控规则。 其含义是限制 /order/{orderId}这个资源的单机QPS为1，即每秒只允许1次请求，超出的请求会被拦截并报错。

流控模式

直接：统计当前资源的请求，触发阈值时对当前资源直接限流，也是默认的模式

关联：统计与当前资源相关的另一个资源，触发阈值时，对当前资源限流

关联模式：统计与当前资源相关的另一个资源，触发阈值时，对当前资源限流 使用场景：比如用户支付时需要修改订单状态，同时用户要查询订单。                  查询和修改操作会争抢数据库锁，产生竞争。业务需求是有限支付和更新订单的业务，                  因此当修改订单业务触发阈值时，需要对查询订单业务限流(当修改订单的请求过多时                  ，对查询订单的请求进行拦截)。 当/write资源访问量触发阈值时，就会对/read资源限流，避免影响/write资源。

链路：统计从指定链路访问到本资源的请求，触发阈值时，对指定链路限流

链路模式：只针对从指定链路访问到本资源的请求做统计，判断是否超过阈值。 例如有两条请求链路： /test1  /common /test2  /common 如果只希望统计从/test2进入到/common的请求(当/test2链路来访问/common的请求达到阈值时拦截/test2请求)

需要注意： 1、Sentinel默认只标记Controller中的方法为资源，如果要标记其它方法，需要利用@SentinelResource注解，示例：      @SentinelResource("goods")      public void queryGoods() {              System.err.println("查询商品");       } 2、Sentinel默认会将Controller方法做context整合，导致链路模式的流控失效，需要修改application.yml，添加配置：       spring:            cloud:                sentinel:                    web-context-unify: false # 关闭context整合

流控效果

快速失败：达到阈值后，新的请求会被立即拒绝并抛出FlowException异常。是默认的处理方式。

warm up：预热模式，对超出阈值的请求同样是拒绝并抛出异常。                  但这种模式阈值会动态变化，从一个较小值逐渐增加到最大阈值。

warm up也叫预热模式，是应对服务冷启动的一种方案。请求阈值初始值是 threshold / coldFactor， 持续指定时长后，逐渐提高到threshold值。而coldFactor的默认值是3. 例如，我设置QPS的threshold为10，预热时间为5秒，那么初始阈值就是 10 / 3 ，也就是3，然后在5秒后逐渐增长到10.

排队等待：让所有的请求按照先后次序排队执行，两个请求的间隔不能小于指定时长

当请求超过QPS阈值时，快速失败和warm up 会拒绝新的请求并抛出异常。而排队等待则是让所有请求进入一个队列中 ，然后按照阈值允许的时间间隔依次执行。后来的请求必须等待前面执行完成，如果请求预期的等待时间超出最大时长，则会被拒绝。 例如：QPS = 5，意味着每200ms处理一个队列中的请求；timeout = 2000，意味着预期等待超过2000ms的请求会被拒绝并抛出异常

热点参数限流

热点参数限流对默认的SpringMVC资源无效 热点参数是，给需要限流的请求某一参数的取值范围进行限流，凡是在这个范围内的进行限流

隔离和降级

熔断降级的作用

虽然限流可以尽量避免因高并发而引起的服务故障，但服务还会因为其它原因而故障。 而要将这些故障控制在一定范围，避免雪崩，就要靠线程隔离（舱壁模式）和熔断降级手段了。 不管是线程隔离还是熔断降级，都是对客户端（调用方）的保护。

降级

Feign整合Sentinel 实现服务的降级

如何实现

SpringCloud中，微服务调用都是通过Feign来实现的，因此做客户端保护必须整合Feign和Sentinel。 1、修改OrderService的application.yml文件，开启Feign的Sentinel功能 feign:      sentinel:          enabled: true # 开启Feign的Sentinel功能 2、给FeignClient编写失败后的降级逻辑 方式一：FallbackClass，无法对远程调用的异常做处理 方式二：FallbackFactory，可以对远程调用的异常做处理，我们选择这种

实现步骤

步骤一：在feing-api项目中定义类，实现FallbackFactory：            @Slf4j         public class UserClientFallbackFactory implements FallbackFactory<UserClient> {                    @Override                    public UserClient create(Throwable throwable) {                      // 创建UserClient接口实现类，实现其中的方法，编写失败降级的处理逻辑                             return new UserClient() {                                       @Override                                       public User findById(Long id) {                                              // 记录异常信息                                              log.error("查询用户失败", throwable);                                              // 根据业务需求返回默认的数据，这里是空用户                                              return new User();                                      }                           };                  }         }      步骤二：在feing-api项目中的DefaultFeignConfiguration类中将UserClientFallbackFactory注册为一个Bean：      @Bean       public UserClientFallbackFactory userClientFallback(){               return new UserClientFallbackFactory();        } 步骤三：在feing-api项目中的UserClient接口中使用UserClientFallbackFactory：       @FeignClient(value = "userservice", fallbackFactory = UserClientFallbackFactory.class)         public interface UserClient {                   @GetMapping("/user/{id}")                   User findById(@PathVariable("id") Long id);         }

隔离

线程池隔离

特点：基于线程池模式，有额外开销，但隔离控制更强

信号量隔离（Sentinel默认采用）

特点：基于计数器模式，简单，开销小

线程隔离（舱壁模式） QPS：就是每秒的请求数 线程数：是该资源能使用用的tomcat线程数的最大值。也就是通过限制线程数量，实现舱壁模式。

熔断降级

异常比例或异常数

统计指定时间内的调用，如果调用次数超过指定请求数，并且出现异常的比例达到设定的比例阈值（或超过指定异常数），则触发熔断。 例如：统计最近1000ms内的请求，如果请求量超过10次，并且异常比例不低于0.5，则触发熔断，熔断时长为5秒。 然后进入half-open状态，放行一次请求做测试。

慢调用

业务的响应时长（RT）大于指定时长的请求认定为慢调用请求。在指定时间内，如果请求数量超过设定的最小数量，慢调用比例大于设定的阈值，则触发熔断 举例：设置RT超过500ms的调用是慢调用，统计最近10000ms内的请求，如果请求量超过10次，并且慢调用比例不低于0.5，则触发熔断，熔断时长为5秒。然后进入half-open状态，放行一次请求做测试。

什么是熔断降级

熔断降级是解决雪崩问题的重要手段。其思路是由断路器统计服务调用的异常比例、慢请求比例，如果超出阈值则会熔断该服务。即拦截访问该服务的一切请求；而当服务恢复时，断路器会放行访问该服务的请求。

服务授权

什么是服务授权

授权规则可以对调用方的来源做控制，有白名单和黑名单两种方式。 白名单：来源（origin）在白名单内的调用者允许访问 黑名单：来源（origin）在黑名单内的调用者不允许访问 例如，我们限定只允许从网关来的请求访问order-service，那么流控应用中就填写网关的名称

如何实现

案例： Sentinel是通过RequestOriginParser这个接口的parseOrigin来获取请求的来源的。 public interface RequestOriginParser {            /**             * 从请求request对象中获取origin，获取方式自定义      */            String parseOrigin(HttpServletRequest request);  } 1、所以首先我们要在被请求的服务里创建一个请求头，我们尝试从request中获取一个名为origin的请求头，作为origin的值： @Component  public class HeaderOriginParser implements RequestOriginParser {          @Override          public String parseOrigin(HttpServletRequest request) {                  String origin = request.getHeader("origin");      //这个请求头参数是自定义的，要和gateway添加的请求头参数对应上，只有这样在添加黑白名单的时候才能对请求进行识别，是放行还是拒绝            if(StringUtils.isEmpty(origin)){                          return "blank";                  }                  return origin;          }  } 2、我们还需要在gateway服务中，利用网关的过滤器添加名为gateway的origin头： spring:      cloud:          gateway:              default-filters:                  - AddRequestHeader=origin,gateway # 添加名为origin的请求头，值为gateway 给/order/{orderId} 配置授权规则： 在sentinel控制台配置，资源名：/order/{orderId},流控应用：gateway，授权类型：白名单

自定义异常结果

默认情况下，发生限流、降级、授权拦截时，都会抛出异常到调用方。如果要自定义异常时的返回结果，需要实现BlockExceptionHandler接口： public interface BlockExceptionHandler {           /**             * 处理请求被限流、降级、授权拦截时抛出的异常：BlockException             */         void handle(HttpServletRequest request, HttpServletResponse response, BlockException e) throws Exception;   } 而BlockException包含很多个子类，分别对应不同的场景： FlowException  限流异常 ParamFlowException  热点参数异常 DegradeException  降级异常 AuthorityException  授权规则异常 SystemBlockException  系统规则异常

规则持久化

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

实现push模式

push模式实现最为复杂，依赖于nacos，并且需要改在Sentinel控制台。整体步骤如下： ## 一、修改order-service服务 修改OrderService，让其监听Nacos中的sentinel规则配置。 具体步骤如下： ### 1.引入依赖 在order-service中引入sentinel监听nacos的依赖： ```xml <dependency>     <groupId>com.alibaba.csp</groupId>     <artifactId>sentinel-datasource-nacos</artifactId> </dependency> ``` ### 2.配置nacos地址 在order-service中的application.yml文件配置nacos地址及监听的配置信息： datasource:         flow:  # 限流规则           nacos:             server-addr: localhost:8848 # 监听nacos地址             dataId: orderservice-flow-rules             groupId: SENTINEL_GROUP             rule-type: flow         degrade: # 降级规则           nacos:             server-addr: localhost:8848 # 监听nacos地址             dataId: orderservice-degrade-rules             groupId: SENTINEL_GROUP             rule-type: degrade ## 二、修改sentinel-dashboard源码 SentinelDashboard默认不支持nacos的持久化，需要修改源码。 ### 1. 解压 解压sentinel源码包，然后并用IDEA打开这个项目。 ### 2. 修改nacos依赖 在sentinel-dashboard源码的pom文件中，nacos的依赖默认的scope是test，只能在测试时使用，这里要去除 将sentinel-datasource-nacos依赖的scope去掉： <dependency>     <groupId>com.alibaba.csp</groupId>     <artifactId>sentinel-datasource-nacos</artifactId> </dependency> ### 3. 添加nacos支持 在sentinel-dashboard的test包下，已经编写了对nacos的支持，我们需要将其拷贝到main下。 ### 4. 修改刚刚拷贝的nacos包下的NacosConfig类，修改其中的nacos地址 @Bean  public ConfigService nacosConfigService() throws Exception {          return ConfigFactory.createConfigService("localhost:8848");   } 在sentinel-dashboard的application.properties中添加nacos地址配置： ```properties nacos.addr=localhost:8848 ``` ### 5. 配置nacos数据源 另外，还需要修改com.alibaba.csp.sentinel.dashboard.controller.v2包下的FlowControllerV2类：     @Autowired     @Qualifier("flowRuleNacosProvider")     private DynamicRuleProvider<List<FlowRuleEntity>> ruleProvider;     @Autowired     @Qualifier("flowRuleNacosPublisher")     private DynamicRulePublisher<List<FlowRuleEntity>> rulePublisher; 让我们添加的Nacos数据源生效： ### 6. 修改前端页面 接下来，还要修改前端页面，添加一个支持nacos的菜单。 修改src/main/webapp/resources/app/scripts/directives/sidebar/目录下的sidebar.html文件： 将其中的这部分注释打开：  修改其中的文本： <li ui-sref-active="active" ng-if="entry.appType==0">             <a ui-sref="dashboard.flow({app: entry.app})">               <i class="glyphicon glyphicon-filter">&nbsp;&nbsp;流控规则-NACOS</a>           </li> ### 7. 重新编译、打包项目 运行IDEA中的maven插件，编译和打包修改好的Sentinel-Dashboard： ![image-20210618202701492](assets/image-20210618202701492.png) ### 8.启动 启动方式跟官方一样： ```sh java -jar sentinel-dashboard.jar ``` 如果要修改nacos地址，需要添加参数： ```sh java -jar -Dnacos.addr=localhost:8848 sentinel-dashboard.jar ```

多级缓存

多级缓存分层

Nginx缓存

Redis缓存

Canal数据同步

分布式事务

事务的ACID原则

原子性

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

一致性

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

隔离性

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

持久性

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

分布式事务的理论基础

CAP定理

一致性（Consistence）

用户访问分布式系统任意节点，得到的数据必须一致

可用性（Availability）

用户在访问分布式系统任意健康节点，必须得到响应，而不是超时或拒绝

分区容错性（partition tolerance ）

Partition（分区）：因为网络故障或其它原因导致分布式系统中的部分节点与其它节点失去连接，形成独立分区。 Tolerance（容错）：在集群出现分区时，整个系统也要持续对外提供服务

BASE理论

BASE理论是对CAP的一种解决思路，包含三个思想： Basically Available （基本可用）：分布式系统在出现故障时，允许损失部分可用性，即保证核心可用。 Soft State（软状态）：在一定时间内，允许出现中间状态，比如临时的不一致状态。 Eventually Consistent（最终一致性）：虽然无法保证强一致性，但是在软状态结束后，最终达到数据一致。 而分布式事务最大的问题是各个子事务的一致性问题，因此可以借鉴CAP定理和BASE理论： AP模式：各子事务分别执行和提交，允许出现结果不一致，然后采用弥补措施恢复数据即可，实现最终一致。 CP模式：各个子事务执行后互相等待，同时提交，同时回滚，达成强一致。但事务等待过程中，处于弱可用状态。

分布式事务模型

初识Seata

Seata框架

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

Seata四种分布式事务解决方案

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

Seata的部署和集成

Seata的部署

1、下载Seata

下载地址：http://seata.io/en-us/blog/download.html

2、解压

3、修改配置

修改conf目录下的registry.conf文件： registry {   # tc服务的注册中心类，这里选择nacos，也可以是eureka、zookeeper等   type = "nacos"   nacos {     # seata tc 服务注册到 nacos的服务名称，可以自定义     application = "seata-tc-server"     serverAddr = "127.0.0.1:8848"     group = "DEFAULT_GROUP"     namespace = ""     cluster = "SH"     username = "nacos"     password = "nacos"   } } config {   # 读取tc服务端的配置文件的方式，这里是从nacos配置中心读取，这样如果tc是集群，可以共享配置   type = "nacos"   # 配置nacos地址等信息   nacos {     serverAddr = "127.0.0.1:8848"     namespace = ""     group = "SEATA_GROUP"     username = "nacos"     password = "nacos"     dataId = "seataServer.properties"   } }

4.在nacos添加配置

特别注意，为了让tc服务的集群可以共享配置，我们选择了nacos作为统一配置中心。因此服务端配置文件seataServer.properties文件需要在nacos中配好。 格式如下： 配置内容如下： ```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=123 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 ``` ==其中的数据库地址、用户名、密码都需要修改成你自己的数据库信息。==

5.创建数据库表

特别注意：tc服务在管理分布式事务时，需要记录事务相关数据到数据库中，你需要提前创建好这些表。 新建一个名为seata的数据库，运行seata官方提供的sql文件： 这些表主要记录全局事务、分支事务、全局锁信息： ```mysql SET NAMES utf8mb4; SET FOREIGN_KEY_CHECKS = 0; -- ---------------------------- -- 分支事务表 -- ---------------------------- DROP TABLE IF EXISTS `branch_table`; CREATE TABLE `branch_table`  (   `branch_id` bigint(20) NOT NULL,   `xid` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,   `transaction_id` bigint(20) NULL DEFAULT NULL,   `resource_group_id` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `resource_id` varchar(256) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `branch_type` varchar(8) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `status` tinyint(4) NULL DEFAULT NULL,   `client_id` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `application_data` varchar(2000) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `gmt_create` datetime(6) NULL DEFAULT NULL,   `gmt_modified` datetime(6) NULL DEFAULT NULL,   PRIMARY KEY (`branch_id`) USING BTREE,   INDEX `idx_xid`(`xid`) USING BTREE ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact; -- ---------------------------- -- 全局事务表 -- ---------------------------- DROP TABLE IF EXISTS `global_table`; CREATE TABLE `global_table`  (   `xid` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,   `transaction_id` bigint(20) NULL DEFAULT NULL,   `status` tinyint(4) NOT NULL,   `application_id` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `transaction_service_group` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `transaction_name` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `timeout` int(11) NULL DEFAULT NULL,   `begin_time` bigint(20) NULL DEFAULT NULL,   `application_data` varchar(2000) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,   `gmt_create` datetime NULL DEFAULT NULL,   `gmt_modified` datetime NULL DEFAULT NULL,   PRIMARY KEY (`xid`) USING BTREE,   INDEX `idx_gmt_modified_status`(`gmt_modified`, `status`) USING BTREE,   INDEX `idx_transaction_id`(`transaction_id`) USING BTREE ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact; SET FOREIGN_KEY_CHECKS = 1; ```

6.启动TC服务

进入bin目录，运行其中的seata-server.bat即可： 启动成功后，seata-server应该已经注册到nacos注册中心了。

Seata集成微服务

1、首先，引入seata相关依赖： <dependency>          <groupId>com.alibaba.cloud</groupId>          <artifactId>spring-cloud-starter-alibaba-seata</artifactId>          <exclusions>                                 <exclusion>                          <artifactId>seata-spring-boot-starter</artifactId>                          <groupId>io.seata</groupId>                  </exclusion>          </exclusions>  </dependency>    <dependency>          <groupId>io.seata</groupId>          <artifactId>seata-spring-boot-starter</artifactId>          <version>${seata.version}</version>  </dependency>

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

3、启动微服务

AT模式

AT模式同样是分阶段提交的事务模型，不过缺弥补了XA模型中资源锁定周期过长的缺陷。 阶段一RM的工作：   注册分支事务   记录undo-log（数据快照）   执行业务sql并提交报告事务状态 阶段二提交时RM的工作：   删除undo-log即可 阶段二回滚时RM的工作：   根据undo-log恢复数据到更新前

TCC模式

TCC的空回滚和业务悬挂

当某分支事务的try阶段阻塞时，可能导致全局事务超时而触发二阶段的cancel操作。 在未执行try操作时先执行了cancel操作，这时cancel不能做回滚，就是空回滚。 对于已经空回滚的业务，如果以后继续执行try，就永远不可能confirm或cancel， 这就是业务悬挂。应当阻止执行空回滚后的try操作，避免悬挂

Saga模式

XA模式

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

可靠消息服务

镜像集群

延迟队列

仲裁队列

惰性队列

消息三方确认

分布式缓存

数据持久化

RDB持久化

1、Redis内部有触发RDB的机制，可以在redis.conf文件中找到，格式如下： # 900秒内，如果至少有1个key被修改，则执行bgsave ， 如果是save "" 则表示禁用RDB save 900 1   save 300 10   save 60 10000  #代表60秒内至少执行1000次修改则触发RDB 2、RDB的其它配置也可以在redis.conf文件中设置： # 是否压缩 ,建议不开启，压缩也会消耗cpu，磁盘的话不值钱 rdbcompression yes # RDB文件名称 dbfilename dump.rdb   # 文件保存的路径目录 dir ./  3、RDB的缺点？ RDB执行间隔时间长，两次RDB之间写入数据有丢失的风险 fork子进程、压缩、写出RDB文件都比较耗时

AOF持久化

1、AOF默认是关闭的，需要修改redis.conf配置文件来开启AOF： # 是否开启AOF功能，默认是no appendonly yes # AOF文件的名称 appendfilename "appendonly.aof" 2、AOF的命令记录的频率也可以通过redis.conf文件来配： # 表示每执行一次写命令，立即记录到AOF文件 appendfsync always  # 写命令执行完先放入AOF缓冲区，然后表示每隔1秒将缓冲区数据写到AOF文件，是默认方案 appendfsync everysec  # 写命令执行完先放入AOF缓冲区，由操作系统决定何时将缓冲区内容写回磁盘 appendfsync no 3、因为是记录命令，AOF文件会比RDB文件大的多。而且AOF会记录对同一个key的多次写操作，但只有最后一次写操作才有意义。通过执行bgrewriteaof命令，可以让AOF文件执行重写功能，用最少的命令达到相同效果。 Redis也会在触发阈值时自动去重写AOF文件。阈值也可以在redis.conf中配置： # AOF文件比上次文件 增长超过多少百分比则触发重写  auto-aof-rewrite-percentage 100  # AOF文件体积最小多大以上才触发重写   auto-aof-rewrite-min-size 64mb

AOF与RDB各自适用场景

RDB 的持久化方式是对整个内存做快照，通过设置Save时间，来确定是多久更新一次，文件体积小 这会导致数据可能不完整，但宕机时恢复速度快，他在执行写操作时会占用CPU资源和内存资源， 适用于可容忍部分数据丢失，追求更快的启动速度的场景 AOF 是记录每次的执行命令，数据相对完整，文件体积大，但宕机时恢复速度慢，因为要一条一条的执行命令。 数据完整性高，系统资源占用低，但在重写时，会占用大量的CPU和内存 适用于对数据安全性较高的场景

redis搭建

单机搭建Redis

一、下载并解压Redis 1、执行下面的命令下载redis： wget https://download.redis.io/releases/redis-6.2.6.tar.gz 2、解压redis： tar xzf redis-6.2.6.tar.gz 3、移动redis目录，一般都会将redis目录放置到 /usr/local/redis目录： mv redis-6.2.6 /usr/local/redis 二、编译并安装redis 1、进入redis安装目录，执行make命令编译redis： cd /usr/local/redis make 等待make命令执行完成即可。 如果执行make命令报错：cc 未找到命令，原因是虚拟机系统中缺少gcc，执行下面命令安装gcc： yum -y install gcc automake autoconf libtool make 如果执行make命令报错：致命错误:jemalloc/jemalloc.h: 没有那个文件或目录，则需要在make指定分配器为libc。 执行下面命令即可正常编译： make MALLOC=libc make命令执行完，redis就编译完成了。 2、执行下面命令安装redis，并指定安装目录 make install PREFIX=/usr/local/redis 至此，redis即安装成功。 三、启动redis 1、进入redis安装目录，执行下面命令启动redis服务 ./bin/redis-server redis.conf

Redis主从搭建

2.Redis主从集群 ## 2.1.集群结构 我们搭建的主从集群结构，共包含三个节点，一个主节点，两个从节点。 这里我们会在同一台虚拟机中开启3个redis实例，模拟主从集群，信息如下： |       IP        | PORT |  角色  | | :-------------: | :--: | :----: | | 192.168.150.101 | 7001 | master | | 192.168.150.101 | 7002 | slave  | | 192.168.150.101 | 7003 | slave  | ## 2.2.准备实例和配置 要在同一台虚拟机开启3个实例，必须准备三份不同的配置文件和目录，配置文件所在目录也就是工作目录。 1）创建目录 我们创建二个文件夹，名字分别叫6380、6381： # 进入/tmp目录 cd /tmp # 创建目录 mkdir 6380 6381 2）配置从节点redis.config port 6380 #从节点端口号 pidfile /var/run/redis_6380.pid  # 把pid进程号写入pidfile配置的文件 logfile "6380.log" dir /usr/local/redis-5.0.3/data/6380  # 指定数据存放目录 # 需要注释掉bind # bind 127.0.0.1（bind绑定的是自己机器网卡的ip，如果有多块网卡可以配多个ip， 代表允许客户端通过机器的哪些网卡ip去访问，内网一般可以不配置bind，注释掉即可） replicaof 192.168.0.60 6379   # 配置主从复制，从本机6379的redis实例复制数据，Redis 5.0之前使用slaveof replica-read-only yes  # 配置从节点只读 3）拷贝配置文件到新创建的目录下6380 cp ./redis.conf  /tmp/6380 4）重复2，3步再配置一个6381从节点 2.3.启动 为了方便查看日志，我们打开3个ssh窗口，分别启动3个redis实例，启动命令： 进入redis的bin目录，输入以下命令，启动三个节点 redis-server redis.conf    （主） redis-server /tmp/6380/redis.conf   （从） redis-server /tmp/6381/redis.conf   （从） 启动后： 如果要一键停止，可以运行下面命令： printf '%s\n' 7001 7002 7003 | xargs -I{} -t redis-cli -p {} shutdown ## 2.5.测试 执行下列操作以测试： - 利用redis-cli连接6379，执行```set num 123``` - 利用redis-cli连接6380，执行```get num```，再执行```set num 666``` - 利用redis-cli连接6381，执行```get num```，再执行```set num 888``` 可以发现，只有在6379这个master节点上可以执行写操作，7002和7003这两个slave节点只能执行读操作。

Redis哨兵集群搭建

# 3.搭建哨兵集群 ## 3.1.集群结构 这里我们搭建一个三节点形成的Sentinel集群，来监管之前的Redis主从集群。 三个sentinel实例信息如下： | 节点 |       IP        | PORT  | | ---- | :-------------: | :---: | | s1   | 192.168.150.101 | 26379 | | s2   | 192.168.150.101 | 26380 | | s3   | 192.168.150.101 | 26381 | ## 3.2.准备实例和配置 要在同一台虚拟机开启3个实例，必须准备三份不同的配置文件和目录，配置文件所在目录也就是工作目录。 我们创建三个文件夹，名字分别叫s1、s2、s3： ```sh # 进入/tmp目录 cd /tmp # 创建目录 mkdir s1 s2 s3 ``` 然后我们在s1目录创建一个sentinel.conf文件，添加下面的内容： #是否开启守护线程 daemonize no # 不需要创建该文件，文件会自动生成，这个文件是在开启守护线程时生成的 pidfile "/var/run/redis-sentinel-26379.pid"  # 不需要创建该文件，文件会自动生成（注意，不填写时，日志会在终端直接输出，填写后日志输出到文件中） logfile "26379.log"  # 当前sentinel实例的端口号 port 26379   # 指定主节点信息  - `mymaster`：主节点名称 `2`：选举master时的quorum值 sentinel monitor mymaster 192.168.177.128 6379 2 # 主节点宕机后的等待时间 sentinel down-after-milliseconds mymaster 5000 sentinel failover-timeout mymaster 60000 dir "/tmp/s1" 然后将s1/sentinel.conf文件拷贝到s2、s3两个目录中（在/tmp目录执行下列命令）： # 方式一：逐个拷贝 cp s1/sentinel.conf s2 cp s1/sentinel.conf s3 # 方式二：管道组合命令，一键拷贝 echo s2 s3 | xargs -t -n 1 cp s1/sentinel.conf 修改s2、s3两个文件夹内的配置文件，将端口分别修改为27002、27003： sed -i -e 's/27001/27002/g' -e 's/s1/s2/g' s2/sentinel.conf sed -i -e 's/27001/27003/g' -e 's/s1/s3/g' s3/sentinel.conf ## 3.3.启动 为了方便查看日志，我们打开3个ssh窗口，分别启动3个redis实例，启动命令： # 第1个 redis-sentinel s1/sentinel.conf # 第2个 redis-sentinel s2/sentinel.conf # 第3个 redis-sentinel s3/sentinel.conf ## 3.4.测试 尝试让master节点7001宕机， 查看sentinel日志，查看7003的日志，查看7002的日志

Redis分片集群搭建

4.1.集群结构 分片集群需要的节点数量较多，这里我们搭建一个最小的分片集群，包含3个master节点，每个master包含一个slave节点 这里我们会在同一台虚拟机中开启6个redis实例，模拟分片集群，信息如下： |       IP        | PORT |  角色  | | :-------------: | :--: | :----: | | 192.168.150.101 | 7001 | master | | 192.168.150.101 | 7002 | master | | 192.168.150.101 | 7003 | master | | 192.168.150.101 | 8001 | slave  | | 192.168.150.101 | 8002 | slave  | | 192.168.150.101 | 8003 | slave  | ## 4.2.准备实例和配置 删除之前的7001、7002、7003这几个目录，重新创建出7001、7002、7003、8001、8002、8003目录： # 进入/tmp目录 cd /tmp # 创建目录 mkdir 7001 7002 7003 8001 8002 8003 在/tmp下准备一个新的redis.conf文件，内容如下： port 6379 # 开启集群功能 cluster-enabled yes # 集群的配置文件名称，不需要我们创建，由redis自己维护 cluster-config-file /tmp/6379/nodes.conf # 节点心跳失败的超时时间 cluster-node-timeout 5000 # 持久化文件存放目录 dir /tmp/6379 # 绑定地址 bind 0.0.0.0 # 让redis后台运行（守护进程） daemonize yes # 注册的实例ip replica-announce-ip 192.168.177.128 # 保护模式（不需要做什么用户名密码的校验了） protected-mode no # 数据库数量 databases 1 # 日志 logfile /tmp/6379/run.log 将这个文件拷贝到每个目录下： # 进入/tmp目录 cd /tmp # 执行拷贝 echo 7001 7002 7003 8001 8002 8003 | xargs -t -n 1 cp redis.conf 修改每个目录下的redis.conf，将其中的6379修改为与所在目录一致： # 进入/tmp目录 cd /tmp # 修改配置文件 printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t sed -i 's/6379/{}/g' {}/redis.conf ## 4.3.启动 因为已经配置了后台启动模式，所以可以直接启动服务： # 进入/redis目录 cd /usr/local/redis/ # 一键启动所有服务 printf  '%s\n'  /tmp/7001 /tmp/7002 /tmp/7003 /tmp/8001 /tmp/8002 /tmp/8003 | xargs -I{} -t ./bin/redis-server {}/redis.conf 通过ps查看状态： ps -ef | grep redis 发现服务都已经正常启动 如果要关闭所有进程，可以执行命令： ps -ef | grep redis | awk '{print $2}' | xargs kill 或者（推荐这种方式）： printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t redis-cli -p {} shutdown ## 4.4.创建集群 虽然服务启动了，但是目前每个服务之间都是独立的，没有任何关联。 我们需要执行命令来创建集群，在Redis5.0之前创建集群比较麻烦，5.0之后集群管理命令都集成到了redis-cli中。 1）Redis5.0之前 Redis5.0之前集群命令都是用redis安装包下的src/redis-trib.rb来实现的。因为redis-trib.rb是有ruby语言编写的所以需要安装ruby环境。  # 安装依赖  yum -y install zlib ruby rubygems  gem install redis 然后通过命令来管理集群： # 进入redis的src目录 cd /tmp/redis-6.2.4/src # 创建集群 ./redis-trib.rb create --replicas 1 192.168.150.101:7001 192.168.150.101:7002 192.168.150.101:7003 192.168.150.101:8001 192.168.150.101:8002 192.168.150.101:8003 2）Redis5.0以后 我们使用的是Redis6.2.4版本，集群管理以及集成到了redis-cli中，格式如下： redis-cli --cluster create --cluster-replicas 1 192.168.150.101:7001 192.168.150.101:7002 192.168.150.101:7003 192.168.150.101:8001 192.168.150.101:8002 192.168.150.101:8003 命令说明： - `redis-cli --cluster`或者`./redis-trib.rb`：代表集群操作命令 - `create`：代表是创建集群 - `--replicas 1`或者`--cluster-replicas 1` ：指定集群中每个master的副本个数为1，此时`节点总数 ÷ (replicas + 1)` 得到的就是master的数量。因此节点列表中的前n个就是master，其它节点都是slave节点，随机分配到不同master 然后会问，是否同意这么分配节点，输入yes，则集群开始创建 通过命令可以查看集群状态： redis-cli -p 7001 cluster nodes ## 4.5.测试 尝试连接7001节点，存储一个数据： # 连接 redis-cli -c -p 7001 # 存储数据 set num 123 # 读取数据 get num # 再次存储 set a 1

Redis主从集群

数据同步原理

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

数据同步流程

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

全量同步和增量同步

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

集群优化

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

哨兵机制

哨兵的作用

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

服务状态监控

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

选举新的master

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

如何实现故障转移

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

RedisTemplate的哨兵模式

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

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

Redis分片集群

分片集群结构

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

散列插槽

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

集群伸缩

添加节点到集群

1、创建新节点7004，并启动       （步骤同分片集群搭建） 2、将新节点连接到集群成为主节点(不添加 cluster-slave 则默认为主节点)       ##第一个为新增节点，第二个为集群中已知存在节点（任意一个）       ./bin/redis-cli add-node 192.168.177.128:7004 192.168.177.128:7001  3、为新节点7004分配slot槽       ./bin/redis-cli --cluster reshare 192.168.177.128:7001       然后会问你分配多少。输入3000，回车       再问你分配给谁，输入新节点7004的ID，回车 4、添加从节点，创建新节点8004，并启动    （步骤同分片集群搭建） 5、将新节点连接到集群       ./bin/redis-cli add-node 192.168.177.128:8004 192.168.177.128:7001  6、用客户端连接8004节点，执行命令，将其变成从节点       ./bin/redis-cli -c -h 192.168.177.128 -p 8004       192.168.177.128:8004> cluster replicate bdb0ce9cc5cca545b9182a09346246827af40490

从集群中删除节点

1、将要删除的主节点插槽分配给其他节点     （步骤同添加时分配插槽，只不过角色互换一下） 2、从集群中删除节点       ./bin/redis-cli del-node 192.168.177.128:7004 6101db56f13af2252e356eb08977b6bd2bb3e2f9 3、删除从节点      直接删除即可

故障转移

数据迁移

具体实现原理

1.slave节点告诉master节点拒绝任何客户端请求. 2.&nbsp;master返回当前的数据offset给slave 3.等待数据offset与master一致 4.开始故障转移（原master变slave,指定的slave变master） 5.标记自己为master，广播故障转移的结果 6.收到广播，开始处理客户端读请求

步骤

利用cluster failover 命令可以手动让集群中的某个master宕机，切换到执行cluster failover命令的这个slave节点，实现无感知的数据迁移。 1、连接需要变成master 的节点上  ./bin/redis-cli -c -p 192.168.177.128:7002 2、192.168.177.128:7002> 输入 cluster failover 回车

RedisTemplate的分片集群模式

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