您当前的位置:首页 > 计算机 > 服务器 > 网络服务

微服务11:熔断、降级的Hystrix实现(附源码)

时间:10-10来源:作者:点击数:

1 介绍

前面的章节,我们学习了微服务中对熔断降级的原理,参考这篇《服务治理:熔断、降级、限流》。了解了固定窗口算法、滑动窗口算法、 漏桶原理和令牌桶原理,本文对Hystrix做进一步的分析。

Hystrix是Netflix开源的一款具备熔断、限流、降级能力的容错系统,设计目的是将应用中的系统访问、多链路服务调用、第三方依赖服务的调用,通过流量资源控制的方式隔离开。

避免了在分布式系中的某个服务故障沿着调用链向上传递,出现整体的服务雪崩,并以此提升系统的稳定性和健壮性。

image

1.1 Hystrix是用来解决哪些问题的?

  • 对所依赖的服务的延迟和故障进行容错(防护+控制)
  • 对服务的故障进行妥善的处理
  • 对延迟过久的请求进行快速失败并迅速恢复,避免队列阻塞
  • 返回默认值或者默认的处理(fallback),实现优雅的降级(如给用户一个友好的提示)
  • 近实时的数据监控与异常告警,及时发现问题并快速止损

1.2 Hystrix如何解决这些问题

  • HystrixCommand、HystrixObservableCommand在单独线程中执行,防止单线依赖,消耗整个服务的资源
  • 服务过载的时候立即断开并快速失败,防止队列阻塞,即线程池或信号量满的时候直接拒绝请求
  • 当超时或者失败时,提供fallback能力避免用户直接面对故障,提供优雅的反馈。
  • 采用隔离技术(流量泳道和断路器模式)来避免单个依赖导致这个链路的雪崩
  • 近实时的数据监控与异常告警,及时发现问题并快速止损(提供监控和预警能力)

2 Hystrix 基础模型

2.1 设计模式:命令模式(Command Pattern)

以往的访问模式,是A链路 与 B链路(A -> B)的直接访问。而命令模式(Command Pattern)的作用则是通过建立命令对象来解耦A、B链路。

在执行过程中,命令对象可以对请求进行排队、记录请求日志、执行故障注入、超时/故障 快速返回等操作,如 A -> Command Work -> B。

2.2 隔离模式:线程池和信号量隔离

在计算机中,线程是系统运行的基本单位,我们可以通过对线程池资源的管理,如异步请求,请求超时断开,请求熔断,来对系统资源进行隔离,当部分类型的资源有限,请求过载时,进行系统保护。

Java程序中,Semaphore(信号量)是用来控制同时访问特定资源的线程数量,通过协调各个线程以保证合理地使用公共资源。也保证了资源竞争的隔离性。

3 Hystrix 工作原理

如下图所示(图片源自官网),Hystrix的工作流程上大概会有如下9个步骤,下文将详细介绍每个流程:

image

3.1 创建命令

创建HystrixCommand 或者 HystrixObservableCommand 命令

3.2 执行命令

执行命令,如图中的,一共有四种方式来执行run()/construct()

  • execute
  • queue
  • observer
  • toObserver

单个实例只能执行一次这4个方法。HystrixObservableCommand没有execute()和queue()。

执行方式 说明 可用对象
execute() 阻塞式同步执行,返回依赖服务的单一返回结果(或者抛出异常) HystrixCommand
queue() 基于Future的异步方式执行,返回依赖服务的单一返回结果(或者抛出异常) HystrixCommand
observe() 基于Rxjava的Observable方式,返回通过Observable表示的依赖服务返回结果,代调用代码先执行(Hot Obserable) HystrixObservableCommand
toObvsevable() 基于Rxjava的Observable方式,返回通过Observable表示的依赖服务返回结果,执行代码等到真正订阅的时候才会执行(cold observable) HystrixObservableCommand
  • execute()
    以同步堵塞方式执行run(),调用execute()后,hystrix会先创建一个新线程运行run(),执行excute时一直出于堵塞状态,直到run()运行完成。
  • queue()
    以异步非堵塞方式执行run()。一调用queue()就直接返回一个Future对象,同时hystrix创建一个新线程运行run(),调用程序通过Future.get()拿到run()的返回结果,而Future.get()是堵塞执行的。
  • observe()
    事件注册前执行run()/construct()。
    • 事件注册前,先调用observe()自动触发执行run()/construct()(如果继承的是HystrixCommand,hystrix将创建新线程非堵塞执行run();如果继承的是HystrixObservableCommand,将以调用程序线程堵塞执行construct())
    • observe()返回结果后,调用程序调用subscribe()完成事件注册,如果run()/construct()执行成功则触发onNext()和onCompleted() 方法,如果执行异常则触发 onError() 方法
  • toObservable()
    事件注册后执行run()/construct()。
    • 事件注册前,调用toObservable()就立即返回 Observable对象
    • 调用subscribe()完成事件注册后自动触发执行run()/construct(),如果run()/construct()执行成功则触发onNext()和onCompleted()方法,如果执行异常则触发onError() 方法。

