hystrix初探

Posted 2021-04-30 玩物得志技术

服务雪崩

在大中型分布式系统中，通常系统很多依赖，最典型的商品详情依赖很多下游的业务服务，如下图:

如下图：当依赖 I （如评价系统）出现不可用，但是其他依赖仍然可用.

当依赖I 阻塞时,大多数服务器的线程池就出现阻塞(BLOCK),影响整个线上服务的稳定性.如下图: 

设计原则:

• 资源隔离

• 熔断器

• 命令模式

隔离

在复杂的分布式架构的应用程序有很多的依赖，都会不可避免地在某些时候失败。高并发的依赖失败时如果没有隔离措施，当前应用服务就有被拖垮的风险。

不管是线程池／信号量的隔离方式，hystrix command本质上都是在子线程中运行的

熔断

熔断基于的数据存储在滑动窗口，1.4之前hystrix使用自己实现的滑动窗口，1.5及1.5之后版本使用rxjava的滑动窗口实现；
滑动窗口核心的两个概念的窗口时长和窗口滚动时长,根据这两个参数窗口会被分为n个BUCKET 

完整链路

使用

public class CommandHello extends HystrixCommand<String> {

    private final String name;

    public CommandHelloFailure(String name) {
        super(HystrixCommandGroupKey.Factory.asKey("HelloGroup"));
        this.name = name;
    }

    //正常逻辑
    @Override
    protected String run() {
        throw new RuntimeException("Hello World");
    }
    
    //降级
    @Override
    protected String getFallback() {
        return "Failure " + name + "!";
    }
}

hystrix调用概览

调用hystrix的command有4种执行方式 

• queue() 异步方法，返回future对象

• execute() 阻塞方法，只是对queue()方法的简单包装

• observe() 返回rxjava中的Observable对象,本质上是阻塞调用

• toObservable() 相当于observe()懒执行版本，在subscribe时才会真正调用

hystrix底层大量使用rxjava响应式编程库 

以queue()方法为例

final Observable<R> o = toObservable();  

toObservable()方法返回一个通过subscribe异步执行的的Observable对象，我们看到下面的逻辑无非是f.isDone()和f.get()这些通用逻辑和一些异常处理，因为核心的逻辑都在toObservable()中，我们一起进去看下

/* 尝试先从缓存中获取 */
if (requestCacheEnabled) {//请求缓存是否启用
    Observable<R> fromCache = requestCache.get(getCacheKey());//getCacheKey方法可以重载
    if (fromCache != null) {
        //开始记录请求相关的数据，以event放送发出去，通知到底层的状态收集器
        ...
        return new CachedObservableResponse<R>((CachedObservableOriginal<R>) fromCache, this);
    }
}

上面缓存命中，直接返回，否则进入正常的调用逻辑

//创建一个Observable，在subscribe时才会真正调用
Observable<R> o = Observable.create(new OnSubscribe<R>() {

    @Override
    public void call(Subscriber<? super R> observer) {
        //异步记录请求日志
        recordExecutedCommand();

        // 如果在钩子里面抛出异常，则默认会以快速失败（不走fallback）方式处理
        executionHook.onStart(_this);

        //熔断器是否允许执行请求
        if (circuitBreaker.allowRequest()) {
                //这里主要针对以信号量方式隔离的方式，如果是线程池方式，tryAcquire永远为true
            final TryableSemaphore executionSemaphore = getExecutionSemaphore();
            if (executionSemaphore.tryAcquire()) {
                try {
                    getRunObservableDecoratedForMetricsAndErrorHandling()
                            ....
                    } catch (RuntimeException e) {
                        observer.onError(e);
                }
            } else {
            //获取信号量失败
                Exception semaphoreRejectionException = new RuntimeException("could not acquire a semaphore for execution");
                ...
            }
        } else {
            //熔断执行失败，进入fallback逻辑，如果没有定义fallback直接抛出异常
            eventNotifier.markEvent(HystrixEventType.SHORT_CIRCUITED, commandKey);
            ...
        }
    }
});        

下面嵌套的逻辑太多太深，就不在有限的篇幅里赘述了

大家可以进入getRunObservableDecoratedForMetricsAndErrorHandling()方法继续跟踪下去，发现最终会走到HystrixCommand#getExecutionObservable方法，在这里调用我们自定义的run()方法。 

参考文献

https://github.com/Netflix/Hystrix/wiki