hystrix如何集成在openfeign中使用

Posted 2021-03-23 lay2017

所有文章

https://www.cnblogs.com/lay2017/p/11908715.html

 

正文

HystrixInvocationHandler

hystrix是开源的一个熔断组件，springcloud将其集成并默认与openfeign组合使用。而openfeign又是基于jdk动态代理生成接口的代理对象的，hystrix肯定是集成在feign的接口调用过程当中的。

所以，hystrix的熔断集成到openfeign当中就落到了jdk动态代理的InvocationHandler上。那么hystrix对它的实现又是什么呢？

那么，我们接着跟进HystrixInvocationHandler的invoke方法

public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
    // ...

    HystrixCommand<Object> hystrixCommand =
        new HystrixCommand<Object>(setterMethodMap.get(method)) {
          @Override
          protected Object run() throws Exception {
            try {
              // 获取并调用MethodHandler，MethodHandler封装了Http请求，ribbon也在这里被集成
              return HystrixInvocationHandler.this.dispatch.get(method).invoke(args);
            } catch (Exception e) {
              throw e;
            } catch (Throwable t) {
              throw (Error) t;
            }
          }

          @Override
          protected Object getFallback() {
            // ... fallback降级的处理
          }
        };

    // ...

    return hystrixCommand.execute();
}

invoke的代码很长，删除后显得清晰一点。hystrix由于采用了rxjava，代码逻辑阅读起来有点恶心。

核心正向调用流程依旧是通过Method来获取对应的MethodHandler，MethodHandler负责执行http请求，并返回结果。

那么hystrix在这里干了啥呢？

比较显而易见的是fallback部分，fallback含义很明显了，达到一定的失败条件就会触发降级处理。所以，我们可以自定义fallback实现，保证高可用。

而熔断的核心逻辑就落到了hystrixCommand的execute方法里面

 

hystrixCommand

接下来，跟进hystrixCommand的execute方法

public R execute() {
    try {
        return queue().get();
    } catch (Exception e) {
        throw Exceptions.sneakyThrow(decomposeException(e));
    }
}

queue返回一个future，get方法将获取异步结果，跟进queue

public Future<R> queue() {
    final Future<R> delegate = toObservable().toBlocking().toFuture();
  
    final Future<R> f = new Future<R>() {
        // ... future实现，调用delegate的对应实现
    };

    // ...

    return f;
}

可以看到，核心逻辑落到了toObservable上，跟进它

public Observable<R> toObservable() {
    final AbstractCommand<R> _cmd = this;

    // ...

    final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
        @Override
        public Observable<R> call() {
            if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
                return Observable.never();
            }
            return applyHystrixSemantics(_cmd);
        }
    };

    // ...
}

方法非常长，这里只关注一下applyHystrixSemantics方法

private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
    // ...

    // 是否允许请求
    if (circuitBreaker.allowRequest()) {
        final TryableSemaphore executionSemaphore = getExecutionSemaphore();
      
        // ...

        if (executionSemaphore.tryAcquire()) {
            try {
                executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
                // 执行业务
                return executeCommandAndObserve(_cmd)
                        .doOnError(markExceptionThrown)
                        .doOnTerminate(singleSemaphoreRelease)
                        .doOnUnsubscribe(singleSemaphoreRelease);
            } catch (RuntimeException e) {
                return Observable.error(e);
            }
        } else {
            // 信号量获取失败，走fallback
            return handleSemaphoreRejectionViaFallback();
        }
    } else {
        // 快速熔断，走fallback
        return handleShortCircuitViaFallback();
    }
}

applyHystrixSemantics通过熔断器的allowRequest方法判断是否需要快速失败走fallback，如果允许执行那么又会经过一层信号量的控制，都通过才会走execute。

所以，核心逻辑就落到了allowRequest上，跟进它

@Override
public boolean allowRequest() {
    // 强制开启熔断
    if (properties.circuitBreakerForceOpen().get()) {
        return false;
    }
    // 强制关闭熔断
    if (properties.circuitBreakerForceClosed().get()) {
        isOpen();
        return true;
    }
    // 未开启熔断 或者 允许单个测试
    return !isOpen() || allowSingleTest();
}

hystrix允许强制开启或者关闭熔断，如果不想有请求执行就开启，如果觉得可以忽略所有错误就关闭。

在没有强制开关的情况下，主要就是判断当前熔断是否开启。另外，我们不免注意到这里还有一个allowSingleTest方法。在熔断器开启的情况下，会在一定时间后允许发出一个测试的请求，来判断是否开启熔断器。

我们先进入isOpen方法，看看如果判断当前熔断器的开启

public boolean isOpen() {
    // 开关是开启的，直接返回
    if (circuitOpen.get()) {
        return true;
    }

    // 开关未开启，获取健康统计
    HealthCounts health = metrics.getHealthCounts();

    //  总请求数太小的情况，不开启熔断
    if (health.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
        return false;
    }

    // 总请求数够了，失败率比较小的情况，不开启熔断
    if (health.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
        return false;
    } else {
        // 总请求数和失败率都比较大的时候，设置开关为开启，进行熔断
        if (circuitOpen.compareAndSet(false, true)) {
            circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());
            return true;
        } else {
            return true;
        }
    }
}

总体逻辑就是判断一个失败次数是否达到开启熔断的条件，如果达到那么设置开启的开关。

在熔断一直开启的情况下，偶尔会放过一个测试请求来判断是否关闭。那么我们就跟进allowSingleTest看看

public boolean allowSingleTest() {
    // 获取熔断开启时间，或者上一次的测试时间
    long timeCircuitOpenedOrWasLastTested = circuitOpenedOrLastTestedTime.get();

    // 如果熔断处于开启状态，且当前时间距离熔断开启时间或者上一次执行测试请求时间已经到了
    if (circuitOpen.get() && System.currentTimeMillis() > timeCircuitOpenedOrWasLastTested + properties.circuitBreakerSleepWindowInMilliseconds().get()) {
        // cas控制熔断开启
        if (circuitOpenedOrLastTestedTime.compareAndSet(timeCircuitOpenedOrWasLastTested, System.currentTimeMillis())) {
            return true;
        }
    }
    // 否则不允许
    return false;
}

其实就是在一定时间窗口下释放一个测试请求出去，这里还使用了cas机制来控制一定时间窗口只会释放一个请求出去。

 

总结

到这里，本文就结束了。hystrix其实就是在feign的调用过程插了一脚，通过对请求的成功失败的统计数据来开关是否进行熔断。又在每个时间窗口内发送一个测试请求出去，来判断是否关闭熔断。总得来说还是很清晰实用的。不过最后还是要说一句，rxjava的代码太恶心了，哈哈