3.3 是否从缓存获取结果返回?

如果当前命令对象配置了允许从结果缓存中取返回结果,并且在结果缓存中已经缓存了请求结果,则立即通过Observable返回。

3.4 是否启用了熔断器?

判断 circuit-breaker 是否打开。如果3.3步骤没有缓存没有命中,则判断一下当前断路器的断路状态是否打开。如果断路器状态为打开状态,则Hystrix将不会执行此Command命令,直接执行步骤3.8 调用Fallback。

如果断路器状态是关闭,则执行 步骤3.5 检查是否有足够的资源运行 Command命令。

3.5 判断资源(线程池/队列/信号量)是否已满?

如果当前要执行的Command命令 先关连的线程池 和队列(或者信号量)资源已经满了,Hystrix将不会运行 Command命令,直接执行步骤8的Fallback降级处理;如果未满,表示有剩余的资源执行Command命令,则执行步骤 3.6。

3.6 执行 construct() 或者 run()

执行 HystrixObservableCommand.construct() 或者 HystrixCommand.run()。

当经过步骤 3.5 判断,有足够的资源执行Command命令时,本步骤将调用Command命令运行方法。调用HystrixCommand的run方法。按照一下两个条件去判断:

  • 判断请求逻辑是否调用成功,如果调用成功返回调用结果。调用出错(5xx),进入步骤 3.8。
  • 判断调用的依赖逻辑调用是否超时,未超时则直接返回成功结果。超时断开,进入步骤3.8。

3.7 计算熔断器健康情况

对于熔断器的信息会做健康的判断。Hystrix 统计Command命令执行执行过程中的 success count、fail count、reject count 和 timeout count, 并将这些信息记录到断路器(Circuit Breaker)中。

断路器会把上面的统计信息按照时间窗统计下来。并判断什么时候可以将请求熔断,在熔断后和熔断窗口期结束之前,请求都不会被Fallback。熔断窗口期结束后会再次校验,通过后熔断开关会被关闭。

3.8 熔断后的请求执行Fallback

Hystrix会在以下场景出现后,触发Fallback操作:

  • 异常场景:run()方法抛出非HystrixBadRequestException异常。
  • 超时场景:run()方法调用超时。
  • 直接熔断:熔断器开启拦截调,所有的请求都会被拦截。
  • 容量满额:线程池/队列/信号量满量之后

3.9 返回成功结果

Hystrix命令对象执行成功,会直接返回结果或者以Observable形式返回结果。返回的Observable 会执行以下流程返回结果。

Hystrix获取返回结果执行流程如下(图片源自官网):

image

4 Hystrix 实现过程

4.1 引入依赖

pom.xml加上以下依赖。我们使用原生hystrix来做案例介绍。

 <dependency>
            <groupId>com.netflix.hystrix</groupId>
            <artifactId>hystrix-core</artifactId>
            <version>1.5.8</version>
        </dependency>

        <dependency>
            <groupId>com.netflix.hystrix</groupId>
            <artifactId>hystrix-metrics-event-stream</artifactId>
            <version>1.4.10</version>
        </dependency>

4.2 fallBack

fallBack是指当程序符合我们执行熔断降级的条件时候,我们默认执行的路线,可以是一个方法或者一个对象。HystrixCommand中已有,我们只需重写即可,类似

