- SpringCloudAlibaba 实战
- 服务发现Nacos
- 服务发现原理
- 负载均衡Ribbon
- 细粒度修改Ribbon负载均衡规则
- 指定Ribbon全局负载均衡规则
- 配置Ribbon支持Nacos权重负载均衡规则
- 开启Ribbon饥饿加载
- Ribbon的配置项包括以下这些，
- 声明式HTTP客户端Fegin
- Feign的组成
- Feign的配置
- 全局配置Feign的日志级别
- Feign支持的配置项
- Feign的多参数请求构造
- Feign性能优化
- 服务容错Sentinel
- 整合Sentinel
- 项目服务中对Sentinel控制台的相关配置项
- 控制台的配置项
- RestTemplate整合Sentinel
- feign同样支持自定义限流/降级的异常处理
- Sentinel规则持久化
- 实现区分来源
- Sentinel配置项总结
- 消息队列RocketMQ
- 下载rocketmq
- 安装rocketmq
- 安装rocketmq控制台
- 生产者发消息
- 消息者消费消息
- RocketMQ事务消息流程图
- SpringCloudStream
- 网关Gateway
- 整合Gateway
- 核心概念
- 路由配置示例
- Gateway内置谓词工厂
- 自定义谓词工厂
- 微服务认证授权方案
- 配置中心
- Nacos数据持久化
- 调用链监控
- Sleuth
- Zipkin
- 代码分析与检查工具
SpringCloudAlibaba 实战
1、新建 SpringBoot 项目
2、整合 SpringCloudAlibaba 之前需先整合 SpringCloud
3、引入 SpringCloud 依赖管理
org.springframework.cloud
spring-cloud-dependencies
${spring.cloud-version}
pom
import
4、引入 SpringCloudAlibaba 依赖管理
com.alibaba.cloud
spring-cloud-alibaba-dependencies
${spring.cloud-alibaba-version}
pom
import
引入后完整的依赖管理，还引入了SpringWeb、MySQL、Lombok
1.8
1.8
Greenwich.SR6
2.1.4.RELEASE
org.springframework.boot
spring-boot-starter-web
mysql
mysql-connector-java
5.1.49
runtime
org.springframework.boot
spring-boot-starter-test
test
tk.mybatis
mapper-spring-boot-starter
4.2.1
org.projectlombok
lombok
1.18.22
provided
org.springframework.cloud
spring-cloud-dependencies
${spring.cloud-version}
pom
import
com.alibaba.cloud
spring-cloud-alibaba-dependencies
${spring.cloud-alibaba-version}
pom
import
SpringBoot、SpringCloud、SpringCloudAlibaba 之前的版本兼容参考版本说明
实战中采用的版本配置
2.1.13.RELEASE
Greenwich.SR6
2.1.4.RELEASE
服务发现Nacos
服务发现原理
1、引入Nacos依赖，版本受依赖管理限制，所以不用指定版本号
com.alibaba.cloud
spring-cloud-starter-alibaba-nacos-discovery
2、写配置
spring:
application:
# 微服务名称
name: user-center
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://localhost:3306/user_center?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false
username: root
password: root
jackson:
date-format: yyyy-MM-dd HH:mm:ss
cloud:
nacos:
# nacos地址
server-addr: localhost:8848
discovery:
# 服务注册
register-enabled: true
# 指定命名空间
namespace: 7a8121f5-219e-4f58-8119-1111d0058b32
# 指定组
group: group1
# 指定集群名称
cluster-name: aaa
# 指定元数据
metadata:
version: v1
3、下载Nacos
项目中用到的是Nacos1.4.3 Windows版本
Nacos1.4.3下载地址
下载完毕后解压，开始配置Nacos，截取自Nacos官网
添加如下配置
spring.datasource.platform=mysql
db.num=1
db.url.0=jdbc:mysql://11.162.196.16:3306/nacos_devtest?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true
db.user=nacos_devtest
db.password=youdontknow
然后通过命令 startup.cmd -m standalone启动nacos
4、启动项目
在Nacos控制台中的服务列表中就可以看到一个服务实例了
接下来再创建一个项目【content-center】，开始服务之间的调用，复制现有的user-center项目，修改端口号、模块名、数据库、服务名称，然后启动项目
nacos服务列表中现在就有了两个服务，下面实现服务之间的通信，在content-center中实现调用user-center中的接口
Spring提供的用于访问Rest服务的客户端RestTemplate，RestTemplate提供了多种便捷访问远程Http服务的方法,能够大大提高客户端的编写效率。
在user-center服务中准备好一个接口
@GetMapping("/{id}")
public User findById(@PathVariable Integer id) {
return userService.findById(id);
}
在content-center服务做以下配置，声明一个RestTemplate Bean
@Configuration
public class ContentCenterConfiguration {
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
接着在content-center中通过RestTemplate调用user-center中的接口，实现服务之间的通信
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class ShareService {
private final ShareMapper shareMapper;
private final RestTemplate restTemplate;
public ShareDTO findById(Integer id) throws RuntimeException {
ShareDTO dto = new ShareDTO();
Share share = shareMapper.selectByPrimaryKey(id);
BeanUtils.copyProperties(share, dto);
// http://locahost:8020是user-center服务的uri
UserDTO userDTO = restTemplate.getForObject("http://locahost:8020/users/{id}", UserDTO.class, share.getUserId());
dto.setWxNickname(userDTO.getWxNickname());
return dto;
}
}
通过DiscoveryClient对以上代码进行优化，深入理解DiscoveryClient
在SpringCloudAlibaba中NacosDiscoveryClient是对DiscoveryClient的实现
并自动注入到SpringIOC容器中
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class ShareService {
private final ShareMapper shareMapper;
private final RestTemplate restTemplate;
// 服务发现
private final DiscoveryClient discoveryClient;
public ShareDTO findById(Integer id) throws RuntimeException {
ShareDTO dto = new ShareDTO();
Share share = shareMapper.selectByPrimaryKey(id);
BeanUtils.copyProperties(share, dto);
// 通过DiscoveryClient从nacos中获取服务实例
List instances = discoveryClient.getInstances("user-center");
UserDTO userDTO = restTemplate.getForObject(instances.get(0).getUri() + "/users/{id}", UserDTO.class, share.getUserId());
dto.setWxNickname(userDTO.getWxNickname());
return dto;
}
}
负载均衡Ribbon
1、引入依赖
在 spring-cloud-starter-alibaba-nacos-discovery依赖了ribbon，无须再次引入
3、写注解
在RestTemplate声明Bean出添加@LoadBalanced注解
@Bean
// ribbon实现负载均衡
@LoadBalanced
public RestTemplate restTemplate() {
return new RestTemplate();
}
默认Ribbon的负载均衡规则是轮询方式
启动多个user-center服务，在nacos中存在两个user-center服务实例
改造ShareService代码
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class ShareService {
private final ShareMapper shareMapper;
private final RestTemplate restTemplate;
// 服务发现
private final DiscoveryClient discoveryClient;
public ShareDTO findById(Integer id) throws RuntimeException {
ShareDTO dto = new ShareDTO();
Share share = shareMapper.selectByPrimaryKey(id);
BeanUtils.copyProperties(share, dto);
// uri替换成服务实例名称
UserDTO userDTO = restTemplate.getForObject("http://user-center/users/{id}", UserDTO.class, share.getUserId());
dto.setWxNickname(userDTO.getWxNickname());
return dto;
}
}
细粒度修改Ribbon负载均衡规则
方式一：Java代码方式
新建配置类，声明规则，该类要避免被Spring扫描到，所以配置类的位置要放在不能被Spring扫描的地方
@Configuration
public class RibbonConfiguration {
@Bean
public IRule ribbonRule() {
return new RandomRule();
}
}
在主配置类中添加@RibbonClient注解，指定name属性和configuration属性，name属性指定该配置为哪个微服务服务，configuration指定配置类
// name属性指定该配置为哪个微服务服务,configuration指定配置类
@RibbonClient(name = "user-center", configuration = RibbonConfiguration.class)
@Configuration
public class ContentCenterConfiguration {
@Bean
// ribbon实现负载均衡
@LoadBalanced
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
方式二：配置文件方式
配置文件方式优先级高于Java代码配置方式
配置文件方式配置Ribbon负载均衡规则，优先级高于Java代码配置方式
服务名称
user-center:
ribbon:
NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule
指定Ribbon全局负载均衡规则
在主配置类中添加@RibbonClients注解，指定defaultConfiguration属性
@RibbonClients(defaultConfiguration = RandomRule.class)
@Configuration
public class ContentCenterConfiguration {
@Bean
// ribbon实现负载均衡
@LoadBalanced
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
配置Ribbon支持Nacos权重负载均衡规则
编写规则
@Slf4j
public class NacosWeightRule extends AbstractLoadBalancerRule {
@Autowired
private NacosDiscoveryProperties nacosDiscoveryProperties;
@Override
public void initWithNiwsConfig(IClientConfig iClientConfig) {
// 读取配置文件并初始化
}
@Override
public Server choose(Object o) {
try {
// ribbon入口
BaseLoadBalancer loadBalancer = (BaseLoadBalancer) this.getLoadBalancer();
log.info("lb={}", loadBalancer);
// 想要请求的微服务的名称
String name = loadBalancer.getName();
NamingService namingService = nacosDiscoveryProperties.namingServiceInstance();
Instance instance = namingService.selectOneHealthyInstance(name);
log.info("instance={}", instance);
return new NacosServer(instance);
} catch (NacosException e) {
e.printStackTrace();
}
return null;
}
}
修改RibbonClients的defaultConfiguration属性为NacosWeightRule.class，全局负载均衡规则就设置为以Nacos权重方式规则
@RibbonClients(defaultConfiguration = NacosWeightRule.class)
在本次实战中使用的SpringCloudAlibaba版本为2.1.4.RELEASE，在这个版本中，官方提供了现成的基于权重的负载均衡规则——NacosRule，已经不用自己再动手写配置类了，同时该负载均衡规则支持同集群内优先调用，同一个集群内的服务优先调用
@RibbonClients(defaultConfiguration = NacosRule.class)
开启Ribbon饥饿加载
ribbon:
eager-load:
# Ribbon默认懒加载，在第一次调用时创建RibbonClient，此配置开启Ribbon饥饿加载
enabled: true
# 配置哪些微服务采用饥饿加载
clients: user-center
Ribbon的配置项包括以下这些，
Java代码配置：方式类似于配置负载均衡规则，参照配置负载均衡规则的方式
配置属性方式
Ribbon扩展——基于元数据的版本控制【http://www.imooc.com/article/288674】
服务不可跨namespace调用
声明式HTTP客户端Fegin
加依赖
org.springframework.cloud
spring-cloud-starter-openfeign
写注解
启动类上添加 @EnableFeignClients注解
写接口
@FeignClient(name = "user-center")
public interface UserCenterFeignClient {
@GetMapping("users/{id}")
UserDTO findById(@PathVariable Integer id);
}
修改ShareService代码
@Autowired
private final UserCenterFeignClient userCenterFeignClient;
public ShareDTO findById(Integer id) throws RuntimeException {
ShareDTO dto = new ShareDTO();
Share share = shareMapper.selectByPrimaryKey(id);
BeanUtils.copyProperties(share, dto);
// 使用Feign实现服务间调用，Feign整合了Ribbon，所以Ribbon的配置仍然有效
UserDTO userDTO = userCenterFeignClient.findById(share.getUserId());
dto.setWxNickname(userDTO.getWxNickname());
return dto;
}
Feign的组成
Feign的配置
Fiegn的日志级别，默认不打印日志
细粒度配置Feign的日志级别
Java代码方式
1、编写配置类
/**
* 这里不要加@Configuration注解了，否则，该类就要挪到@ComponentScan扫描到的包之外
**/
public class UserCenterFeignClientConfiguration {
@Bean
public Logger.Level level() {
return Logger.Level.FULL;
}
}
2、在FeignClient上指定该配置类
@FeignClient(name = "user-center", configuration = UserCenterFeignClientConfiguration.class)
public interface UserCenterFeignClient {}
3、在配置文件添加如下配置
logging:
level:
# 指定该类的日志级别，必须指定为debug否则不打印
com.itmuch.contentcenter.feignclient.UserCenterFeignClient: debug
配置文件方式( 这种方式简单 )
在配置文件中添加以下配置
logging:
level:
# 指定该类的日志级别，必须指定为debug否则不打印
com.itmuch.contentcenter.feignclient.UserCenterFeignClient: debug
feign:
client:
config:
# 想要调用的微服务名称
user-center:
loggerLevel: full
配置完后的效果
2022-07-13 16:45:44.133 DEBUG 13520 --- [nio-8010-exec-1] c.i.c.feignclient.UserCenterFeignClient : [UserCenterFeignClient#findById] ---> GET http://user-center/users/2 HTTP/1.1
2022-07-13 16:45:44.133 DEBUG 13520 --- [nio-8010-exec-1] c.i.c.feignclient.UserCenterFeignClient : [UserCenterFeignClient#findById] ---> END HTTP (0-byte body)
2022-07-13 16:45:44.161 WARN 13520 --- [nio-8010-exec-1] c.alibaba.cloud.nacos.ribbon.NacosRule : A cross-cluster call occurs，name = user-center, clusterName = BJ, instance = [Instance{instanceId='192.168.137.1#8021#DEFAULT#DEFAULT_GROUP@@user-center', ip='192.168.137.1', port=8021, weight=5.0, healthy=true, enabled=true, ephemeral=true, clusterName='DEFAULT', serviceName='DEFAULT_GROUP@@user-center', metadata={preserved.register.source=SPRING_CLOUD}}, Instance{instanceId='192.168.137.1#8020#DEFAULT#DEFAULT_GROUP@@user-center', ip='192.168.137.1', port=8020, weight=1.0, healthy=true, enabled=true, ephemeral=true, clusterName='DEFAULT', serviceName='DEFAULT_GROUP@@user-center', metadata={preserved.register.source=SPRING_CLOUD}}]
2022-07-13 16:45:44.201 DEBUG 13520 --- [nio-8010-exec-1] c.i.c.feignclient.UserCenterFeignClient : [UserCenterFeignClient#findById]
全局配置Feign的日志级别
代码配置
1、写配置类
public class GlobalFeignClientConfiguration {
@Bean
public Logger.Level level() {
return Logger.Level.FULL;
}
}
2、在项目启动类的@EnableFeignClients注解上指定defaultConfiguration属性的配置类
@EnableFeignClients(defaultConfiguration = GlobalFeignClientConfiguration.class)
3、配置文件中的日记级别要设置为debug
logging:
level:
# 该包下的所有类的日志级别都调整为debug
com.itmuch.contentcenter: debug
配置文件配置( 这种方式简单 )
logging:
level:
# 该包下的所有类的日志级别都调整为debug
com.itmuch.contentcenter: debug
feign:
client:
config:
# 全局配置
default:
loggerLevel: full
Feign支持的配置项
代码方式支持的配置项，配置方式和配置日志级别类似
属性方式支持的配置项
Feign的多参数请求构造
在写FeignClient时，对于GET请求传递多个参数，如果参数是封装在一个对象中的，需要在参数前添加@SpringQueryMap注解
@GetMapping("q")
UserDTO query(@SpringQueryMap UserDTO userDTO);
其他的与写SpringMVC接口的编写体验是一致的
Feign性能优化
配置连接池
Feign默认使用UrlConnection发送请求，UrlConnection是没有连接池的
可以选择Apache的HttpClient进行请求发送
方式如下
添加依赖
io.github.openfeign
feign-httpclient
添加配置
feign:
httpclient:
enabled: true
# feign的最大连接数
max-connections: 200
# 单个路径的最大连接数
max-connections-per-route: 50
服务容错Sentinel
服务容错的四种方式：
- 超时：设置请求超时时间
- 限流：限制访问流量QPS
- 仓壁：设置独立的线程池
- 断路器：降级，类似保险丝的作用，达到阈值断路器打开，等待时间窗口过后，断路器变为半开状态并尝试调用，如果调用成功断路器关闭，如果调用失败，断路器打开，等待下一个时间窗口后再重复上述过程。
整合Sentinel
1、添加依赖
com.alibaba.cloud
spring-cloud-starter-alibaba-sentinel
2、下载sentinel控制台
https://github.com/alibaba/Sentinel/releases
下载下来jar后java -jar运行
用户名和密码都为sentinel
3、项目配置文件中指定控制台地址
spring:
cloud:
sentinel:
transport:
# 指定sentinel控制台地址
dashboard: localhost:8080
启动项目后访问一下接口就可以在sentinel控制台中看到服务名称了
流控
- 直接
读某个资源直接进行限流 - 关联
- 链路
降级
- RT
当1s内持续进入5个请求，对应时刻的平均响应时间(秒级）均超过阈值（count，以ms为单位)，那么在接下的时间窗口（以s为单位)之内，对这个方法的调用都会自动地熔断
- 异常比例
秒级别的异常比例
- 异常数
分钟级别的异常数
热点
可以对指定的参数限流
对限流的资源添加@SentinelResource注解
@GetMapping("hot")
@SentinelResource("hot")
public String hot(@RequestParam(required = false) String a, @RequestParam(required = false) String b) {
return a + "-" + b;
}
选择资源添加热点规则
指定参数索引，可对指定参数值做限流，限制的参数类型必须是基本数据类型或字符串
授权
限制服务消费者对资源的访问
项目服务中对Sentinel控制台的相关配置项
控制台的配置项
SentinelAPI
- Sph
- Tracer
- ContextUtil
@SentinelResource注解详解
RestTemplate整合Sentinel
在创建RestTemplate的方法上加@SentinelRestTemplate注解
@Bean
// ribbon实现负载均衡
@LoadBalanced
// 整合sentinel
@SentinelRestTemplate
public RestTemplate restTemplate() {
return new RestTemplate();
}
写代码测试
@GetMapping("test-sentinel-rest-template/{userId}")
public UserDTO testSentinelRestTemplate(@PathVariable Integer userId) {
return restTemplate.getForObject("http://user-center/users/{id}", UserDTO.class, userId);
}
sentinel控制台多出了链路资源
@SentinelRestTemplate可自定义限流/降级异常处理，方式同@SentinelSource注解
通过配置文件可关闭@SentinelRestTemplate
resttemplate:
sentinel:
# 关闭@SentinelREstTemplate注解
enabled: false
Feign整合Sentinel
添加配置
feign:
sentinel:
# feign整合sentinel
enabled: true
整合好后重启项目，sentinel控制台中可以看到链路资源了
feign同样支持自定义限流/降级的异常处理
方式一
新建一个类，实现对应的FeignClient接口，该类要作为一个组件，如下
@Slf4j
@Component
public class UserCenterFeignClientFallback implements UserCenterFeignClient {
@Override
public UserDTO findById(Integer id) {
UserDTO dto = new UserDTO();
log.info("被限流/降级了");
dto.setWxNickname("一个默认用户");
return dto;
}
}
在FeignClient注解上将fallback属性指定为UserCenterFeignClientFallback
@FeignClient(
name = "user-center",
fallback = UserCenterFeignClientFallback.class
)
public interface UserCenterFeignClient {
@GetMapping("users/{id}")
UserDTO findById(@PathVariable Integer id);
}
这种方式无法获取异常信息，所以还有第二种方式
方式二
新建一个类，实现FallbackFactory接口，该类要作为一个组件
@Slf4j
@Component
public class UserCenterFeignClientFallbackFactory implements FallbackFactory {
@Override
public UserCenterFeignClient create(Throwable throwable) {
return new UserCenterFeignClient() {
@Override
public UserDTO findById(Integer id) {
UserDTO dto = new UserDTO();
log.warn("被限流/降级了",throwable);
dto.setWxNickname("一个默认用户");
}
};
}
}
在FeignClient注解上将fallbackFactory属性指定为UserCenterFeignClientFallbackFactory
FeignClient注解的fallback属性和fallbackFactory二者只能留其一，不能同时配置
@FeignClient(
name = "user-center",
// fallback = UserCenterFeignClientFallback.class
fallbackFactory = UserCenterFeignClientFallbackFactory.class
)
public interface UserCenterFeignClient {
@GetMapping("users/{id}")
UserDTO findById(@PathVariable Integer id);
}
Sentinel规则持久化
推模式
拉模式
错误页优化
在Sentinel1.8之前实现UrlBlockHandler，sentinel1.8之后该类不存在了，替代他的是BlockExceptionHandler
此实战中使用的是1.8版本，所以要是实现的是BlockExceptionHandler
新建一个类，实现BlockExceptionHandler接口，实现方法，添加@Component注解，根据具体异常用于区分是限流还是降级还是其他
@Component
public class MyUrlBlockHandler implements BlockExceptionHandler {
@Override
public void handle(HttpServletRequest request, HttpServletResponse response, BlockException e) throws Exception {
ErrorMsg msg = null;
if (e instanceof FlowException) {
msg = ErrorMsg.builder().status(100).msg("被限流了").build();
} else if (e instanceof DegradeException) {
msg = ErrorMsg.builder().status(101).msg("被降级了").build();
} else if (e instanceof ParamFlowException) {
msg = ErrorMsg.builder().status(102).msg("热点参数限流").build();
} else if (e instanceof SystemBlockException) {
msg = ErrorMsg.builder().status(103).msg("系统规则（负载/...不满足要求）").build();
} else if (e instanceof AuthorityException) {
msg = ErrorMsg.builder().status(104).msg("授权规则不通过").build();
}
response.setStatus(500);
response.setCharacterEncoding("utf-8");
response.setHeader("content-type", "application/json;charset=utf-8");
response.setContentType("application/json;charset=utf-8");
new ObjectMapper().writeValue(response.getWriter(), msg);
}
}
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
class ErrorMsg {
private Integer status;
private String msg;
}
效果展示
给【content-center】服务的sheare/{id}接口添加流控规则
疯狂刷新接口
实现区分来源
新建一个类，实现RequestOriginParser接口，实现方法，添加@Component注解，规定每次请求必须携带参数origin，否则抛出异常
@Component
public class MyRequestOriginParser implements RequestOriginParser {
@Override
public String parseOrigin(HttpServletRequest request) {
String origin = request.getParameter("origin");
if (StringUtil.isBlank(origin)) throw new IllegalArgumentException("origin is must");
return origin;
}
}
在请求接口是携带参数origin请求，参数origin其实就表示请求的来源，在Sentinel控制台中配置流控规则和授权规则时就指定针对来源进行限流或授权
效果展示
Sentinel配置项总结
消息队列RocketMQ
实战中SpringCloudAlibaba版本是2.1.4.RELEASE，对应的rocketmq版本是4.4.0
下载rocketmq
https://rocketmq.apache.org/dowloading/releases/
安装rocketmq
http://www.imooc.com/article/290089
作者用的虚拟机安装rocketmq，内存比较小，还需要再修改几个配置
修改bin/runbroker.sh
修改bin/runserver.sh
开放端口
firewall-cmd --zone=public --add-port=10911/tcp --permanent
firewall-cmd --zone=public --add-port=10909/tcp --permanent
firewall-cmd --zone=public --add-port=9876/tcp --permanent
systemctl restart firewalld.service
firewall-cmd --reload
安装rocketmq控制台
新版中rocketmq-console迁移到一个独立的仓库
https://github.com/apache/rocketmq-dashboard
修改配置文件
命令打包
mvn clean package -Dmaven.test.skip=true上传至虚拟机运行
nohup java -jar rocketmq-dashboard-1.0.0.jar &
控制台启动成功
生产者发消息
1、引入依赖
org.apache.rocketmq
rocketmq-spring-boot-starter
2.0.2
2.0.2版本对应的是rocketmq4.4.0版本
2、写配置
rocketmq:
nameServer: 192.168.5.128:9876
producer:
group: test-group
3、写代码
@Autowired
private RocketMQTemplate rocketMQTemplate;
public Share auditById(Integer id, ShareAuditDTO auditDTO) {
Share share = shareMapper.selectByPrimaryKey(id);
if (Objects.isNull(share)) {
throw new IllegalArgumentException("非法参数！该分享不存在！");
}
if (!Objects.equals("NOT_YET", share.getAuditStatus())) {
throw new IllegalArgumentException("该分享已审核通过/不通过！");
}
share.setAuditStatus(auditDTO.getAuditStatusEnum().toString());
share.setReason(auditDTO.getReason());
shareMapper.updateByPrimaryKey(share);
// 给发布人加积分,发消息到MQ
rocketMQTemplate.convertAndSend("add-bonus", AddUserBonusDTO.builder().userId(share.getUserId()).bonus(500).build());
return share;
}
消息者消费消息
1、引入依赖
org.apache.rocketmq
rocketmq-spring-boot-starter
2.0.2
2.0.2版本对应的是rocketmq4.4.0版本
2、写配置
rocketmq:
nameServer: 192.168.5.128:9876
3、写代码
@Service
@RocketMQMessageListener(consumerGroup = "consumer-group", topic = "add-bonus")
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class UserAddBonusListener implements RocketMQListener {
private final UserMapper userMapper;
private final BonusEventLogMapper bonusEventLogMapper;
@Override
public void onMessage(AddUserBonusDTO message) {
Integer userId = message.getUserId();
Integer bonus = message.getBonus();
// 获取用户
User user = userMapper.selectByPrimaryKey(userId);
// 加积分
user.setBonus(user.getBonus() + bonus);
userMapper.updateByPrimaryKeySelective(user);
// 保存积分日志
bonusEventLogMapper.insert(BonusEventLog.builder()
.userId(userId)
.value(bonus)
.event("CONTRIBUTE")
.description("投稿加积分")
.createTime(new Date())
.build());
}
}
RocketMQ事务消息流程图
代码演示
改造ShareService代码
public Share auditById(Integer id, ShareAuditDTO auditDTO) {
Share share = shareMapper.selectByPrimaryKey(id);
if (Objects.isNull(share)) {
throw new IllegalArgumentException("非法参数！该分享不存在！");
}
if (!Objects.equals(AuditStatusEnum.NOT_YET.toString(), share.getAuditStatus())) {
throw new IllegalArgumentException("该分享已审核通过/不通过！");
}
if (Objects.equals(AuditStatusEnum.PASS, auditDTO.getAuditStatusEnum())) {
// 审核通过，发送事务消息
String transactionId = UUID.randomUUID().toString();
AddUserBonusDTO message = AddUserBonusDTO.builder().userId(share.getUserId()).bonus(500).build();
// 发送半消息
rocketMQTemplate.sendMessageInTransaction(
"tx-add-bonus-group",
"add-bonus",
MessageBuilder
.withPayload(message)
.setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
.setHeader("share_id", id)
.build(),
auditDTO
);
} else {
auditByIdInDB(id, auditDTO);
}
return share;
}
@Transactional
public Share auditByIdInDB(Integer id, ShareAuditDTO auditDTO) {
Share share = Share.builder()
.auditStatus(auditDTO.getAuditStatusEnum().toString())
.reason(auditDTO.getReason())
.id(id)
.build();
shareMapper.updateByPrimaryKeySelective(share);
return share;
}
@Transactional(rollbackFor = Exception.class)
public Share auditByIdInDBWithLog(Integer id, ShareAuditDTO auditDTO, String transactionId) {
Share share = auditByIdInDB(id, auditDTO);
rocketmqTransactionLogMapper.insert(RocketmqTransactionLog
.builder()
.transactionId(transactionId)
.log("分享审核。。。")
.build());
return share;
}
新建一个类，实现RocketMQLocalTransactionListener接口
@RocketMQTransactionListener(txProducerGroup = "tx-add-bonus-group")
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class AddBonusTransactionListener implements RocketMQLocalTransactionListener {
private final ShareService shareService;
private final RocketmqTransactionLogMapper rocketmqTransactionLogMapper;
/**
* 执行本地事务，发送半消息成功后执行
*
* @param msg
* @param arg
* @return
*/
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
String transactionId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
Integer shareId = Integer.valueOf((String) msg.getHeaders().get("share_id"));
try {
shareService.auditByIdInDBWithLog(shareId, (ShareAuditDTO) arg, transactionId);
// 二次确认提交
return RocketMQLocalTransactionState.COMMIT;
} catch (Exception e) {
// 二次确认回滚
return RocketMQLocalTransactionState.ROLLBACK;
}
}
/**
* 回查本地事务回，一般发生在执行本地事务过程中，服务突然宕机等原因导致二次确认MQ没有收到，MQ会回查本地事务
*
* @param msg
* @return
*/
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
String transactionId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
// 通过日志记录本地事务情况
RocketmqTransactionLog transactionLog = rocketmqTransactionLogMapper.selectOne(RocketmqTransactionLog.builder().transactionId(transactionId).build());
if (Objects.nonNull(transactionLog)) {
return RocketMQLocalTransactionState.COMMIT;
} else {
return RocketMQLocalTransactionState.ROLLBACK;
}
}
}
SpringCloudStream
提供了一套更加通用的操作MQ的通信
SpringCloudStream架构
Destination Binder（目标绑定器）
与消息中间件通信的组件
Destination Bindings（目标绑定）
Binding是连接应用程序跟消息中间件的桥梁，用于消息的消费和生产，由Binder创建
input与output是相对于服务来说消息的走向
整合SpringCloudStream
生产者发送消息
1、引依赖
com.alibaba.cloud
spring-cloud-starter-stream-rocketmq
2、写注解
启动类上加上注解 @EnableBinding(Source.class)
3、写配置
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.137.110:9876
bindings:
output:
# 用来指定topic
destination: stream-test-topic
4、写代码发送消息
@Autowired
private Source source;
@GetMapping("test-stream")
public String testStream() {
source.output().send(MessageBuilder.withPayload("消息体").build());
return "success";
}
消费者消费消息
1、引依赖
不同版本的SpringCloudAlibaba引入时的groupId不同，点进SpringCloudAlibaba依赖版本管理中
com.alibaba.cloud
spring-cloud-starter-stream-rocketmq
2、写注解
启动类上加上注解 @EnableBinding(Sink.class)
3、写配置
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.137.110:9876
bindings:
input:
destination: stream-test-topic
# 如果用的是RocketMQ一定要设置，其他MQ可留空
group: binder-group
4、写代码消费消息
@Slf4j
@Service
public class TestStreamConsumer {
@StreamListener(Sink.INPUT)
public void receive(String messageBody) {
log.info("通过stream收到了消息:messageBody={}", messageBody);
}
}
消息过滤
SpringCloudStream异常处理
自定义消息发送、SpringCloudStream+RocketMQ实现分布式事务
1、新建一个类
public interface AddBonusSource {
String OUTPUT = "add-bonus-source-output";
@Output(OUTPUT)
MessageChannel output();
}
2、修改启动类注解 @EnableBinding({Source.class, AddBonusSource.class})，将 AddBonusSource类加入进去
3、修改配置文件，添加 ===区域配置
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.137.110:9876
bindings:
output:
# 用来指定topic
destination: stream-test-topic
=======================================
# 自己自定义的output值
add-bonus-source-output:
# 用来指定topic
destination: add-bonus
=======================================
这样就可以通过注入 AddBonusSource来发送消息了
接下来通过stream实现发送事务消息
配置文件添加 ===区域配置
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.137.110:9876
===============================================
bindings:
add-bonus-source-output:
producer:
transactional: true
group: tx-add-bonus-group
===============================================
bindings:
output:
# 用来指定topic
destination: stream-test-topic
add-bonus-source-output:
destination: add-bonus
修改ShareService发送消息代码，注入 AddBonusSource， ===为修改部分
private final AddBonusSource addBonusSource;
public Share auditById(Integer id, ShareAuditDTO auditDTO) {
Share share = shareMapper.selectByPrimaryKey(id);
if (Objects.isNull(share)) {
throw new IllegalArgumentException("非法参数！该分享不存在！");
}
if (!Objects.equals(AuditStatusEnum.NOT_YET.toString(), share.getAuditStatus())) {
throw new IllegalArgumentException("该分享已审核通过/不通过！");
}
if (Objects.equals(AuditStatusEnum.PASS, auditDTO.getAuditStatusEnum())) {
// 审核通过，发送事务消息
String transactionId = UUID.randomUUID().toString();
AddUserBonusDTO message = AddUserBonusDTO.builder().userId(share.getUserId()).bonus(1000).build();
// ====================================================================================================
// 用stream模型发送事务消息
addBonusSource.output().send(
MessageBuilder
.withPayload(message)
.setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
// send方法没有多余的参数传递auditDTO，之前通过arg参数传递的auditDTO改为用header传递，header只能传递String
.setHeader("dto", JSON.toJSONString(auditDTO))
.setHeader("share_id", id)
.build());
// ====================================================================================================
} else {
auditByIdInDB(id, auditDTO);
}
return share;
}
修改AddBonusTransactionListener的executeLocalTransaction方法
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
MessageHeaders headers = msg.getHeaders();
String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
Integer shareId = Integer.valueOf((String) headers.get("share_id"));
// 改为从header中获取dto，无法通过arg获取
String dtoString = (String) headers.get("dto");
ShareAuditDTO auditDTO = JSON.parseObject(dtoString, ShareAuditDTO.class);
try {
shareService.auditByIdInDBWithLog(shareId, auditDTO, transactionId);
// 二次确认提交
return RocketMQLocalTransactionState.COMMIT;
} catch (Exception e) {
log.error("发生异常了", e);
// 二次确认回滚
return RocketMQLocalTransactionState.ROLLBACK;
}
}
自定义消息接收
新建一个类
public interface AddBonusSink {
String INPUT = "add-bonus-sink";
@Input(INPUT)
SubscribableChannel input();
}
修改启动类注解 @EnableBinding({Sink.class, AddBonusSink.class}),将 AddBonusSink加入进去
修改配置
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.137.110:9876
bindings:
input:
destination: stream-test-topic
# 如果用的是RocketMQ一定要设置，其他MQ可留空
group: binder-group
=========================================================
add-bonus-sink:
# 指定Topic
destination: add-bonus
# 如果用的是RocketMQ一定要设置，其他MQ可留空
group: consumer-group
=========================================================
新建消费者
@Service
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
public class AddBonusStreamConsumer {
private final UserService userService;
@StreamListener(AddBonusSink.INPUT)
public void receive(AddUserBonusDTO addUserBonusDTO) {
userService.addBonus(addUserBonusDTO);
}
}
SpringCloudStream知识盘点
网关Gateway
整合Gateway
1、新建SpringBoot项目，SpringBoot版本为 2.1.13.RELEASE
引入依赖
1.8
Greenwich.SR6
2.1.4.RELEASE
org.springframework.boot
spring-boot-starter
org.springframework.boot
spring-boot-starter-actuator
com.alibaba.cloud
spring-cloud-starter-alibaba-nacos-discovery
org.springframework.cloud
spring-cloud-starter-gateway
org.springframework.cloud
spring-cloud-dependencies
${spring.cloud-version}
pom
import
com.alibaba.cloud
spring-cloud-alibaba-dependencies
${spring.cloud-alibaba-version}
pom
import
2、写配置
server:
port: 8040
spring:
application:
name: gateway
cloud:
nacos:
discovery:
server-addr: 127.0.0.1:8848
gateway:
discovery:
locator:
enabled: true
management:
endpoints:
web:
exposure:
include: '*'
endpoint:
health:
show-details: always
3、启动项目
访问 http://localhost:8040/content-center/shares/1
请求成功分发到content-center服务
转发规律：访问${GATEWAY_URL}/{微服务X}/ 会转发到 微服务X的/ 路径
核心概念
路由配置示例
Gateway内置谓词工厂
推荐文章 路由谓词工厂详解
自定义谓词工厂
谓词工厂必须以 RoutePredicateFactory为固定后缀
新建 TimeBetweenRoutePredicateFactory谓词工厂，继承AbstractRoutePredicateFactory，指定配置类
@Component
public class TimeBetweenRoutePredicateFactory extends AbstractRoutePredicateFactory {
public TimeBetweenRoutePredicateFactory() {
super(TimeBetweenRoutePredicateFactory.Config.class);
}
/**
* 核心方法
*/
@Override
public Predicate apply(TimeBetweenRoutePredicateFactory.Config config) {
LocalTime start = config.getStart();
LocalTime end = config.getEnd();
return serverWebExchange -> {
LocalTime now = LocalTime.now();
// 在配置时间范围内允许访问
return now.isAfter(start) && now.isBefore(end);
};
}
/**
* 用于指定配置类和配置文件里参数的映射规则
*/
@Override
public List shortcutFieldOrder() {
return Arrays.asList("start", "end");
}
@Data
public static class Config {
private LocalTime start;
private LocalTime end;
}
public static void main(String[] args) {
// Gateway对时间格式的处理
DateTimeFormatter formatter = DateTimeFormatter.ofLocalizedTime(FormatStyle.SHORT);
System.out.println(formatter.format(LocalTime.now().plus(3, ChronoUnit.HOURS)));
}
}
添加配置
spring:
application:
name: gateway
cloud:
nacos:
discovery:
server-addr: 127.0.0.1:8848
gateway:
discovery:
locator:
enabled: true
#====================================================
routes:
- id: time-between
uri: lb://content-center
predicates:
- TimeBetween=上午10:00,下午2:00
#====================================================
微服务认证授权方案
处处安全方案
推荐文章 https://www.cnblogs.com/cjsblog/p/10548022.html
外部无状态、内部有状态
网关认证授权、内部裸奔
内部裸奔改进方案
配置中心
使用Nacos做为配置服务器
整合Nacos配置中心
1、引依赖
com.alibaba.cloud
spring-cloud-starter-alibaba-nacos-config
2、写配置
约定
创建文件bootstrap.yml
spring:
cloud:
nacos:
config:
server-addr: localhost:8848
file-extension: yaml
application:
name: content-center
profiles:
active: dev
配置动态刷新
在类上添加注解@RefreshScope即可
改配置要放在远程配置文件中才生效，放在application.yml或者bootstrap.yml文件中是不生效的
配置共享
方式1:shared-dataids
spring:
cloud:
nacos:
config:
# 共享配置的DataId，多个使用,分隔
# 越靠后，优先级越高；common2.yml > common1.yaml
# .yaml后缀不能少，只支持yaml/properties
shared-dataids: common1.yaml,common2.yaml
# 哪些共享配置支持动态刷新，多个使用,分隔
refreshable-dataids: common1.yaml
server-addr: 127.0.0.1:8848
file-extension: yaml
application:
name: content-center
profiles:
active: dev
方式2：ext-config
spring:
cloud:
nacos:
config:
server-addr: 127.0.0.1:8848
file-extension: yaml
ext-config:
# 需共享的DataId，yaml后缀不能少，只支持yaml/properties
# 越靠后，优先级越高 优先级common2.yaml > common1.yaml
- data-id: common1.yaml
# common1.yaml所在的group
group: DEFAULT_GROUP
# 是否允许刷新，默认false
refresh: true
- data-id: common2.yaml
group: DEFAULT_GROUP
refresh: true
application:
name: content-center
profiles:
active: dev
优先级：shared-dataids < ext-config < 自动
引导上下文
- 连接配置服务器，读取外部配置
- Applicaiton Context的父上下文
bootstrap.yml是引导上下文的配置文件
配置优先级
远程配置专用配置>远程配置通用配置>本地配置
可通过配置修改配置的优先级
Nacos数据持久化
Nacos配置中心实践总结
调用链监控
Sleuth
Sleuth集成微服务上，负责产生监控数据
Sleuth术语
Zipkin
搭建Zipkin Server
整合Zipkin
1、引入依赖
org.springframework.cloud
spring-cloud-starter-zipkin
2、添加配置
spring:
zipkin:
base-url: http://localhost:9411/
# 如果整合后Nacos报错，将该属性设置false表示始终将baseUrl视为URL
discovery-client-enabled: false
sleuth:
sampler:
# 抽取率
probability: 1.0
整合Zipkin后Nacos报错
推荐文章 http://www.imooc.com/article/291578
在本次使用的SpringCloud版本中，整合Zipkin后并没有报错，应该是在新版本中已经被修复
Zipkin持久化
使用elasticsearch存储数据
在使用的zipkin-server-2.12.9进行持久化设置时，要求elasticsearch的版本不能太高，这次用的到是elasticsearch6.8.4
1、首先安装elasticsearch6.8.4
下载地址：https://www.elastic.co/cn/downloads/past-releases/elasticsearch-6-8-4
上传至linux并解压 tar -zxvf elasticsearch-6.8.4
进到config文件夹下修改配置
jvm.options
################################################################
Xms represents the initial size of total heap space
Xmx represents the maximum size of total heap space
虚拟机内存只有1G，修改的小一点
-Xms512m
-Xmx512m
################################################################
elasticsearch.yml中加入以下配置
xpack.ml.enabled: false
network.host: 0.0.0.0
bootstrap.system_call_filter: false
ES启动不能以ROOT用户来进行，所以需要创建一个用户
useradd es
将/usr/local/elasticsearch-6.8.4授权给es用户
chown -R jamysong:jamysong /usr/local/elasticsearch-6.8.4
/etc/security/limits.conf最后加入以下配置
es soft nofile 65536
es hard nofile 65536
进入bin目录下，用es用户启动elasticsearch
su es
./elasticsearch -d
-d 后台运行
2、启动zipkin-server
java -jar zipkin-server-2.12.9-exec.jar --STORAGE_TYPE=elasticsearch --ES_HOSTS=192.168.137.110:9200
代码分析与检查工具
SonarQube
推荐文章 http://www.imooc.com/article/291857
Original: https://www.cnblogs.com/lm66/p/16471117.html
Author: Liming_Code
Title: SpringCloudAlibaba 实战
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/575977/
转载文章受原作者版权保护。转载请注明原作者出处！