微服务技术栈
认识微服务
单体架构
单体架构:将业务的所有功能集中在一个项目中开发,打成一个包部署
优点:
- 架构简单
- 部署成本低
缺点:
- 耦合度高
分布式架构
分布式架构:根据业务功能对系统进行拆分,每个业务模块作为独立项目开发,称为一个服务
优点:
- 降低服务耦合
- 有利于服务升级拓展
需要考虑的问题:
- 服务拆分粒度如何?
- 服务集群地址如何维护?
- 服务之间如何实现远程调用
- 服务健康状态如何感知?
微服务
微服务是一种经过良好架构设计的分布式架构方案,微服务架构特征:
- 单一职责:微服务拆分粒度更小,每一个服务都对应唯一的业务能力,做到单一职责,避免重复业务开发
- 面向服务:微服务对外暴露业务接口
- 自治:团队独立、技术独立、数据独立、部署独立
- 隔离性强:服务调用做好隔离、容错、降级,避免出现级联问题
微服务架构
微服务这种方案需要技术框架落地,全球的互联网公司都在积极尝试自己的微服务落地技术。在国内最知名的就是SpringCloud和阿里巴巴的Dubbo
Dubbo | SpringCloud | SpringCloudAlibaba | |
---|---|---|---|
注册中心 | zookeeper、Redis | Eureka、Consul | Nacos、Eureka |
服务远程调用 | Dubbo协议 | Feign(HTTP协议) | Dubbo、Feign |
配置中心 | 无 | SpringCloudCOnfig | Spring Cloud Config、Nacos |
服务网关 | 无 | SpringCloudGateway、Zuul | SpringCloudGayeway、Zuul |
服务监控和保护 | dubbo-admin,功能弱 | Hystrix | Sentinel |
SpringCloud
SpringCloud是目前国内使用最广泛的微服务框架。官网地址:Spring Cloud
SpringCloud集成了各种微服务功能组件,并基于SpringBoot实现了这些组件的自动装配,从而提供了良好的开箱即用体验
服务拆分及远程调用
服务拆分注意事项
- 不同微服务,不要重复开发相同业务
- 微服务数据独立,不要访问其它微服务的数据库
- 微服务可以将自己的业务暴露为接口,供其它微服务调用
实现远程调用案例
在order-service服务中,创建一个根据id查询订单的接口:
import com.dc.order.pojo.Order;
import com.dc.order.service.OrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("order")
public class OrderController {@Autowiredprivate OrderService orderService;@GetMapping("{orderId}")public Order queryOrderByUserId(@PathVariable("orderId") Long orderId) {// 根据id查询订单并返回return orderService.queryOrderById(orderId);}
}
根据id查询订单,返回值是order对象
其中的user为null
在user-service中有一个根据id查询用户的接口:
import com.dc.user.pojo.User;
import com.dc.user.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;@Slf4j
@RestController
@RequestMapping("/user")
public class UserController {@Autowiredprivate UserService userService;/*** 路径: /user/1** @param id 用户id* @return 用户*/@GetMapping("/{id}")public User queryById(@PathVariable("id") Long id) {return userService.queryById(id);}
}
查询结果如图:
需求
修改order-service中的根据id查询订单业务,要求在查询订单的同时,根据订单中包含的userId查询出用户信息,一起返回。
因此,需要在order-service中,向user-service发起一个http请求,调用http://localhost:80/user/{userId}这个接口
步骤:
- 注册一个RestTemplate的实例到Spring容器中
- 修改order-service服务中的OrderService类中的queryOrderById方法,根据Order对象中的userId查询user
- 将查询到的User填充到Order对象,一起返回
注册RestTemplate
首先,需要在order-service服务中的OrderApplication启动类中,注册RestTemplate实例:
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.web.client.RestTemplate;@MapperScan("com.dc.order.mapper")
@SpringBootApplication
public class OrderApplication {public static void main(String[] args) {SpringApplication.run(OrderApplication.class, args);}@Beanpublic RestTemplate restTemplate() {return new RestTemplate();}
}
实现远程调用
修改order-service服务中的OrderService类中的queryOrderById方法
import com.dc.order.mapper.OrderMapper;
import com.dc.order.pojo.Order;
import com.dc.order.pojo.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;@Service
public class OrderService {@Autowiredprivate OrderMapper orderMapper;@Autowiredprivate RestTemplate restTemplate;public Order queryOrderById(Long orderId) {// 1.查询订单Order order = orderMapper.findById(orderId);// 远程查询User// url地址String url = "http://localhost:80/user/" + order.getUserId();// 发起调用User user = restTemplate.getForObject(url, User.class);order.setUser(user);// 4.返回return order;}
}
结果如图:
提供者与消费者
在服务调用关系中,会有两个不同的角色:
服务提供者:一次业务中,被其它微服务调用的服务。(提供接口给其他微服务)
服务消费者:一次业务中,调用其他微服务的服务。(调用其它微服务提供的接口)
但是,服务提供者与服务消费者的角色并不是绝对的,而是相对于业务而言。
-
对于A调用B的业务而言:A是服务消费者,B是服务提供者
-
对于B调用C的业务而言:B是服务消费者,C是服务提供者
因此,服务B既可以是服务提供者,也可以是服务消费者
Eureka注册中心
eureka的作用:
- 消费者该如何获取服务提供者具体信息?
- 服务提供者启动时向eureka注册自己的信息
- eureka保存这些信息
- 消费者提供服务名称向eureka拉取提供者信息
- 如果有多个服务提供者,消费者该如何选择?
- 服务消费者利用负载均衡算法,从服务列表中挑选一个
- 消费者如何感知服务提供者健康状态?
- 服务提供者会每隔30秒向EurekaServer发送心跳请求,报告健康状态
- eureka会更新记录到服务列表信息,心跳不正常会被剔除
- 消费者就可以拉取到zui’xin
搭建eureka-server
首先需要在cloud-demo父工程下,创建一个子模块eureka-server
引入依赖
引入SpringCloud为eureka提供的starter依赖
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-server</artifactId>
</dependency>
编写启动类
给eureka-server服务编写一个启动类,一定要添加一个@EnableEurekaServer注解,开启eureka的注册中心功能:
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.server.EnableEurekaServer;@SpringBootApplication
@EnableEurekaServer
public class DemoApplication {public static void main(String[] args) {SpringApplication.run(DemoApplication.class, args);}
}
编写配置文件
编写一个application.yml文件,内容如下:
server:port: 10086
spring:application:name: euureka-server
eureka:client:service-url:defaultZone: http://127.0.0.1:10086/eureka
启动服务
启动微服务,然后在浏览器访问:http://127.0.0.1:10086
出现如下图的结果就表示成功了:
服务注册
将user-service注册到eureka-server中
引入依赖
在user-service的pom.xml文件中,引入下面的eureka-client依赖
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
配置文件
在user-service中,修改application.yml配置文件,添加服务名称、erueka地址:
spring:application:name: userservice
eureka:client:service-url:defaultZone: http://127.0.0.1:10086/eureka
启动多个user-service实例
首先,复制user-service启动配置:
选中user-service,右键
配置相关属性
启动两个user-service(端口:一个80,一个82)
查看eureka-server管理界面:
服务发现
将order-service的逻辑修改:向erueka-server拉去user-service的信息,实现服务发现
引入依赖
服务发现、服务注册统一都封装在eureka-client依赖
在order-service的pom文件中,引入下面的eureka-client依赖
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
配置文件
在order-service中,修改application.yml文件,添加服务名称、eureka地址:
spring:application:name: orderservice
eureka:client:service-url:defaultZone: http://127.0.0.1:10086/eureka
服务拉取和负载均衡
实现负载均衡只需要添加一些注解即可
在order-service的OrderApplication中,给RestTemplate这个Bean添加一个@LoadBalanced注解:
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.loadbalancer.LoadBalanced;
import org.springframework.context.annotation.Bean;
import org.springframework.web.client.RestTemplate;@MapperScan("com.dc.order.mapper")
@SpringBootApplication
public class OrderApplication {public static void main(String[] args) {SpringApplication.run(OrderApplication.class, args);}@Bean@LoadBalancedpublic RestTemplate restTemplate() {return new RestTemplate();}
}
修改order-service服务中的OrderService类中的queryOrderById方法,修改访问的url路径,用服务名代替ip、端口:
import com.dc.order.mapper.OrderMapper;
import com.dc.order.pojo.Order;
import com.dc.order.pojo.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;@Service
public class OrderService {@Autowiredprivate OrderMapper orderMapper;@Autowiredprivate RestTemplate restTemplate;public Order queryOrderById(Long orderId) {// 1.查询订单Order order = orderMapper.findById(orderId);// 远程查询User// url地址//String url = "http://localhost:80/user/" + order.getUserId();String url = "http://userservice/user/" + order.getUserId();// 发起调用User user = restTemplate.getForObject(url, User.class);order.setUser(user);// 4.返回return order;}
}
spring自动从eureka-server端,根据userservice这个服务名称,获取实例列表,而后完成负载均衡
结果如下:
负载均衡
原理
SpringCloud底层其实利用了一个名为Ribbon的组件,来实现负载均衡功能的
Nacos注册中心
国内公司都比较推崇阿里巴巴的技术,比如注册中心,SpringCloudAlibaba也推出了一个名叫Nacos的注册中心
认识和安装Nacos
Nacos是阿里巴巴的产品,现在是SpringCloud中的一个组件。相比于Eureka功能更加丰富,在国内受欢迎程度较高
安装
在Nacos的GitHub页面可以下载源码和编译好的服务端
主页地址:https://github.com/alibaba/nacos
下载地址:https://github.com/alibaba/nacos/releases
如图:
红线标记的是服务端
解压:
将压缩包解压到非中文路径下
其中
bin:启动脚本
conf:配置文件
端口配置
Nacos的默认端口是8848,如果无法关闭占用8848端口的进程,可以进入nacos的conf目录,修改配置文件application.properties中的端口
启动
进入bin目录中,然后进入命令行,
输入
startup.cmd -m standalone
在浏览器中输入地址:http://127.0.0.1:8848/nacos,用户名和密码为nacos,进入主页如下:
服务注册到nacos
Nacos是SpringCloudAlibaba的组件,而SpringCloudAlibaba也遵循SpringCloud中定义的服务注册、服务发现规范。因此使用Nacos和使用Ereka对于微服务来说,并没有太大区别
主要差别在于:
- 依赖不同
- 服务地址不同
引入依赖
在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>
注意:把eureka的依赖注释掉
配置nacos地址
在user-service和order-service的application.yml配置文件中添加nacos地址:
spring:cloud:nacos:discovery:server-addr: localhost:8848
注意:把关于eureka的配置文件注释
重启
重启微服务后,登录nacos管理页面,可以看到微服务信息:
服务分级存储模型
注意:一个服务可以有多个实例(即:一个服务可以包含多个集群,而每个集群下可以有多个实例,形成分级模型)
微服务互相访问时,应该尽可能访问同集群实例,因为本地访问速度更快。当本集群内不可用时,才访问其他集群。
给user-service配置集群
修改user-service 的application.yml文件,添加集群配置:
spring:cloud:nacos:discovery:server-addr: localhost:8848cluster-name: MZ # 集群名称
重启两个user-service实例后,可以在nacos控制台看到下面结果:
这是再复制一个user-service启动配置,添加属性:
-Dserver.port=81 -Dspring.cloud.nacos.discovery.cluster-name=SH
这次启动三个启动类,再次查看nacos控制台
同集群优先的负载均衡
默认规则是ZoneAvoidanceRule
(即:基于分区下的服务器的可用性选出可用分区列表,再从可用分区列表中随机选择一个分区,采用轮询的策略选择该分区的一个服务器),但这种规则并不能实现根据同集群优先来实现负载均衡
因此Nacos中提供了一个NacosRule
(1、优先选择同集群服务实例列表,2、本地集群中找不到提供者,才去其他集群中寻找,并且给出警告。3、确定可用实例后,在采用随机负载均衡挑选实例)这种规则可以优先从同集群中挑选实例
1、给order-service配置集群信息
修改order-service的application.yml文件,添加集群配置:
spring:cloud:nacos:server-addr: localhost:8848discovery:cluster-name: MZ # 集群名称
2、修改负载均衡规则
修改order-service的application.yml文件,修改负载均衡规则:
userservice:ribbon:NFLoadBalancerRuleClassName: com.alibaba.cloud.nacos.ribbon.NacosRule # 负载均衡规则
权重配置
Nacos提供了权重配置来控制访问频率,权重越大则访问频率越高
注意:如果权重修改为0,则该实例永远不会被访问
在nacos控制台,找到user-service的实例列表,点击编辑,即可修改权重:
在弹出的窗口中,修改权重:
环境隔离
Nacos提供了namespace来实现环境隔离功能。
- nacos中可以有多个namespace
- namespace下可以有group、service等
- 不同namespace之间相互隔离,例如不同namespace的服务互相不可见
配置命名空间的步骤如下:
-
创建namespace:
默认情况下,所有的service、data、group都在同一个namespace下,默认是public:
点击新增命名空间,添加有一个命名空间
命名空间id可以自动生成,标红的两项是必填项
点击确定之后,就可以在页面中看到一个新的命名空间:
给微服务配置namespace
给微服务配置namespace只能通过修改配置来实现
例如:给user-service中的application.yml文件中配置namespace:
重启user-service后,访问nacos控制台,会在dev中出现userservice服务
如果此时访问order-service服务,会报错,这是因为order-service和user-service出现在不同的namespace下
Nacos与Eureka的区别
Nacos的服务实例分为两种类型:
- 临时实例:如果实例宕机超过一定时间,会从服务列表剔除,默认类型
- 非临时实例:如果实例宕机,不会从列表剔除,也可以叫做永久实例
永久实例的配置;
spring:cloud:nacos:discovery:ephemeral: false # 设置为非临时实例
Nacos和Eureka整体结构类似,服务注册、服务拉取、心跳等待,但是也存在一些差异:
共同点:
- 都支持服务注册和服务拉取
- 都支持服务提供者心跳方式做健康检测
区别:
- nacos支持服务端主动检测提供者状态:临时实例采用心跳模式,非临时实例采用主动检测模式
- 临时实例心跳不正常会被剔除,非临时实例则不会被剔除
- nacos支持服务列表变更的消息推送模式,服务列表更新更及时
- nacos集群默认采用方式,当集群中存在非临时实例时,采用CP模式;Eureka采用AP方式。
AP:可用性|分区容错性
CP:一致性|分区容错性
CAP原则:一致性,可用性、分区容错性最多只能同时实现两点,不可能三者兼顾
- 一致性:在分布式系统中的所有数据备份,在同一个时刻是否同样的值(等同于所有节点访问同一份最新的数据副本)
- 可用性:在集群中一部分节点故障后,集群整体是否还能响应客户端的读写请求(对数据更新具备高可用性)。
- 分区容错性:大多数分布式系统都分布在多个子网络。每个子网络就叫做一个区。
分区容错:区间通信可能失败。
Nacos配置管理
统一配置管理
当微服务部署的实例越来越多,达到数十、数百时,逐个修改微服务配置就会很麻烦。这时就需要统一配置管理方案,可以集中管理所有实例的配置。
nacos一方面可以将配置集中管理,及时通知微服务,实现配置的热更新
在nacos中添加配置文件
首先在配置详情中的配置列表中,点击+号
然后在弹出的表单中,填写配置信息,格式的话,目前只支持yaml和properteis文件
从微服务中拉取配置
微服务要拉取nacos中管理的配置,并且与本地的application.yml配置合并,才能完成项目启动
这个获取nacos地址的过程需要借助外界的帮助,这时就需要bootstrap.yml为文件,会在application.yml之前被读取,流程:
引入nacos-config依赖
首先需要在user-service服务中,引入nacos-config的客户端依赖
<!--nacos配置管理依赖-->
<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
添加bootstrap.yml
在user-service中resouce目录下创建bootstrap.yml文件,内容如下:
spring:application:name: userservice # 服务名称profiles:active: dev # 开发环境,这里是devcloud:nacos:server-addr: localhost:8848 #nacos地址config:file-extension: yaml # 后缀名namespace: 29de2d5a-2658-4621-ad52-1b32bcff8224 #命名空间
其实就是读取userservice-dev.yaml文件
注意:这里bootstrap配置文件要与nacos中的配置一一对应,否则会出现
读取nacos配置
在userservice中的usercontroller中添加业务逻辑,读取pattern.dateformat配置信息:
@Value("${pattern.dateformat}")private String dateformat;@GetMapping("now")public String now() {return LocalDateTime.now().format(DateTimeFormatter.ofPattern(dateformat));}
}
重启user-service服务,在页面访问得到结果:
配置热更新
热更新就是修改nacos配置后,微服务无需重启即可让配置生效,也就是配置热更新
实现方式
方式一
在@Value注入的变量所在类上添加@RefreshScope注解:
package com.dc.user.web;import com.dc.user.pojo.User;
import com.dc.user.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.web.bind.annotation.*;import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;@Slf4j
@RestController
@RequestMapping("/user")
@RefreshScope
public class UserController {@Value("${pattern.dateformat}")private String dateformat;@GetMapping("now")public String now() {return LocalDateTime.now().format(DateTimeFormatter.ofPattern(dateformat));}
}
这是重启user-service后,在nacos控制台修改配置文件,如下:
再次访问页面,结果如下:
方式二
使用@ConfigurationProperties注解代替@Value注解,这时需要创建一个类,来读取配置信息
package com.dc.user.utils;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;@Component
@Data
@ConfigurationProperties(prefix = "pattern")
public class PatternProperties {private String dateformat;
}
UserController类
package com.dc.user.web;import com.dc.user.pojo.User;
import com.dc.user.service.UserService;
import com.dc.user.utils.PatternProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.web.bind.annotation.*;import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;@Slf4j
@RestController
@RequestMapping("/user")
@RefreshScope
public class UserController {@Autowiredprivate PatternProperties patternProperties;@GetMapping("/now")public String now() {return LocalDateTime.now().format(DateTimeFormatter.ofPattern(patternProperties.getDateformat()));}
}
然后重启user-service服务之后,访问页面:
这是在naocos控制台修改配置文件
然后刷新页面
配置共享
微服务启动时,会去nacos读取多个配置文件
添加一个环境共享配置
在nacos配置列表中添加一个userservice.yaml配置文件
在userservice中读取共享配置
在userservice服务中,修改PatternProperties类,读取新添加的属性
package com.dc.user.utils;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;@Component
@Data
@ConfigurationProperties(prefix = "pattern")
public class PatternProperties {private String dateformat;private String envSharedValue;
}
然后修改UserController,添加一个方法
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;@Slf4j
@RestController
@RequestMapping("/user")
@RefreshScope
public class UserController {@Autowiredprivate PatternProperties patternProperties;@GetMapping("/prop")public PatternProperties prop() {return patternProperties;}
}
运行两个userservice服务,并设置不同的profile(配置文件)
将一个设置为dev,另一个设置为test
启动两个服务,并通过页面访问,结果
这时可以看到,无论是dev还是test,都可以读取到enSharedValue这个属性的值
配置共享的优先级
当nacos、服务本地同时出现相同属性时,优先级有高低之分
Feign远程调用
之前使用RestTemplete发起远程调用的方式,存在以下问题:
- 代码可读性差,变成体验不统一
- 参数复杂URL难以维护
Feign是一个声明式的http客户端,官网地址:https://github.com/OpenFeign/feign
其作用就是实现http请求的发送,解决上面的问题
Openfeign:是一种声明式的web
工具,可以使用它的注解创建接口,从而实现服务的远程调用,OpenFeign不做任何请求处理,通过处理注解相关信息生成Request
,并对调用返回的数据进行解码,从而实现 简化 HTTP API 的开发
需要创建一个接口并对其添加Feign
相关注解,另外Feign
还支持可插拔编码器和解码器,致力于打造一个轻量级HTTP
客户端,Feign
最早是由 Netflix 公司进行维护的,后来Netflix
不再对其进行维护,最终 Feign 由社区进行维护,更名为Openfeign
Feign替代RestTemplate
使用步骤:
引入依赖
在order-service服务的pom文件中,引入fegin的依赖
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
添加注解
在order-service服务的启动类上添加注解开启feign的功能
package com.dc.order;import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.loadbalancer.LoadBalanced;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.annotation.Bean;
import org.springframework.web.client.RestTemplate;@MapperScan("com.dc.order.mapper")
@SpringBootApplication
@EnableFeignClients
public class OrderApplication {public static void main(String[] args) {SpringApplication.run(OrderApplication.class, args);}@Bean@LoadBalancedpublic RestTemplate restTemplate() {return new RestTemplate();}
}
编写feign的客户端
在order-service服务中新建一个UserClient接口
package com.dc.order.client;import com.dc.order.pojo.User;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;@FeignClient("userservice")
public interface UserClient {@GetMapping("/user/{id}")User findById(@PathVariable("id") Long id);
}
这个客户端是基于SpringMVC的注解来声明远程调用的信息,如:
- 服务名称:userservice
- 请求方式:Get
- 请求路径:/user/{id}
- 请求参数:Long id
- 返回值类型: User
这样,就无需使用RestTemplate来发送http请求了
测试
修改order-service中的OrderService类中的queryById方法,使用Feign客户端来代替RestTemplate
package com.dc.order.service;import com.dc.order.client.UserClient;
import com.dc.order.mapper.OrderMapper;
import com.dc.order.pojo.Order;
import com.dc.order.pojo.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;@Service
public class OrderService {@Autowiredprivate OrderMapper orderMapper;/*@Autowiredprivate RestTemplate restTemplate;*/@Autowiredprivate UserClient userClient;public Order queryOrderById(Long orderId) {// 1.查询订单Order order = orderMapper.findById(orderId);// 远程查询UserUser user = userClient.findById(order.getUserId());/*// url地址//String url = "http://localhost:80/user/" + order.getUserId();String url = "http://userservice/user/" + order.getUserId();// 发起调用User user = restTemplate.getForObject(url, User.class);*/order.setUser(user);// 4.返回return order;}
}
查询结果:
自定义配置
feign可以支持很多的自定义配置,如下表所示
类型 | 作用 | 说明 |
---|---|---|
feign.Logger.Level | 修改日志级别 | 包含四种不同的级别:NONE、BASIC、HEADERS、FULL |
feign.codec.Decoder | 响应结果的解析器 | http远程调用的结果做解析,例如解析json字符串为Java对象 |
feign.codec.Encoder | 请求参数编码 | 将请求参数编码,便于通过http请求发送 |
feign.Contract | 支持的注解格式 | 默认是SpringMVC的注解 |
feign.Retryer | 失败重试机制 | 请求失败的重试机制,默认是没有,不过会使用Ribbon的重试 |
一般请求下,默认值就能满足使用,如果使用自定义时,只需要创建自定义的 @Bean覆盖默认Bean即可
日志案例
配置文件方式
方式一:配置文件
基于配置文件修改feign的日志级别可以针对单个服务
feign:client:config:userservice:logger-level: FULL
也可以针对所有服务:
feign: client:config: default: # 这里用default就是全局配置,如果是写服务名称,则是针对某个微服务的配置loggerLevel: FULL # 日志级别
注意:使用配置文件方式时,需要设置log日志级别,这样才能看到
结果如下:
07-22 15:59:54:502 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById] <--- HTTP/1.1 200 (1091ms)
07-22 15:59:54:502 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById] connection: keep-alive
07-22 15:59:54:502 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById] content-type: application/json
07-22 15:59:54:502 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById] date: Sat, 22 Jul 2023 07:59:54 GMT
07-22 15:59:54:502 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById] keep-alive: timeout=60
07-22 15:59:54:502 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById] transfer-encoding: chunked
07-22 15:59:54:503 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById]
07-22 15:59:54:503 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById] {"id":2,"username":"文二狗","address":"陕西省西安市"}
07-22 15:59:54:503 DEBUG 15304 --- [nio-8080-exec-1] com.dc.order.client.UserClient : [UserClient#findById] <--- END HTTP (62-byte body)
而日志级别可以分为四种
-
NONE:不记录任何日志信息,这是默认值
-
BASIC:仅记录请求的方法,URL以及响应状态码和执行时间
-
HEADERS:在BASIC的基础上,额外记录了请求和响应的头信息
-
FULL:记录请求和响应的明细,包括头信息、请求体、元数据
Java代码方式
也可以基于Java代码来修改日志级别,先声明一个类,然后声明一个Logger.level的对象
public class DefaultFeignConfiguration {@Beanpublic Logger.Level feignLogLevel(){return Logger.Level.BASIC; // 日志级别为BASIC}
}
如果要全局生效,将其字节码放到@EnableFeignClients这个注解中:
@EnableFeignClients(defaultConfiguration = DefaultFeignConfiguration .class)
如果是局部生效,则把它放到对应的@FeignClient这个注解中
@FeignClient(value = "userservice", configuration = DefaultFeignConfiguration .class)
Feign使用优化
Feign底层发起http请求,依赖于其它框架。其底层客户端实现包括:
- URLConnection:默认实现,不支持连接池
- Apache HttpClient:支持连接池
- OKHttp:支持连接池
因此提高Feign的性能主要手段就是使用连接池代替默认的URLConnection
配置
引入依赖
在order-service的pom文件中引入Apache的HttpClient依赖
<!--httpClient的依赖 -->
<dependency><groupId>io.github.openfeign</groupId><artifactId>feign-httpclient</artifactId>
</dependency>
配置连接池
在order-service的application.yml中添加配置
feign:client:config:userservice:logger-level: FULLhttpclient:enabled: true #开启feign对HttpClient的支持max-connections: 200 #最大的连接数max-connections-per-route: 50 #每个路径的最大连接数
总结:
- 日志级别尽量使用basic
- 使用HttpClient或OKHttp代替URLConnection
- 引入feign-httpClient依赖
- 配置文件开启httpClient功能,设置连接池参数
最佳实践
所谓最佳实践就是使用过程中总结的经验,最好的一种使用方式
简化Feign客户端代码
继承方式
一样的代码可以通过继承来共享:
- 定义一个API接口,利用定义方法,并基于SpringMVC注解做声明
- Feign客户端和Controller都集成该接口
优点:
- 简单
- 实现了代码共享
缺点:
- 服务提供方、服务消费方紧耦合
- 参数列表中的注解映射并不会继承,因此Controller中必须再次声明方法、参数列表、注解
抽取方式
将Feign的Client抽取为独立模块,并且把接口有关的POJO、默认的Feign配置都放到这个模块中,提供给所有消费者使用,如:
实现基于抽取的最佳实践
抽取
首先创建一个module,命名为feign-api
然后导入feign的starter依赖
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
然后将order-service中编写的UserClient、User、DefaultFeignConfiguration剪切到feign-api中
在order-service中使用feign-api
在order-service中引入feign-api的依赖
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
扫描包
方式一
指定Feign应该扫描的包:
@EnableFeignClients(basePackages = "com.dc.feign.client")
方式二
指定需要加载的Client接口:
@EnableFeignClients(clients = {UserClient.class})
3.4 OpenFeign的底层原理
1.在 Spring 项目启动阶段,启动类上的@EnableFeignClients注解,会引入一个FeignClientsRegistrar(Feign客户端注册类),它会从指定的目录下扫描并加载所有被 @FeignClient 注解修饰的接口类(interface),然后将这些接口类型注册成 Bean对象,统一交给 Spring 来管理。
2.@FeignClient 修饰的接口类的方法,经过 MVC Contract 协议的解析后,放入 MethodMetadata(方法元数据)数组中。
3.然后创建一个动态代理对象Proxy ,指向了一个存放着key为@FeignClient 修饰的接口类的方法名,和 value为方法名对应的MethodHandler (MethodHandler 记录着MethodMetadata方法元数据的引用)的 HashMap。然后把动态代理对象Proxy添加到 Spring 容器中,并注入到对应的服务里。
4.当服务调用FeignClient接口类的方法时,从动态代理对象 Proxy 中找到一个 MethodHandler 实例,生成一个包含该方法URL的 Http请求(不包含服务的 IP)。
5.经过loadbalancer负载均衡算法找到一个服务的 IP 地址,拼接出完整的 URL,并发起请求。
6.被调用服务收到Http请求,就可以响应请求,返回数据给调用者了。
GateWay服务网关
Gateway网关是所有微服务的统一入口
网关的核心功能特性:
- 请求路由
- 权限控制
- 限流
架构图:
权限控制:网关作为微服务入口,需要校验用户是否有请求资格,如果没有则进行拦截
路由和负载均衡:一切请求都必须先经过gateway,但网关不处理业务,而是根据某种规则,把请求转发到某个微服务,这个过程就是路由,当路由的目标服务有多个时,还需要做负载均衡
限流:当请求流量过高时,在网关中按照下流的微服务能够接受的速度来放行请求,避免服务压力过大
在SpringCloud中网关的实现包括两种:
- gateway
- zuul
Zuul是基于Servlet的实现,属于阻塞式编程。而SpringCloudGateway则是基于Spring5中提供的WebFlux,属于响应式的实现,具备更好的性能
gateway入门
步骤:
- 创建SpringBoot工程gateway,引入网关依赖
- 编写启动类
- 编写基础配置和路由规则
- 启动网关服务进行测试
创建gateway服务,引入依赖
<!--网关-->
<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>
编写启动类
package com.dc.gate;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;@SpringBootApplication
public class GatewayApplication {public static void main(String[] args) {SpringApplication.run(GatewayApplication.class, args);}
}
编写配置文件
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/** # 这是按照路径匹配,只要以/use/开头就符合要求
重启测试
重启服务,当访问http://localhost:10010/user/1时,符合/user/**的规则,请求转发到uri:http://userservice/user/1得到结果:
注意:若出现如下报错:
***************************
APPLICATION FAILED TO START
***************************
Description:
Parameter 0 of method modifyResponseBodyGatewayFilterFactory in org.springframework.cloud.gateway.config.GatewayAutoConfiguration required a bean of type 'org.springframework.http.codec.ServerCodecConfigurer' that could not be found.Action:
Consider defining a bean of type 'org.springframework.http.codec.ServerCodecConfigurer' in your configuration.
删除pom文件中的spring-boot-start-web依赖即可。这是因为spring cloud gateway是基于webflux的,它与spring cloud网关不兼容。
网关路由的流程图
整个访问的流程如下:
总结:
网管搭建步骤:
- 创建项目,引入nacos服务发现和gateway依赖
- 配置application.yml,包括服务基本信息、nacos地址、路由
路由配置包括:
- 路由id:路由的唯一标识
- 路由目标(uri):路由的目标地址,http代表固定地址,lb代表根据服务名负载均衡
- 路由断言(predicates):判断路由的规则
- 路由过滤器(filters):对请求或响应做处理
断言工厂
在配置文件中的断言规则只是字符串,这些字符串会被Predicate Factory读取并处理,转变为路由判断的条件
如Path=/user/**是按照路径匹配的,这个规则是由org.springframework.cloud.gateway.handler.predicate.PathRoutePredicateFactory类来处理的,像这样的断言工厂在SpringCloudGateway还有十几个:
名称 | 说明1 | 示例 |
---|---|---|
After | 是某个时间点后的请求 | - After=2037-01-20T17:42:47.789-07:00[America/Denver] |
Before | 是某个时间点之前的请求 | - Before=2031-04-13T15:14:47.433+08:00[Asia/Shanghai] |
Between | 是某两个时间点之前的请求 | - Between=2037-01-20T17:42:47.789-07:00[America/Denver], 2037-01-21T17:42:47.789-07:00[America/Denver] |
Cookie | 请求必须包含某些cookie | - Cookie=chocolate, ch.p |
Header | 请求必须是指定方式 | - Header=X-Request-Id, \d+ |
Host | 请求必须是访问某个域名(host) | - Host=.somehost.org,.anotherhost.org |
Method | 请求方式必须是指定方式 | - Method=GET,POST |
Path | 请求路径必须符合指定规则 | - Path=/red/{segment},/blue/** |
Query | 请求参数必须包含指定参数 | - Query=name, Jack或者- Query=name |
RemoteAddr | 请求者的ip必须是指定范围 | - RemoteAddr=192.168.1.1/24 |
Weight | 权重处理 |
过滤器工厂
GatewayFilter是网关中提供的一种过滤器,可以对进入网关的请求和微服务返回的响应做处理:
路由过滤器的种类
Spring提供了31种不同的路由过滤器工厂。如:
名称 | 说明 |
---|---|
AddRequestHeader | 给当前请添加一个请求头 |
RemoveRequestHeader | 移除请求中的一个请求头 |
AddResponseHeader | 给响应结果中添加一个响应头 |
RemoveResponseHeader | 从响应结果中移除一个响应头 |
RequestRateLimiter | 限制请求的流量 |
案例
需求:给所有访问userservice的请求添加一个请求头:Truth=hello dc!
修改gateway服务的application.yml文件,添加路由过滤即可(局部实现):
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/** # 这是按照路径匹配,只要以/use/开头就符合要求filters:- AddRequestHeader=Truth, hello dc! #请求添加请求头
默认过滤:
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/** # 这是按照路径匹配,只要以/use/开头就符合要求default-filters:- AddRequestHeader=Truth, hello dc! #请求添加请求头
修改user-service服务中的UserController中的方法:
@GetMapping("/{id}")
public User queryById(@PathVariable("id") Long id, @RequestHeader("Truth") String truth) {System.out.println(truth);return userService.queryById(id);
}
结果:
总结:
过滤器的作用是什么?
- 对路由的请求或响应做加工处理,比如添加请求头
- 配置在路由下的过滤器只对当前路由的请求生效
defaultFilters的作用是什么?
对所有路由都生效的过滤器
全局过滤器作用
全局过滤器的作用是处理一切进入网关的请求和微服务响应,与GatewayFilter的作用一样。区别在于GatewayFilter通过配置定义,处理逻辑是固定的;而GlobalFilter的逻辑需要自己写代码实现
定义方式是需要实现GlobalFilter接口
在filter中编写自定义逻辑,可以实现下列功能
- 登录状态判断
- 权限判断
- 请求限制流等
自定义全局过滤器
需求:定义全局过滤器,拦截请求,判断请求的参数是否满足下面条件:
- 参数中是否有authorization
- authorization参数值是否为admin
如果同时满足则放行,否则拦截
实现:
在gateway中定义一个过滤器
package com.dc.gate.filter;import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.annotation.Order;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;@Order(-1)
@Component
public class AuthorizeFilter implements GlobalFilter {@Overridepublic Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {// 1、获取请求参数MultiValueMap<String, String> queryParams = exchange.getRequest().getQueryParams();// 2、获取authorization参数String auth = queryParams.getFirst("authorization");// 校验if ("admin".equals(auth)){// 放行return chain.filter(exchange);}// 拦截// 禁止访问,设置状态码exchange.getResponse().setStatusCode(HttpStatus.FORBIDDEN);// 结束处理return exchange.getResponse().setComplete();}
}
注意:order注解的作用是定义Spring IOC容器中Bean的执行顺序的优先级,而不是定义Bean的加载顺序,Bean的加载顺序不受@Order或Ordered接口的影响
order默认是最低优先级,值越小,优先级越高
此时在访问就会出现禁止访问的提示
如果在请求地址中加上authorization=admin的条件就可以访问
过滤器执行顺序
请求进入网关会碰到三类过滤器:当前路由的过滤器、DefaultFilter、GlobalFilter请求路由后,会将当前路由过滤器和DefaultFilter、GlobalFilter,合并到一个过滤器链(集合)中,排序后依次执行每个过滤器:
排序的规则:
- 每个过滤器都必须指定一个int类型的order值,order值越小,优先级越高,执行顺序越靠前
- GlobalFilter通过实现Ordered接口,或者添加@Order注解来指定order值,由我们自己指定
- 路由过滤器和defaultFilter的order由Spring指定,默认是按照声明顺序从1递增
- 当过滤器的order值一样时,会按照defaultFilter > 路由过滤器 > GlobalFilter的顺序执行
跨域问题
跨域:域名不一致就是跨域,主要包括:
- 域名不同:www.taobao.com 和 www.taobao.org 和 www.jd.com 和 miaosha.jd.com
- 域名相同,端口不同:localhost:8080和localhost8081
跨域问题:浏览器禁止请求的发起者与服务端发生跨域ajax请求,请求被拦截器拦截问题
解决跨域问题
在gateway服务的applicaiton.yml文件中,添加如下配置
server:port: 10010 #网关端口
spring:application:name: gateway #服务名称cloud:nacos:server-addr: localhost:8848 #nacos地址gateway:# 。。。globalcors: # 全局的跨域处理add-to-simple-url-handler-mapping: true # 解决options请求被拦截问题corsConfigurations:'[/**]':allowedOrigins: # 允许哪些网站的跨域请求- "http://localhost:8090"allowedMethods: # 允许的跨域ajax的请求方式- "GET"- "POST"- "DELETE"- "PUT"- "OPTIONS"allowedHeaders: "*" # 允许在请求中携带的头信息allowCredentials: true # 是否允许携带cookiemaxAge: 360000 # 这次跨域检测的有效期routes: #网关路由配置- id: user-service #路由id,自定义,只要唯一即可# uri: http://127.0.0.1:8081 # 路由的目标地址 http就是固定地址uri: lb://userservice #路由的目标地址 lb是负载均衡,后面是服务名称predicates: # 路由断言,也就是判断请求是否符合路由规则的条件- Path=/user/** # 这是按照路径匹配,只要以/use/开头就符合要求filters:- AddRequestHeader=Truth, hello dc! #请求添加请求头
前端页面:
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><meta http-equiv="X-UA-Compatible" content="ie=edge"><title>Document</title>
</head>
<body>
</body>
<script src="https://unpkg.com/axios/dist/axios.min.js"></script>
<script>axios.get("http://localhost:10010/user/1?authorization=admin").then(resp => console.log(resp.data)).catch(err => console.log(err))
</script>
</html>
Docker实用
简介
微服务虽然具备各种各样的优势,但服务的拆分通用给部署带来了很多的麻烦
- 分布式系统中,依赖的组件非常多,不同组件之间部署时往往会产生一些冲突
- 在数百上千台服务中重复部署,环境不一定一致,会遇到各种问题
应用部署的环境问题
大型项目组件较多,运行环境也较为复杂,部署时会遇到一些问题如:
- 依赖关系复杂,容易出现兼容性问题
- 开发、测试、生产环境有差异
docker解决依赖兼容问题
两种手段:
- 将应用的libs(函数库)、deps(依赖)、配置与应用一起打包
- 将每个应用放到一个隔离容器中运行,避免互相干扰
这样打包好的应用包中,既包含应用本身,也保护应用所需要的libs、deps,无需在操作系统上安装这些,自然就不存在不同应用之间的兼容问题
docker解决操作系统环境差异
首先介绍一下操作系统的结构
Ubuntu操作系统:
- 计算机硬件:CPU、内存、磁盘等
- 系统内核:所有Linux发行版的内核都是Linux,例如CentOS、Ubuntu、Fedora等。内核可以与计算机硬件交互,对外提供内核命令,用于操作计算机硬件
- 系统应用:操作系统本身提供的应用、函数库。这些函数库是对内核指令的封装,使用更加方便
应用于计算机交互的流程:
- 应用调用操作系统应用(函数库),实现各种功能
- 系统函数是对内核指令集的封装,会调用内核指令
- 内核指令操作计算机硬件
Ubuntu和CentOS都是基于Linux内核,无非是系统应用不同,提供的函数库有差异:
如果将一个Ubuntu版本的MySQL应用安装到CentOS系统,MySQL在调用Ubuntu函数库时,会发现找不到或者不匹配,就会报错
docker解决不同系统环境的问题:
- docker将用户程序于所需要调用的系统函数库一起打包
- docker运行到不同操作系统时,直接基于打包的函数库,借助于操作系统的Linux内核来运行
总结:
docker如何解决大型项目依赖关系复杂和不同组件依赖的兼容性问题?
- docker允许开发中将应用、依赖、函数库、配置一起打包,形成可移植镜像
- docker应用运行在容器中,使用沙箱机制,相互隔离
docker如何解决开发、测试、生产环境有差异的问题?
- docker镜像中包含完整运行环境,包括系统函数库,仅依赖系统的Linux内核,因此可以在任意Linux操作系统上运行
docker是一个快速交付应用、运行应用的技术,具备下列优势:
- 可以将程序及其依赖、运行环境一起打包为一个镜像,可以迁移到任意Linux操作系统
- 运行时利用沙箱机制形成隔离容器,各个应用互不干扰
- 启动、移除都可以通过一行命令完成,方便快捷
docker和虚拟机的区别
虚拟机:(virtual machine)是在操作系统中模拟硬件设备,然后运行另一个操作系统,比如在windows系统中运行Ubuntu系统,这样就可以运行任意的ubuntu应用了
docker:仅仅是封装函数库,并没有模拟完整的操作系统
对比:
特性 | docker | 虚拟机 |
---|---|---|
性能 | 接近原生 | 性能较差 |
硬盘占用 | 一般为MB | 一般为GB |
启动 | 妙级 | 分钟级 |
docker和虚拟机的差异:
-
docker是一个系统进程;虚拟机是在操作系统中的操作系统
-
docker体积小、启动速度快、性能好;虚拟机体积大、启动速度慢、性能一般
docker架构
概念
镜像(Image):docker将应用程序及其所需的依赖、函数库、环境、配置等文件打包在一起,称为镜像
容器(Container):镜像中的应用程序运行后形成的进程就是容器,只是docker会给容器进程做隔离,对外不可见
一切应用最终都是代码组成,都是硬盘中的一个个的字节形成的文件。只有运行时,才会加载到内存,形成进程
镜像就是把一个应用在硬盘上的文件、及其运行环境、部分系统函数库文件一起打包形成的文件包。这个文件包是只读的
容器是将这些文件中编写的程序、函数加载到内存中允许,形成进程,并且需要隔离。因此一个镜像可以启动多次,形成多个容器进程
DockerHub
开源的应用有很多,但是打包这些应用是重复且乏味的劳动。因此就出现了镜像托管的网站
- Dockerhub:DockerHub是一个官方的Docker镜像的托管平台。这种平台称为Docker Registry
- 国内类似的公开服务,比如网易云镜像、阿里云镜像等
可以拉取自己需要的镜像
Docker架构
Docker是一个CS架构的程序,有两个部分组成:
- 服务端(server):Docker守护进行,负责处理Docker指令,管理镜像、容器等
- 客户端(client):通过命令或RestAPI向Docker服务端发送指令。可以在本地或远程向服务端发送指令
Docker安装
想要安装docker需要先将依赖的环境下载,默认下载的地址是国外的服务器,速度较慢,可以设置为阿里云的镜像源,速度会更快
yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo
采用yum的方式安装
yum makecache fast
yum -y install docker-ce
注意:docker应用到各种端口,需要逐一去修改防火墙设置。这里直接关闭防火墙
# 关闭
systemctl stop firewalld
# 禁止开机启动防火墙
systemctl disable firewalld
测试
安装成功后,需要手动启动,设置为开启自启,并测试一下docker
# 启动Docker服务
systemctl start docker
# 设置开机自动启动
systemctl enable docker
# 测试 运行hello-world 镜像 根据这个镜像 创建容器
docker run hello-world
基本操作
镜像名称
首先镜像的名称组成:
- 镜像名称一般分为两个部分:[repository]:[tag]
- 在没有指定tag时,默认是latest,代表最新版本的镜像
镜像命令
常见的镜像命令如图:
案例
需求:从dockerhub中拉去一个nginx镜像并查看
步骤:
- 首先去镜像仓库中搜索nginx镜像,比如https://hub.docker.com
- 根据查看到的镜像名称,拉取自己需要的镜像,通过命令docker pull nginx
- 通过命令:docker images 查看拉取到的镜像
需求:利用docker save 将nginx镜像导出磁盘,然后通过load加载回来
步骤:
-
查看save命令用法,可以输出命令
docker save --help
-
命令格式
docker save -o [保存的目标文件名称] [镜像名称]
-
使用docker save导出镜像到磁盘
docker save -o nginx.tar nginx:latest
-
使用docker 加载load加载镜像
docker rmi nginx:latest
然后运行命令,加载本地文件
docker load -i nginx.tar
容器操作
容器相关命令
容器保护三个状态:
- 运行:进程正常运行
- 暂停:进程暂停、CPU不再运行,并不释放内存
- 停止:进程停止,回收进程占用的内存、cpu等资源
其中:
- docker run:创建并运行一个容器,处于运行状态
- docker pause:让一个运行的容器暂停
- docker unpause:r让一个容器从暂停状态恢复运行
- docker stop:停止一个运行的容器
- docker start:让一个停止的容器再次运行
- docker rm:删除一个容器
需求:创建并运行一个容器
docker run --name containerName -p 80:80 -d nginx
命令解读:
- docker run:创建并运行一个容器
- –name:给容器一个名字
- -p:将宿主机端口与容器端口映射,冒号左侧是宿主机端口,右侧是容器端口
- -d:后台运行容器
- nginx:镜像名称
这个命令是将容器和宿主机关联,这样当访问宿主机是就可以映射到容器中
需求:进入nginx容器,修改HTML文件内容
步骤:
-
进入容器
docker exec -it mn bash
命令解读:
- docker exec:进入容器内部,执行一个命令
- -it:给当前进入的容器创建一个标准输入、输出终端,允许与容器交互
- mn:要进入的容器的名称
- bash:进入容器后执行的命令,bash是一个linux终端交互命令
-
进入nginx的HTML文件所在目录/usr/share/nginx/html
进入该目录
cd /usr/share/nginx/html
-
修改index.html的内容
sed -i -e 's#Welcome to nginx#HELLO#g' -e 's#<head>#<head><meta charset="utf-8">#g' index.html
数据卷
数据卷(volume):是一个虚拟目录,指向宿主机文件系统中的某个目录
一旦完成数据卷挂载,对容器的一切操作都会作用在数据卷对应的宿主机目录
操作命令
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
挂载数据卷
通过-v参数来挂在一个数据卷到某个容器内目录,命令格式:
docker run \--name mn \-v html:/root/html \-p 8080:80nginx \
-v就是挂载数据卷的命令
- -v html:/root/html:把html数据卷挂载到容器内的/root/html这个目录中
Docker-Compose
下载Docker-Compose
去github官网搜索docker-compose,下载1.24.1版本的Docker-Compose
下载路径:https://github.com/docker/compose/releases/download/1.24.1/docker-compose-Linux-x86_64
设置权限
需要将DockerCompose文件的名称修改一下,给予DockerCompose文件一个可执行的权限
mv docker-compose-Linux-x86_64 docker-compose
chmod 777 docker-compose
配置环境变量
方便后期操作,配置一个环境变量
将docker-compose文件移动到了/usr/local/bin , 修改了/etc/profile文件,给/usr/local/bin配置到了PATH中
mv docker-compose /usr/local/binvi /etc/profile
# 添加内容: export PATH=$JAVA_HOME:/usr/local/bin:$PATHsource /etc/profile
安装方式(手动)
# 直接联网下载到本地 /usr/local/bin/docker-compose
curl -L https://get.daocloud.io/docker/compose/releases/download/1.26.2/docker-compose-`uname -s`-`uname -m` > /usr/local/bin/docker-composecd /usr/local/bin # 进入该目录
chmod 777 docker-compose # 给这个文件授权# 在任意目录 测试 docker-compose 命令
测试
在任意目录下输入docker-compose
测试效果 |
---|
Docker-Compose管理MySQL和Tomcat容器
yml文件以key: value方式来指定配置信息
多个配置信息以换行+缩进的方式来区分
在docker-compose.yml文件中,不要使用制表符
version: '3.1'
services:mysql: # 服务的名称restart: always # 代表只要docker启动,那么这个容器就跟着一起启动image: daocloud.io/library/mysql:5.7.4 # 指定镜像路径container_name: mysql # 指定容器名称ports:- 3306:3306 # 指定端口号的映射environment:MYSQL_ROOT_PASSWORD: root # 指定MySQL的ROOT用户登录密码TZ: Asia/Shanghai # 指定时区volumes:- /opt/docker_mysql_tomcat/mysql_data:/var/lib/mysql # 映射数据卷tomcat:restart: alwaysimage: daocloud.io/library/tomcat:8.5.15-jre8container_name: tomcatports:- 8080:8080environment:TZ: Asia/Shanghaivolumes:- /opt/docker_mysql_tomcat/tomcat_webapps:/usr/local/tomcat/webapps- /opt/docker_mysql_tomcat/tomcat_logs:/usr/local/tomcat/logs
使用docker-compose命令管理容器
在使用docker-compose的命令时 ,默认会在当前目录下找docker-compose.yml文件
# 1. 基于docker-compose.yml启动管理的容器
docker-compose up -d# 2. 关闭并删除容器
docker-compose down# 3. 开启|关闭|重启已经存在的由docker-compose维护的容器
docker-compose start|stop|restart# 4. 查看由docker-compose管理的容器
docker-compose ps# 5. 查看日志
docker-compose logs -f
RabbitMQ
介绍
微服务间的通讯有同步和异步两种方式:
同步通讯:需要实时响应,如打电话
异步通讯:不需要马上响应,如邮件
优缺点:
同步:虽然可以立即响应,但是不能跟多个人同时通话
异步:可以给多个人同时发邮件,但是响应会有延迟
同步通讯
Feign调用就是同步方式,虽然可以实时得到结果,但存在以下问题:
总结:
同步调用的优点:
时效性较强,可以立即得到结果
同步调用的问题:
- 耦合度强
- 性能和吞吐能力下降
- 有额外的资源消耗
- 有级联失效问题
异步通讯
异步调用可以避免上述问题:
例如:发布者(publisher)和订阅者(consumer)问题。
为了解除事件订阅者和发布者之间的耦合,两者并不是直接通讯的,而是通过中间人(Broker)。发布者发布事件到broker,不关心谁来订阅事件。订阅者从broker订阅事件,不关心谁发来消息
Broker是一个数据总线一样的东西,所有的服务器要接收数据和发送数据都发到这个总线上,这个总线就像协议一样,让服务间的通讯变得标准和可控
好处:
- 吞吐量提升:无需等待订阅者处理完成,响应更快速
- 故障隔离:服务没有直接调用,不存在级联失效问题
- 耦合度极低:每个服务都可以灵活插拔,可替换
- 流量削峰:不管发布事件的流量波动多大,都有broker接收,订阅者可以按照自己的速度处理事件
缺点:
- 架构复杂,业务没有明显的流程线,不好管理
- 需要依赖于broker的可靠、安全、性能
技术对比
MQ:消息队列(MessageQueue就是存放消息的队列。也就是事件驱动架构中的broker
常见的MQ实现:
- ActiveMQ
- RabbitMQ
- RocketMQ
- Kafka
几种常见MQ的对比:
RabbitMQ | ActiveMQ | RocketMQ | Kafka | |
---|---|---|---|---|
公司/社区 | Rabbit | Apache | 阿里 | Apache |
开发语言 | Erlang | Java | Java | Scala&Java |
协议支持 | AMQP、XMPP、SMTP、STOMP | OpenWire、STOMP、REST、XMPP、AMQP | 自定义协议 | 自定义协议 |
可用性 | 高 | 一般 | 高 | 高 |
单机吞吐量 | 一般 | 差 | 高 | 非常高 |
消息延迟 | 微秒级 | 毫秒级 | 毫秒级 | 毫秒以内 |
消息可靠性 | 高 | 一般 | 高 | 一般 |
追求可用性:Kafka、RocketMQ、RabbitMQ
追求可靠性:RabbitMQ、RocketMQ
快速入门
RabbitMQ消息模型
基本结构:
RabbitMQ中的一些角色:
-
publisher:生产者
-
consumer:消费者
-
exchange:交换机,负责消息路由
-
queue:队列,存储消息
-
virtualHost:虚拟主机,隔离不同租户的exchange、queue、消息的隔离
入门案例
简单队列模式的模型图:
官方的helloworld是基于最基础的消息队列模型来实现的,只包括三个角色:
- publisher:消息发布者、将消息发送到队列queue
- queue:消息队列,负责接收并缓存消息
- consumer:订阅队列,处理队列中的消息
publisher实现
思路:
- 建立连接
- 创建Channel
- 声明队列
- 发送消息
- 关闭连接hechannel
代码实现:
package com.dc.mqtest.helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import org.springframework.boot.test.context.SpringBootTest;import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class PublsherTest {@Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.发送消息String message = "hello, rabbitmq!";channel.basicPublish("", queueName, null, message.getBytes());System.out.println("发送消息成功:【" + message + "】");// 5.关闭通道和连接channel.close();connection.close();}
}
consumer实现
思路:
- 建立连接
- 创建channel
- 声明队列
- 订阅消息
代码实现:
package com.dc.mqtest.helloworld;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {// 5.处理消息String message = new String(body);System.out.println("接收到消息:【" + message + "】");}});System.out.println("等待接收消息。。。。");}
}
消息模式
点对点消息
发送端直接把消息发送到队列中,消费者直接从队列中获取消息,消息只能消费一次
代码展示:
创建消息队列
package com.dc.mptest01.config;import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitCreateConfig {// 创建RabbitMQ的队列 存储消息@Beanpublic Queue createQ() {return new Queue("java");}
}
实现消息发送
package com.dc.mptest01.controller;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("/api/mq/")
public class RabbitMQSendController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/send")public String sendMsg(String msg){/*** 发送消息* 参数说明 :* 交换器* 路由关键字或队列名称* 消息内容*/rabbitTemplate.convertAndSend("","java",msg);return "ok";}}
实现消息消费
package com.dc.mptest01.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class ReceiveListener {/*** @RabbitListener 设置需要监听的队列名* 一旦队列有了消息,修饰的方法自动执行* 方法的参数:消息的数据类型*/@RabbitListener(queues = "java")public void handler(String msg) {log.info("接收消息:{}", msg);}}
工作队列消息
一个队列可以有多个消费端,消息只能被消费一次
核心:一个队列有多个消费者
代码展示:
创建消息队列:
package com.dc.mptest01.config;import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitCreateConfig {// 创建RabbitMQ的队列 存储消息@Beanpublic Queue createQ() {return new Queue("java");}@Beanpublic Queue createQ1(){return new Queue("java-m-01");}
}
实现消息发送
package com.dc.mptest01.controller;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("/api/mq/")
public class RabbitMQSendController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/send")public String sendMsg(String msg){/*** 发送消息* 参数说明 :* 交换器* 路由关键字或队列名称* 消息内容*/rabbitTemplate.convertAndSend("","java",msg);return "ok";}@GetMapping("/send/{msg}")public String send(@PathVariable String msg) {rabbitTemplate.convertAndSend("", "java-m-01", msg + "-" + System.currentTimeMillis());return "ok";}
}
实现消息消费:
package com.dc.mptest01.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class ReceiveListener {/*** @RabbitListener 设置需要监听的队列名* 一旦队列有了消息,修饰的方法自动执行* 方法的参数:消息的数据类型*/@RabbitListener(queues = "java")public void handler(String msg) {log.info("接收消息:{}", msg);}@RabbitListener(queues = "java-m-01")public void handler01(String msg) {log.info("接收消息:{}", msg);}}
发布订阅消息
发送端发送消息给交换器(Exchange-Fanout),交换器再把消息发送到绑定的队列(可能有一个或多个),每个队列又有自己的消费端,所以最终实现一个消息可以被消费多次
代码展示:
创建交换器、消息队列、绑定
@Configuration
public class MqInitConfig3 {/**1.创建交换器 发布定义 Exchange-Fanout(直接转发)*/@Beanpublic FanoutExchange createFE(){return new FanoutExchange("java-fanout-dc");}/**2.创建队列*/@Beanpublic Queue createQ3(){return new Queue("java-m-03");}@Beanpublic Queue createQ4(){return new Queue("java-m-04");}/**3.实现绑定*/@Beanpublic Binding createBd1(FanoutExchange fe){return BindingBuilder.bind(createQ3()).to(fe);}@Beanpublic Binding createBd2(FanoutExchange fe){return BindingBuilder.bind(createQ4()).to(fe);}
}
实现消息发送
@RestController
@RequestMapping("/api/mq3/")
public class SendController3 {@Resourceprivate RabbitTemplate template;@GetMapping("send/{msg}")public String send(@PathVariable String msg){template.convertAndSend("java-fanout-dc","",msg+"-"+System.currentTimeMillis());return "OK";}
}
实现消息消费
@Component
@Slf4j
public class MqListener4 {@RabbitListener(queues = "java-m-03")public void handler(String msg){log.info("发布订阅消息,队列03,消息内容:{}",msg);}
}
@Component
@Slf4j
public class MqListener5 {@RabbitListener(queues = "java-m-04")public void handler(String msg){log.info("发布订阅消息,队列04,消息内容:{}",msg);}
}
路由模式
发送端发送消息给交换器,交换器根据消息的路由关键字匹配对应的队列,每个队列又有自己的消费端,所以最终会实现一个消息被多次消费
代码展示:
创建交换器、消息队列、绑定
package com.dc.mptest01.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitCreateConfig {/*** 创建队列*/@Beanpublic Queue create2() {return new Queue("java-m-02");}@Beanpublic Queue create3() {return new Queue("java-m-03");}/*** 创建交换器(路由匹配)*/@Beanpublic DirectExchange createDE() {return new DirectExchange("java-direct-dc");}/*** 实现绑定*/@Beanpublic Binding createBd3(DirectExchange de) {return BindingBuilder.bind(create3()).to(de).with("red");}@Beanpublic Binding createBd4(DirectExchange de) {return BindingBuilder.bind(create2()).to(de).with("blue");}
}
消息发送:
package com.dc.mptest01.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;@RestController
@RequestMapping("/api/mq")
@Slf4j
public class RabbitMQSendController {@Resourceprivate RabbitTemplate rabbitTemplate;@GetMapping("/send/{msg}/{rk}")public String send(@PathVariable String msg, @PathVariable String rk) {rabbitTemplate.convertAndSend("java-direct-dc", rk, msg + System.currentTimeMillis());return "ok";}}
消息消费:
package com.dc.mptest01.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;/*** -----在希望中绽放,在苦难中坚持------** @author 暮辰*/
@Slf4j
@Component
public class ReceiveListener {@RabbitListener(queues = "java-m-02")public void handle1(String msg){log.info("路由消息,red,消息内容:{}", msg);}@RabbitListener(queues = "java-m-03")public void handle2(String msg) {log.info("路由消息,blue,消息内容:{}", msg);}
}
主题模式
发送端发送消息给交换器,交换器根据消息的路由关键字匹配对应的队列
注意:路由关键字匹配支持模糊匹配
*表示一个单词
#表示任意个单词,内容随意
代码展示:
创建交换器、消息队列、绑定
package com.dc.mptest01.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** -----在希望中绽放,在苦难中坚持------** @author 暮辰*/
@Configuration
public class RabbitCreateConfig {/*** 创建队列*/@Beanpublic Queue create2() {return new Queue("java-m-02");}@Beanpublic Queue create3() {return new Queue("java-m-03");}/*** 创建交换器(路由匹配)*/@Beanpublic TopicExchange createDE() {return new TopicExchange("java-topic-dc");}/*** 实现绑定*/@Beanpublic Binding createBd3(TopicExchange de) {return BindingBuilder.bind(create3()).to(de).with("stu.*");}@Beanpublic Binding createBd4(TopicExchange de) {return BindingBuilder.bind(create2()).to(de).with("tea.#");}
}
消息发送
package com.dc.mptest01.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;@RestController
@RequestMapping("/api/mq")
@Slf4j
public class RabbitMQSendController {@Resourceprivate RabbitTemplate rabbitTemplate;@GetMapping("/send/{msg}/{rk}")public String send(@PathVariable String msg, @PathVariable String rk) {rabbitTemplate.convertAndSend("java-topic-dc", rk, msg + System.currentTimeMillis());return "ok";}}
消息消费:
package com.dc.mptest01.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class ReceiveListener {@RabbitListener(queues = "java-m-02")public void handle1(String msg){log.info("主题消息,stu.*,消息内容:{}", msg);}@RabbitListener(queues = "java-m-03")public void handle2(String msg) {log.info("主题消息,tea.#,消息内容:{}", msg);}
}
事务
RabbitMQ也支持事务,一般用来保证消息的发送。如果出现异常事务就会回滚,如果没有出现异常事务就会提交,消息就会被发送成功。也可以一次发送多个消息,为了保证多个消息的一致性也需要开启事务
RabbitMQ的事务:事务可以保证消息传递,通过事务的回滚记录日志,之后定时发送当前消息。
实现步骤:
创建消息队列、事务管理器
package com.dc.mptest01.config;import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MqInitConfig {// 创建队列@Beanpublic Queue createQ() {return new Queue("java-m-tran");}/*** 创建事务管理对象*/@Beanpublic RabbitTransactionManager createRTM(ConnectionFactory factory) {return new RabbitTransactionManager(factory);}
}
发送消息-开启事务
package com.dc.mptest01.controller;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("/api/mq")
public class SendController1 {@Autowiredprivate RabbitTemplate rabbitTemplate;@Transactional@GetMapping("/send/{msg}/{num}")public String send(@PathVariable String msg, @PathVariable int num) {// 开启RabbitMQ事务rabbitTemplate.setChannelTransacted(true);for (int i = 0; i < num; i++) {if (i > 1) {System.err.println(1 / 0); // 模拟异常}rabbitTemplate.convertAndSend("", "java-m-tran", msg+"-"+i+"-"+System.currentTimeMillis());}return "OK";}
}
消息消费
package com.dc.mptest01.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
@Slf4j
public class MqListener1 {@RabbitListener(queues = "java-m-tran")public void handler(String msg) {log.info("事务消息,演示事务,消息内容:{}", msg);}
}
消费端—手动ACK
为了保证消息从队列可靠的达到消费者,RabbitMQ提供了消息确认机制。消费者在订阅队列时,可以指定autoACK参数,当AutoAck等于false时,RabbitMQ会等待消费者显示的回复确认信号后才从内存中移除消息(实际是先打上删除标记,之后再删除)。当autoAck等于true时,RabbitMQ会自动把发送出去的消息置为确认,然后从内存中删除,而不管消费者是否真正的消费到了这些消息
采用消息确认机制后,只要设置autoAck参数为false,消费者就有足够的时间处理消息,不用担心处理消息过程中消费者进程挂掉后消息丢失的问题,因为RabbitMQ会一直等待持有消息直到消费者显示调用Basic.Ack命令为止
实现步骤:
在配置文件中设置手动应答
spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestvirtual-host: /listener: # 监听器 消息消费设置simple:acknowledge-mode: manual # 手动确认 ackpublisher-confirm-type: simple # 通过确认机制publisher-returns: true # 重新发送监听server:port: 8686
编写代码,初始化创建、消息发送
package com.dc.mptest01.config;import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** -----在希望中绽放,在苦难中坚持------** @author 暮辰*/
@Configuration
public class MqInitConfig2 {@Beanpublic Queue create() {return new Queue("java-m-ack");}
}
package com.dc.mptest01.controller;import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("/api/mq")
public class SendController2 {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/send/{msg}")public String send(@PathVariable String msg) {rabbitTemplate.convertAndSend("", "java-m-ack", msg+System.currentTimeMillis());return "ok";}}
消息消费-手动应答
package com.dc.mptest01.listener;import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;@Component
@Slf4j
public class MqListener2 {@RabbitListener(queues = "java-m-ack")public void handler(String msg, Channel channel, Message message) throws IOException {// 手动应答log.info("获取消息:{}" , msg);// 消息成功 RabbitMQ就会删除消息
// channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);//拒绝消息 RabbitMQ就不会删除消息 参数说明:1.消息的唯一id 2.是的应答 false:拒绝 3.是否把消息重新放回到队列 true:放回 下次继续消费 false:unAckedchannel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}
}
死信和延迟队列
死信:RabbitMQ的消息队列中的消息,如果满足一定的条件就会变成死信,死信不会被消费,只能转发到死信转发器
死信交换器DLX(Dead-Letter-Exchange):当消息在一个队列中变成死信之后,被重新发送到另一个交换器中,这个交换器就是DLX,绑定DLX的队列就称为死信队列
死信产生的条件:
- 消息被拒绝
- 消息过期
- 队列达到最大长度
如果使用死信队列的话,需要在定义队列中设置队列参数:x-dead-letter-exchange
RabbitMQ支持两种ttl(有效期)设置:
- 单独消息进行配置ttl
- 整个队列进行配置ttl(居多)
如果队列也有有效期,消息也有有效期,以时间短的为准
实际应用的业务,超时自动处理的业务(订单超时未支付自动取消)
代码展示:
创建队列、交换器、绑定
package com.dc.mptest01.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;@Configuration
public class RabbitInitConfig {/*** 创建队列* 生成死信:有效期 设置死信交换器*/@Beanpublic Queue create() {HashMap<String, Object> map = new HashMap<>();// 设置有效期 15秒map.put("x-message-ttl", 15000);// 设置 对应的死信交换器的名称map.put("x-dead-letter-exchange", "dead-exchange");// 设置 死信消息的路由关键字map.put("x-dead-letter-routing-key", "first");// 创建消息队列return QueueBuilder.durable("x-dead-ttl").withArguments(map).build();}// 接收死信消息@Beanpublic Queue create1() {return new Queue("x-dead-msg");}/*** 创建死信交换器* 死信交换 路由模式*/@Beanpublic DirectExchange createDe() {return new DirectExchange("dead-exchange");}/*** 绑定 实现消息从交换器到队列*/@Beanpublic Binding createB1(DirectExchange de) {return BindingBuilder.bind(create1()).to(de).with("first");}
}
实现消息发送:
package com.dc.mptest01.controller;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("/api/mq")
public class DeadMsgController {@Autowiredprivate RabbitTemplate rabbitTemplate;// 队列的有效期@GetMapping("/send/{msg}")public String send(@PathVariable String msg) {rabbitTemplate.convertAndSend("", "x-dead-ttl", msg+"-"+System.currentTimeMillis());return "OK";}// 消息的有效期@GetMapping("/send2/{msg}")public String send2(@PathVariable String msg) {// 消息对象Message message = new Message((msg + "-" + System.currentTimeMillis()).getBytes());// 设置消息的有效期message.getMessageProperties().setExpiration("25000");rabbitTemplate.convertAndSend("x-dead-ttl", message);return "ok";}
}
实现消息消费
package com.dc.mptest01.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;import java.util.HashMap;@Component
@Slf4j
public class DelayMsgListener {@RabbitListener(queues = "x-dead-msg")public void handler(String msg) {log.info("延迟时间:{}, {}", msg, System.currentTimeMillis());}
}
面试总结
生产者丢失消息
生产者将数据发送到rabbitmq的时候,可能因为网络问题导致数据就在半路给搞丢了
1.使用事务(性能差)
RabbitMQ 客户端中与事务机制相关的方法有三个: channel.txSelect 、channel.txCommit 和 channel.txRollback。channel.txSelect 用于将当前的信道设置成事务模式,channel.txCommit 用于提交事务,channel.txRollback 用于事务回滚。在通过 channel.txSelect 方法开启事务之后,我们便可以发布消息给 RabbitMQ 了,如果事务提交成功,则消息一定到达了 RabbitMQ 中,如果在事务提交执行之前由于 RabbitMQ异常崩溃或者其他原因抛出异常,这个时候我们便可以将其捕获,进而通过执行channel.txRollback 方法来实现事务回滚。注意这里的 RabbitMQ 中的事务机制与大多数数据库中的事务概念并不相同,需要注意区分。
事务确实能够解决消息发送方和 RabbitMQ 之间消息确认的问题,只有消息成功被RabbitMQ 接收,事务才能提交成功,否则便可在捕获异常之后进行事务回滚,与此同时可以进行消息重发。但是使用事务机制会“吸干”RabbitMQ 的性能。
报文
2.发送回执确认(推荐)
生产者将信道设置成 confirm(确认)模式,一旦信道进入 confirm 模式,所有在该信道上面发布的消息都会被指派一个唯一的 ID(从 1 开始),一旦消息被投递到所有匹配的队列之后,RabbitMQ 就会发送一个确认(Basic.Ack)给生产者(包含消息的唯一 ID),这就使得生产者知晓消息已经正确到达了目的地了。如果消息和队列是可持久化的,那么确认消息会在消息写入磁盘之后发出。RabbitMQ 回传给生产者的确认消息中的 deliveryTag 包含了确认消息的序号,此外 RabbitMQ 也可以设置 channel.basicAck 方法中的 multiple 参数,表示到这个序号之前的所有消息都已经得到了处理,注意辨别这里的确认和消费时候的确认之间的异同。
2.RabbitMQ弄丢了数据
为了防止rabbitmq自己弄丢了数据,这个你必须开启rabbitmq的持久化,就是消息写入之后会持久化到磁盘,哪怕是rabbitmq自己挂了,恢复之后会自动读取之前存储的数据,一般数据不会丢。除非极其罕见的是,rabbitmq还没持久化,自己就挂了,可能导致少量数据会丢失的,但是这个概率较小。
设置持久化有两个步骤,第一个是创建queue的时候将其设置为持久化的,这样就可以保证rabbitmq持久化queue的元数据,但是不会持久化queue里的数据;第二个是发送消息的时候将消息的deliveryMode设置为2,就是将消息设置为持久化的,此时rabbitmq就会将消息持久化到磁盘上去。必须要同时设置这两个持久化才行,rabbitmq哪怕是挂了,再次重启,也会从磁盘上重启恢复queue,恢复这个queue里的数据。
而且持久化可以跟生产者那边的confirm机制配合起来,只有消息被持久化到磁盘之后,才会通知生产者ack了,所以哪怕是在持久化到磁盘之前,rabbitmq挂了,数据丢了,生产者收不到ack,你也是可以自己重发的。
若生产者那边的confirm机制未开启的情况下,哪怕是你给rabbitmq开启了持久化机制,也有一种可能,就是这个消息写到了rabbitmq中,但是还没来得及持久化到磁盘上,结果不巧,此时rabbitmq挂了,就会导致内存里的一点点数据会丢失。
3.消费端弄丢了数据
为了保证消息从队列可靠地达到消费者,RabbitMQ 提供了消息确认机制(message acknowledgement)。消费者在订阅队列时,可以指定 autoAck 参数,当 autoAck 等于 false时,RabbitMQ 会等待消费者显式地回复确认信号后才从内存(或者磁盘)中移去消息(实质上是先打上删除标记,之后再删除)。当 autoAck 等于 true 时,RabbitMQ 会自动把发送出去的消息置为确认,然后从内存(或者磁盘)中删除,而不管消费者是否真正地消费到了这些消息。
采用消息确认机制后,只要设置 autoAck 参数为 false,消费者就有足够的时间处理消息(任务),不用担心处理消息过程中消费者进程挂掉后消息丢失的问题,因为 RabbitMQ 会一直等待持有消息直到消费者显式调用 Basic.Ack 命令为止。
总结:RabbitMQ消息可靠性保障策略
1、生产者开启消息确认机制
2、消息队列数据持久化
3、消费者手动ack
4、生产者消息记录+定期补偿机制
5、服务幂等处理
6、消息积压处理等
dSend(“”, “x-dead-ttl”, msg+“-”+System.currentTimeMillis());
return “OK”;
}
// 消息的有效期
@GetMapping("/send2/{msg}")
public String send2(@PathVariable String msg) {// 消息对象Message message = new Message((msg + "-" + System.currentTimeMillis()).getBytes());// 设置消息的有效期message.getMessageProperties().setExpiration("25000");rabbitTemplate.convertAndSend("x-dead-ttl", message);return "ok";
}
}
实现消息消费```java
package com.dc.mptest01.listener;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;import java.util.HashMap;@Component
@Slf4j
public class DelayMsgListener {@RabbitListener(queues = "x-dead-msg")public void handler(String msg) {log.info("延迟时间:{}, {}", msg, System.currentTimeMillis());}
}
面试总结
生产者丢失消息
生产者将数据发送到rabbitmq的时候,可能因为网络问题导致数据就在半路给搞丢了
[外链图片转存中…(img-sRUfGMVd-1691365835666)]
1.使用事务(性能差)
RabbitMQ 客户端中与事务机制相关的方法有三个: channel.txSelect 、channel.txCommit 和 channel.txRollback。channel.txSelect 用于将当前的信道设置成事务模式,channel.txCommit 用于提交事务,channel.txRollback 用于事务回滚。在通过 channel.txSelect 方法开启事务之后,我们便可以发布消息给 RabbitMQ 了,如果事务提交成功,则消息一定到达了 RabbitMQ 中,如果在事务提交执行之前由于 RabbitMQ异常崩溃或者其他原因抛出异常,这个时候我们便可以将其捕获,进而通过执行channel.txRollback 方法来实现事务回滚。注意这里的 RabbitMQ 中的事务机制与大多数数据库中的事务概念并不相同,需要注意区分。
事务确实能够解决消息发送方和 RabbitMQ 之间消息确认的问题,只有消息成功被RabbitMQ 接收,事务才能提交成功,否则便可在捕获异常之后进行事务回滚,与此同时可以进行消息重发。但是使用事务机制会“吸干”RabbitMQ 的性能。
报文
2.发送回执确认(推荐)
生产者将信道设置成 confirm(确认)模式,一旦信道进入 confirm 模式,所有在该信道上面发布的消息都会被指派一个唯一的 ID(从 1 开始),一旦消息被投递到所有匹配的队列之后,RabbitMQ 就会发送一个确认(Basic.Ack)给生产者(包含消息的唯一 ID),这就使得生产者知晓消息已经正确到达了目的地了。如果消息和队列是可持久化的,那么确认消息会在消息写入磁盘之后发出。RabbitMQ 回传给生产者的确认消息中的 deliveryTag 包含了确认消息的序号,此外 RabbitMQ 也可以设置 channel.basicAck 方法中的 multiple 参数,表示到这个序号之前的所有消息都已经得到了处理,注意辨别这里的确认和消费时候的确认之间的异同。
2.RabbitMQ弄丢了数据
为了防止rabbitmq自己弄丢了数据,这个你必须开启rabbitmq的持久化,就是消息写入之后会持久化到磁盘,哪怕是rabbitmq自己挂了,恢复之后会自动读取之前存储的数据,一般数据不会丢。除非极其罕见的是,rabbitmq还没持久化,自己就挂了,可能导致少量数据会丢失的,但是这个概率较小。
设置持久化有两个步骤,第一个是创建queue的时候将其设置为持久化的,这样就可以保证rabbitmq持久化queue的元数据,但是不会持久化queue里的数据;第二个是发送消息的时候将消息的deliveryMode设置为2,就是将消息设置为持久化的,此时rabbitmq就会将消息持久化到磁盘上去。必须要同时设置这两个持久化才行,rabbitmq哪怕是挂了,再次重启,也会从磁盘上重启恢复queue,恢复这个queue里的数据。
而且持久化可以跟生产者那边的confirm机制配合起来,只有消息被持久化到磁盘之后,才会通知生产者ack了,所以哪怕是在持久化到磁盘之前,rabbitmq挂了,数据丢了,生产者收不到ack,你也是可以自己重发的。
若生产者那边的confirm机制未开启的情况下,哪怕是你给rabbitmq开启了持久化机制,也有一种可能,就是这个消息写到了rabbitmq中,但是还没来得及持久化到磁盘上,结果不巧,此时rabbitmq挂了,就会导致内存里的一点点数据会丢失。
3.消费端弄丢了数据
为了保证消息从队列可靠地达到消费者,RabbitMQ 提供了消息确认机制(message acknowledgement)。消费者在订阅队列时,可以指定 autoAck 参数,当 autoAck 等于 false时,RabbitMQ 会等待消费者显式地回复确认信号后才从内存(或者磁盘)中移去消息(实质上是先打上删除标记,之后再删除)。当 autoAck 等于 true 时,RabbitMQ 会自动把发送出去的消息置为确认,然后从内存(或者磁盘)中删除,而不管消费者是否真正地消费到了这些消息。
采用消息确认机制后,只要设置 autoAck 参数为 false,消费者就有足够的时间处理消息(任务),不用担心处理消息过程中消费者进程挂掉后消息丢失的问题,因为 RabbitMQ 会一直等待持有消息直到消费者显式调用 Basic.Ack 命令为止。
总结:RabbitMQ消息可靠性保障策略
1、生产者开启消息确认机制
2、消息队列数据持久化
3、消费者手动ack
4、生产者消息记录+定期补偿机制
5、服务幂等处理
6、消息积压处理等