@Override
protected String getFallback() {
    return "当熔断、降级发生时,返回的默认信息";
}

在3.8 节 我们介绍了Hystrix 触发 fallBack的四种条件,下面我们一个个来测试。

4.2.1 程序异常fallBack

除了HystrixBadRequestException,所有程序抛出的异常,都会触发getFallback(),调用程序将获得getFallback()的执行并返回。

/**
 * @author brand
 * @Description: 模拟异常/超时的场景 
 * @Copyright: Copyright (c) 2022
 * @Company: Helenlyn, Inc. All Rights Reserved.
 * @date 2022/1/8 下午5:35
 * @Update Time:
 * @Updater:
 * @Update Comments:
 */
public class HystrixException extends HystrixCommand<String> {
     /**
     * 实现getFallback()后,执行命令时遇到以上4种情况将被fallback接管,不会抛出异常或其他
     * 下面演示的是异常的情况
     */
    private final String name;

    public HystrixException(String name) {
        super(HystrixCommandGroupKey.Factory.asKey("Command Group:fallbackGroup"));
        this.name = name;
    }
	
    @Override
    protected String run() throws Exception {
        /*---------------以下三种情况触发fallback-------------------*/
        // 1.循环+等待,超时fallBack
//    	int i = 0;
//    	while (true) {
//    		i++;
//            Thread.currentThread().sleep(1000);
//    	}

        // 2.除零导致异常
//    	 int i = 1/0;

        // 3.主动抛出异常
        // throw new Exception("command trigger fallback");

        /*---------------直接抛出HystrixBadRequestException,不触发fallback-----------------*/
        // HystrixBadRequestException,这个是非法参数或非系统错误引起,不触发fallback,也不被计入熔断器
        throw new HystrixBadRequestException("HystrixBadRequestException not trigger fallback");

//    return "success";
    }

    @Override
    protected String getFallback() {
        return "fallback: " + name;
    }
}

编写测试类:

/**
 * @author brand
 * @Description: 测试异常/超时 fallBack
 * @Copyright: Copyright (c) 2022
 * @Company: Helenlyn, Inc. All Rights Reserved.
 * @date 2022/1/8 下午5:35
 * @Update Time:
 * @Updater:
 * @Update Comments:
 */
public class ExceptionTimeOutFallBackTest {    
    @Test
    public void testException() throws IOException {
        try {
          assertEquals("success", new HystrixException("Exception").execute());
        } catch(Exception e) {
            System.out.println("run()抛出HystrixBadRequestException时,会被捕获到这里" + e.getCause());
        }
    }
}

测试类执行直接抛出HystrixBadRequestException,测试类会走到catch函数段中。

测试类执行其他三种情况,会得到以下结果:

image

4.2.2 调用超时fallBack

同上 4.2.1 中的 循环+等待,超时fallBack 的场景

4.3 熔断策略

4.3.1 熔断实现的基本原理

image

图片源自官网,这边就不单独画了。

  1. 断路器设置了生效阈值,并且在时间窗内的请求数超过阈值:circuitBreaker.requestVolumeThreshold。
    超过之后触发断路器开启,否则不开启。比如熔断阈值为100,哪怕你99个都fail,也不会触发熔断。
  2. 请求的错误率超过错误率阈值:errorThresholdPercentage,比如20%,10次有2次就达到要求。
  3. 1与2的条件都满足的时候,原来关闭的断路器将开启。
  4. 断路器开启之后,后续请求过来的流量都会被断开。
  5. 断路的那段时间我们叫做休眠时间窗:sleepWindowInMilliseconds。 休眠时间窗过去之后,再发起请求,这时候断路器半开。
    • 请求失败:断路器状态继续保持未开启,并更新休眠时间窗。
    • 请求成功:则断路器状态改为关闭。

4.3.2 断路器配置参数说明

key值 说明 默认值
circuitBreaker.enabled 是否开启断路器 true
circuitBreaker.requestVolumeThreshold 断路器启用请求数阈值 10
circuitBreaker.sleepWindowInMilliseconds 断路器启用后的睡眠时间窗 5000(ms)
circuitBreaker.errorThresholdPercentage 断路器启用失败率阈值 50(%)
circuitBreaker.forceOpen 是否强制将断路器设置成开启状态 false
circuitBreaker.forceClosed 是否强制将断路器设置成关闭状态 false

