手写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 {

    @Autowired
    private 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 {
    @Autowired
    private 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;
    }
}