- 思路
- 手写断路器
- 1.导入lombok依赖、Web依赖
- 2.创建一个Controller,进行远程调用（无法访问的地址）
- 3.创建状态类，在pom中添加AOP依赖，创建切面类和Anno注解
- 3.在Controller中添加切点注解，在切面中将需要的方法加入到切点中
- 4.创建一个断路器模型
- 5.在切面中补全方法体
- 测试
- 优化
- Hystrix常用配置
启动类中给RestTemplate加入Bean容器
@SpringBootApplication
public class MyHistrixApplication {
public static void main(String[] args) {
SpringApplication.run(MyHistrixApplication.class, args);
}
@Bean
public RestTemplate restTemplate(){
return new RestTemplate();
}
}
@Autowired
private RestTemplate restTemplate;
@GetMapping("doRPC")
public String doRPC(){
String result = restTemplate.getForObject("http://localhost:8888/login",String.class);
return result;
}
3.创建状态类，在pom中添加AOP依赖，创建切面类和Anno注解
public enum Status {
CLOSE,
OPEN,
HALF_OPEN
}
pom中导入依赖
org.springframework.boot spring-boot-starter-aop
创建Anno注解
@Target(ElementType.METHOD) //作用在方法
@Retention(RetentionPolicy.RUNTIME) //运行时有效
@Documented
@Inherited
public @interface MyAnno {
}
3.在Controller中添加切点注解，在切面中将需要的方法加入到切点中
在Controller中添加切点注解@MyAnno
在切面中将方法以注解的形式添加到切点注解中
@Around(value = "@annotation(com.dcits.myhystrix.anno.MyAnno)")
public Object myAround(ProceedingJoinPoint joinPoint) {}
左边出现一个圈说明已经添加
切面的方法作用：
4.创建一个断路器模型
package com.dcits.myhystrix.pojo;
import lombok.Data;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@Data
public class Hystrix {
//最大失败次数
public static final Integer MAX_FAIL_COUNT = 3;
//窗口时间
public static final Integer WINDOW_TIME = 20;
//断路器中有它自己的状态
private Status status = Status.CLOSE;
private AtomicInteger currentFailkCount = new AtomicInteger(0);
private ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
4,
8,
30,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(2000),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
{
poolExecutor.execute(()->{
while (true){
try {
TimeUnit.SECONDS.sleep(WINDOW_TIME);
}catch (InterruptedException e){
e.printStackTrace();
}
}
});
}
public void addFailCount() {
int i = currentFailkCount.incrementAndGet(); // ++i
if (i >= MAX_FAIL_COUNT){
//说明失败次数已到了阈值
//修改当前状态为open
this.setStatus(Status.OPEN);
//当断路器打开以后就不能去访问 需要将它变成半开
//等待一个时间窗口 让断路器变成半开
poolExecutor.execute(()->{
try {
TimeUnit.SECONDS.sleep(WINDOW_TIME);
}catch (InterruptedException e){
e.printStackTrace();
}
this.setStatus(Status.HALF_OPEN);
//重置失败次数
this.currentFailkCount.set(0);
});
}
}
}
5.在切面中补全方法体
package com.dcits.myhystrix.aspect;
import com.dcits.myhystrix.pojo.Hystrix;
import com.dcits.myhystrix.pojo.Status;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
@Component
@Aspect
public class MyAspect {
//因为一个消费者可以调用多个提供者，所以每个提供者都有自己的断路器
//在消费者里面去创建一个断路器的容器
public static Map hmap = new HashMap<>();
static {
//假设需要去调用order-service的服务
hmap.put("order-service",new Hystrix());
}
Random random = new Random();
//不推荐这么写，建议使用Anno
// public static final String POINT_CUT = "execution(* com.dcits.myhystrix.controller.HystrixController.doRPC(..))";
@Around(value = "@annotation(com.dcits.myhystrix.anno.MyAnno)")
public Object myAround(ProceedingJoinPoint joinPoint) {
Object result = null;
//获取当前提供者的断路器
Hystrix hystrix = hmap.get("order-service");
Status status = hystrix.getStatus();
switch (status) {
case OPEN:
return "我是备胎";
case HALF_OPEN:
int i = random.nextInt(5);
System.out.println(i);
if (i == 1) {
try {
result = joinPoint.proceed();
hystrix.setStatus(Status.CLOSE);
} catch (Throwable throwable) {
return "我是备胎";
}
}
case CLOSE:
//正常 去调用 执行目标方法
try {
result = joinPoint.proceed();
return result;
} catch (Throwable throwable) {
//调用失败
hystrix.addFailCount();
return "我是备胎";
}
default:
return "我是备胎";
}
}
}
测试
优化
对断路器模型代码进行优化
private Object lock = new Object();
{
poolExecutor.execute(()->{
while (true){
try {
TimeUnit.SECONDS.sleep(WINDOW_TIME);
}catch (InterruptedException e){
e.printStackTrace();
}
if(this.status.equals(Status.CLOSE)){
//清零
this.currentFailkCount.set(0);
}else {
//半开或者开 不需要记录次数 这个线程可以不工作
synchronized (lock){
try {
lock.wait();
}catch (InterruptedException e){
e.printStackTrace();
}
}
}
}
});
}
修改切面处代码
case HALF_OPEN:
int i = random.nextInt(5);
System.out.println(i);
if (i == 1) {
try {
result = joinPoint.proceed();
hystrix.setStatus(Status.CLOSE);
synchronized (hystrix.getLock()){
hystrix.getLock().notifyAll();
}
} catch (Throwable throwable) {
return "我是备胎";
}
}
Hystrix常用配置
hystrix: #hystrix 的全局控制
command:
default: #default 是全局控制，也可以换成单个方法控制，把 default 换成方法名即可
fallback:
isolation:
semaphore:
maxConcurrentRequests: 1000 #信号量隔离级别最大并发数
circuitBreaker:
enabled: true #开启断路器
requestVolumeThreshold: 3 #失败次数（阀值）
sleepWindowInMilliseconds: 20000 #窗口时间
errorThresholdPercentage: 60 #失败率
execution:
isolation:
Strategy: thread #隔离方式 thread 线程隔离集合和 SEMAPHORE 信号量隔离
thread:
timeoutInMilliseconds: 3000 #调用超时时长
#隔离方式 两种隔离方式 thread 线程池 按照 group（10 个线程）划分服务提供者，用户请求的线程 和做远程的线程不一样 # 好处 当 B 服务调用失败了 或者请求 B 服务的量太大了 不会对 C 服务造成影响 用户访问比较大的情 况下使用比较好 异步的方式 # 缺点 线程间切换开销大，对机器性能影响 # 应用场景 调用第三方服务 并发量大的情况下 # SEMAPHORE 信号量隔离 每次请进来 有一个原子计数器 做请求次数的++ 当请求完成以后 -- # 好处 对 cpu 开销小 # 缺点 并发请求不易太多 当请求过多 就会拒绝请求 做一个保护机制 # 场景 使用内部调用 ，并发小的情况下 # 源码入门 HystrixCommand AbstractCommand HystrixThreadPool