4.3.3 测试案例

  • 通过withCircuitBreakerRequestVolumeThreshold配置10s(默认时间窗)内请求数超过10个时熔断器开始生效
  • 通过withCircuitBreakerErrorThresholdPercentage配置错误比例>50%时开始熔断
  • 然后for循环执行execute()触发run(),在run()里,如果name是小于30的偶数则正常返回,否则异常
  • 通过多次循环后,异常请求占所有请求的比例将大于50%,就会看到后续请求都不进入run()而是进入getFallback(),因为不再打印"running run():" + name了。
  • 除此之外,hystrix还支持多长时间从熔断状态自动恢复等功能,见下文附录。
/**
 * @author brand
 * @Description: 熔断
 * @Copyright: Copyright (c) 2022
 * @Company: Helenlyn, Inc. All Rights Reserved.
 * @date 2022/1/8 下午3:41
 * @Update Time:
 * @Updater:
 * @Update Comments:
 */
public class HystrixCircuitBreaker extends HystrixCommand<String> {

    private final String name;

    public HystrixCircuitBreaker(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("Group:CircuitBreaker"))
                        .andCommandKey(HystrixCommandKey.Factory.asKey("Command:CircuitBreaker"))
                        .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("ThreadPool:CircuitBreakerTest"))
                        .andThreadPoolPropertiesDefaults(    // 配置线程池
                                HystrixThreadPoolProperties.Setter()
                                        .withCoreSize(200)    // 配置线程池里的线程数,设置足够多线程,以防未熔断却打满threadpool
                        )
                        .andCommandPropertiesDefaults(    // 配置熔断器
                                HystrixCommandProperties.Setter()
                                        .withCircuitBreakerEnabled(true)
                                        .withCircuitBreakerRequestVolumeThreshold(10)
                                        .withCircuitBreakerErrorThresholdPercentage(50)
//                		.withCircuitBreakerForceOpen(true)	// true时强制将断路器设置成开启状态,所有请求都将被拒绝,直接到fallback
//                		.withCircuitBreakerForceClosed(true)	// true时强制将断路器设置成关闭状态,将忽略所有错误
//                		.withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)	// 信号量隔离
//                		.withExecutionTimeoutInMilliseconds(5000)
                        )
        );
        this.name = name;
    }

    @Override
    protected String run() throws Exception {
        System.out.println("running num :" + name);
        int num = Integer.valueOf(name);
        if (num % 2 == 0 && num < 30) {    // 符合条件,直接返回
            return name;
        } else {    // 模拟异常
            int j = 0;
            j = num / j;
        }
        return name;
    }

    @Override
    protected String getFallback() {
        return "CircuitBreaker fallback: " + name;
    }
}

执行结果如下,偶数正常返回,奇数进入熔断信息,并且超过30之后全部进入fallBack

image

4.4 线程池/信号量隔离策略

4.4.1 线程池隔离策略

线程池隔离:不同服务通过使用不同线程池,彼此间将不受影响,达到隔离效果。

我们通过andThreadPoolKey配置使用命名为ThreadPoolTest的线程池,实现与其他命名的线程池天然隔离,如果不配置andThreadPoolKey,也可以则使用withGroupKey配置来命名线程池。

4.4.1.1 线程池未隔离情况
/**
 * @author brand
 * @Description: 线程池隔离
 * @Copyright: Copyright (c) 2022
 * @Company: Helenlyn, Inc. All Rights Reserved.
 * @date 2022/1/8 下午5:58
 * @Update Time:
 * @Updater:
 * @Update Comments:
 */
