手写Hystrix基本原理

本文已收录于专栏《中间件合集》

概念说明
- 什么是Hystrix
- Hystrix解决的问题
提供服务
工作流程
代码实现
- HystrixSDK
- - MyHystrixCommand注解
  - MyHystrixProperty注解
  - MyAspect注解解释器
- 发送请求端
- - 引入Hystrix的依赖
  - 调用代码
- 接收请求端
- 执行效果
- - 发送请求端
总结提升

概念说明

什么是Hystrix

Spring Cloud Hystrix 是基于 Netflix 公司的开源组件 Hystrix 实现的，它提供了熔断器功能，能够有效地阻止分布式微服务系统中出现联动故障，以提高微服务系统的弹性。Spring Cloud Hystrix 具有服务降级、服务熔断、线程隔离、请求缓存、请求合并以及实时故障监控等强大功能。

Hystrix解决的问题

在这里插入图片描述
所有服务都需要请求服务T，当B服务有大量的请求，导致服务T所属的服务器CPU太高，无法立即处理请求，导致服务瘫痪。瘫痪之后就会导致所有的请求推挤在服务U然后一直向上堆积。这时候当服务A去访问服务T或者服务U的时候也请求不同。导致整个体统都瘫痪了。我们把这个情况成为灾难性的雪崩效应。Hystrix 就能很好的解决这一问题。

提供服务

Spring Cloud Hystrix提供了一系列强大的功能来处理分布式系统中的延迟和容错问题。下面是对这些功能的详细说明：

「服务降级（Fallback）」：当一个服务发生故障或超时时，Hystrix可以提供一个备用的响应或执行一些预定义的逻辑，从而避免对整个系统的影响。通过定义降级逻辑，例如返回默认值、执行备用方法或从缓存中获取数据，可以保证系统的可用性。
「服务熔断（Circuit Breaker）」：Hystrix可以根据一定的条件自动开启或关闭服务的熔断。当服务调用失败率达到一定阈值时，Hystrix会自动开启熔断，之后的请求将直接返回降级的响应，避免对后端服务的继续调用，从而保护后端服务免受进一步的压力。一段时间后，Hystrix会尝试恢复对服务的调用，如果调用成功，则关闭熔断，否则继续开启熔断。
「线程隔离（Thread Isolation）」：Hystrix通过使用线程池隔离不同的服务调用，可以防止一个服务的故障导致整个系统的故障。每个服务都运行在独立的线程池中，从而提供了更好的资源隔离和保护。这样可以避免由于一个服务的延迟或故障导致其他服务的阻塞。
「请求缓存（Request Caching）」：Hystrix可以缓存相同的请求结果，以避免重复的调用。当多个请求需要调用同一个服务时，Hystrix会先从缓存中查找结果，如果存在则直接返回，而不需要再次调用服务。这样可以减少对后端服务的请求次数，提高系统的性能和吞吐量。
「请求合并（Request Collapsing）」：Hystrix可以将多个相同类型的请求合并为一个批量请求，从而减少网络开销和提高系统的性能。当多个请求需要调用同一个服务时，Hystrix会将这些请求合并为一个批量请求，并一次性发送给后端服务，然后将结果拆分返回给各个请求。这样可以减少网络通信的次数，提高系统的效率。
「实时故障监控（Real-time Monitoring）」：Hystrix提供了实时监控和指标收集功能，可以通过Hystrix Dashboard或Turbine来查看服务的健康状况、请求的响应时间、错误率等指标。这样可以帮助开发人员快速发现和解决问题，保证系统的稳定性和可靠性。

通过使用Spring Cloud Hystrix，我们可以更好地处理分布式系统中的故障和延迟问题，提高系统的可靠性和容错性。

工作流程

当在5秒内A服务向B服务发送了10个请求，并且满足错误率达到50%。那么断路器开启，进行熔断机制，在30秒内不在请求B服务，直接请求我们定义的托底方法。这些参数例如5s、10个请求、错误率都是可以在代码中进行配置的。根据不同的需求填写不同的参数。
在这里插入图片描述

代码实现

HystrixSDK

sdk是每个请求的发送端需要引入的服务，使用sdk中的一些注解来完成具体的服务

MyHystrixCommand注解

package com.example.hystrixdemo.utils;import com.netflix.hystrix.contrib.javanica.annotation.HystrixException;
import com.netflix.hystrix.contrib.javanica.annotation.ObservableExecutionMode;import java.lang.annotation.*;@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited  //表示注解可以被子类继承
public @interface MyHystrixCommand {String groupKey() default "";String commandKey() default "";String threadPoolKey() default "";String fallbackMethod() default "";MyHystrixProperty[] commandProperties() default {};MyHystrixProperty[] threadPoolProperties() default {};Class<? extends Throwable>[] ignoreExceptions() default {};ObservableExecutionMode observableExecutionMode() default ObservableExecutionMode.EAGER;HystrixException[] raiseHystrixExceptions() default {};String defaultFallback() default "";}

MyHystrixProperty注解

package com.example.hystrixdemo.utils;import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface MyHystrixProperty {String name();String value();
}

MyAspect注解解释器