public class HystrixThreadPool extends HystrixCommand<String> {
    private final String name;
    public HystrixThreadPool(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ThreadPoolTestGroup"))  // CommandGroup分组
                .andCommandKey(HystrixCommandKey.Factory.asKey("testCommandKey"))
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("ThreadPoolTest"))  // 线程池key
                .andCommandPropertiesDefaults(
                        HystrixCommandProperties.Setter()
                                .withExecutionTimeoutInMilliseconds(5000)
                )
                .andThreadPoolPropertiesDefaults(
                        HystrixThreadPoolProperties.Setter()
                                .withCoreSize(3)	// 配置线程池里的线程数为3。超过3次进行熔断
                )
        );
        this.name = name;
    }

    @Override
    protected String run() throws Exception {
        /*---------------如果线程数超配,会触发fallback的case,否则休眠1s,进行正常返回-------------------*/
        TimeUnit.MILLISECONDS.sleep(1000); 
        return name;
    }

    @Override
    protected String getFallback() {
        return "fallback: " + name;
    }
}

测试一下,下面都是使用 HystrixThreadPoolKey 为 ThreadPoolTest的线程池命名,所以是公用,会返回fallBack的结果。

  for(int i = 0; i < 3; i++) {
            try {
                Future<String> future = new HystrixThreadPool("thread pool"+i).queue();  // 以异步非堵塞方式执行run(),所以消耗了3个线程
            } catch(Exception e) {
                System.out.println("run()抛出HystrixBadRequestException时,被捕获到这里" + e.getCause());
            }
        }
        for(int i = 0; i < 10; i++) {
            try {
                System.out.println("===========" + new HystrixThreadPool("thread pool").execute());  //上面消耗了所有线程,这边会执行到fallBack中
            } catch(Exception e) {
                System.out.println("run()抛出HystrixBadRequestException时,被捕获到这里" + e.getCause());
            }
        }
image
4.4.1.2 线程池隔离情况

我们做一下调整,让线程池的key(HystrixThreadPoolKey)不一致,再测试是否返回正常的执行结果。

/**
 * @author brand
 * @Description: 线程池隔离
 * @Copyright: Copyright (c) 2022
 * @Company: Helenlyn, Inc. All Rights Reserved.
 * @date 2022/1/8 下午5:58
 * @Update Time:
 * @Updater:
 * @Update Comments:
 */
public class HystrixThreadPool extends HystrixCommand<String> {
    private final String name;
    public HystrixThreadPool(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ThreadPoolTestGroup"))  // CommandGroup分组
                .andCommandKey(HystrixCommandKey.Factory.asKey("testCommandKey"))
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey(name))  // 线程池key,根据请求的入参来算
                .andCommandPropertiesDefaults(
                        HystrixCommandProperties.Setter()
                                .withExecutionTimeoutInMilliseconds(5000)
                )
                .andThreadPoolPropertiesDefaults(
                        HystrixThreadPoolProperties.Setter()
                                .withCoreSize(3)	// 配置线程池里的线程数为3。超过3次进行熔断
                )
        );
        this.name = name;
    }

    @Override
    protected String run() throws Exception {
        /*---------------如果线程数超配,会触发fallback的case,否则休眠1s,进行正常返回-------------------*/
        TimeUnit.MILLISECONDS.sleep(1000); 
        return name;
    }

    @Override
    protected String getFallback() {
        return "fallback: " + name;
    }
}

测试一下,下面都是使用 HystrixThreadPoolKey 为 ThreadPoolTest的线程池命名,所以是公用,会返回fallBack的结果。

  for(int i = 0; i < 3; i++) {
            try {
                Future<String> future = new HystrixThreadPool("thread pool"+i).queue();  // 会有三个线程池组 thread pool1、thread poo2、thread pool3,不互相影响,更不会影响下面excute()的执行
            } catch(Exception e) {
                System.out.println("run()抛出HystrixBadRequestException时,被捕获到这里" + e.getCause());
            }
        }
        for(int i = 0; i < 10; i++) {
            try {
                System.out.println("===========" + new HystrixThreadPool("thread pool").execute());  //与上面隔离,所以这边执行始终不会走到fallBack中
            } catch(Exception e) {
                System.out.println("run()抛出HystrixBadRequestException时,被捕获到这里" + e.getCause());
            }
        }
image

4.5 代码参考

https://github.com/WengZhiHua/Helenlyn.Grocery/tree/master/parent/HystrixDemo

方便获取更多学习、工作、生活信息请关注本站微信公众号城东书院 微信服务号城东书院 微信订阅号
推荐内容
相关内容
栏目更新
栏目热门
本栏推荐