package com.example.hystrixdemo.utils;import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterThrowing;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.stereotype.Component;import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Calendar;
import java.util.Date;/*** @BelongsProject: hystrixDemo* @BelongsPackage: com.example.hystrixdemo.utils* @Author: Wuzilong* @Description: 注解解释器* @CreateTime: 2023-09-21 10:50* @Version: 1.0*/
@Aspect
@Component
public class MyAspect {Integer requestNum = 0;Date newTime = null;Date newDelayTime=null;Boolean isDelayTime = false;@AfterThrowing(pointcut = "execution(* com.example..*.*(..))",throwing = "ex")public Object afterThrowing(JoinPoint joinPoint,Exception ex) throws InvocationTargetException, IllegalAccessException, NoSuchMethodException {try {MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();Method method1 = methodSignature.getMethod();if (method1.isAnnotationPresent(MyHystrixCommand.class)) {MyHystrixCommand annotation = method1.getAnnotation(MyHystrixCommand.class);MyHystrixProperty[] myHystrixProperties = annotation.commandProperties();if (myHystrixProperties.length == 0) {Method method2 = joinPoint.getTarget().getClass().getMethod(annotation.fallbackMethod());return method2.invoke(joinPoint.getTarget());} else {if (newDelayTime != null && new Date().compareTo(newDelayTime)>=0) {isDelayTime=false;newTime = null;newDelayTime=null;}if (isDelayTime) {Method method2 = joinPoint.getTarget().getClass().getMethod(annotation.fallbackMethod());method2.invoke(joinPoint.getTarget());} else {String numValue = null;String timeValue = null;String errorRateValue = null;String delayTimeValue = null;for (MyHystrixProperty property : myHystrixProperties) {if (property.name().equals("requestNum")) {numValue = property.value();} else if (property.name().equals("requestTime")) {timeValue = property.value();} else if (property.name().equals("requestErrorRate")) {errorRateValue = property.value();} else if (property.name().equals("requestDelayTime")) {delayTimeValue = property.value();}}requestNum++;if (newTime==null){// 创建Calendar对象，并设置为当前时间Calendar calendar = Calendar.getInstance();calendar.setTime(new Date());// 将Calendar对象的毫秒数加上任意值（这里假设增加1000毫秒）calendar.add(Calendar.MILLISECOND, Integer.valueOf(timeValue));newTime=calendar.getTime();}if (new Date().compareTo(newTime) >=0){if (requestNum >= Integer.valueOf(numValue)) {double i = Double.valueOf(numValue) / requestNum * 100;if (i >= Integer.valueOf(errorRateValue)) {isDelayTime = true;if (newDelayTime==null){// 创建Calendar对象，并设置为当前时间Calendar calendar = Calendar.getInstance();calendar.setTime(new Date());// 将Calendar对象的毫秒数加上任意值（这里假设增加1000毫秒）calendar.add(Calendar.MILLISECOND, Integer.valueOf(delayTimeValue));newDelayTime=calendar.getTime();}requestNum = 0;Method method2 = joinPoint.getTarget().getClass().getMethod(annotation.fallbackMethod());return method2.invoke(joinPoint.getTarget());}}}}}}} catch (Exception e) {System.out.println("通过异常进入到AOP中，调用了托底方法");}return null;}
}

声明两个注解分别用来设置服务降级和熔断操作的。MyHystrixCommand注解中会指定降级的方法。其中也包括MyHystrixProperty注解来指定熔断操作开启的一些参数，例如请求的时间、请求的次数、请求错误率等等。MyAspect类去识别两个注解，处理对应的业务逻辑。

发送请求端

引入Hystrix的依赖

        <dependency><groupId>com.example</groupId><artifactId>hystrixDemo</artifactId><version>1.0-SNAPSHOT</version></dependency>

调用代码

package com.example.useraservice.service;import com.example.hystrixdemo.utils.MyHystrixCommand;
import com.example.hystrixdemo.utils.MyHystrixProperty;
import com.example.openfeigndemo.config.FeignConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;/*** @BelongsProject: UserAService* @BelongsPackage: com.example.useraservice.service* @Author: Wuzilong* @Description: 描述什么人干什么事儿* @CreateTime: 2023-07-26 08:40* @Version: 1.0*/
@Service
public class UserA {@Autowiredprivate FeignConfig feignConfig;@MyHystrixCommand(fallbackMethod = "errorMsg",commandProperties = {@MyHystrixProperty(name = "requestNum", value = "3"), //请求次数@MyHystrixProperty(name = "requestTime", value = "10000"), //请求的单位时间@MyHystrixProperty(name = "requestErrorRate", value = "50"),  //错误率@MyHystrixProperty(name = "requestDelayTime", value = "20000")   //熔断持续时间})public String userA(){System.out.println("进入A服务的方法了,去访问B服务。");IUserBService iUserBService=(IUserBService)feignConfig.getAgentObject();String returnContext = iUserBService.getServiceBInfo();System.out.println("B服务返回的内容是："+returnContext);return "A服务调用B服务";}public String errorMsg()  {System.out.println("你有异常啦");return "你有异常了";}
}

package com.example.useraservice.controller;import com.example.useraservice.service.UserA;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;/*** @BelongsProject: UserAService* @BelongsPackage: com.example.useraservice.controller* @Author: Wuzilong* @Description: 描述什么人干什么事儿* @CreateTime: 2023-07-26 15:44* @Version: 1.0*/
@RestController
@RequestMapping("/serviceA")
public class UserAController {@Autowiredprivate UserA userA;@RequestMapping(value="serviceAInfo",method= RequestMethod.GET)public void getServiceAInfo(){try{userA.userA();}catch (Exception e){System.out.println("通过异常进入到AOP中，已调用了托底方法");}}
}

接收请求端

import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.net.InetAddress;
import java.net.UnknownHostException;/*** @BelongsProject: ServiceB* @BelongsPackage: com.example.serviceb.Controller* @Author: Wuzilong* @Description: B服务* @CreateTime: 2023-06-07 19:08* @Version: 1.0*/
@RestController
@RequestMapping("/B")
public class ServiceBController {@Value("${server.port}")private String serverPort;@GetMapping("/receiveMessage")public String receiveMessage() throws UnknownHostException {System.out.println("B：我被调用了");//返回的内容是ip地址和端口号return InetAddress.getLocalHost().getHostAddress()+":"+serverPort;}
}