Hystrix失败处理逻辑解析

Posted 2021-04-30 Java学习录

在上篇文章中，我们整体介绍了Hystrix的工作流程，知道了Hystrix会在下面四种情况下发生降级：

1. 熔断器打开

2. 线程池/信号量跑满

3. 调用超时

4. 调用失败

本篇文章则介绍一下在发生降级时Hystrix的处理细节，下面的方法异常的处理逻辑：

final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() { @Override public Observable<R> call(Throwable t) { circuitBreaker.markNonSuccess(); Exception e = getExceptionFromThrowable(t); executionResult = executionResult.setExecutionException(e); if (e instanceof RejectedExecutionException) { return handleThreadPoolRejectionViaFallback(e); } else if (t instanceof HystrixTimeoutException) { return handleTimeoutViaFallback(); } else if (t instanceof HystrixBadRequestException) { return handleBadRequestByEmittingError(e); } else { /* * Treat HystrixBadRequestException from ExecutionHook like a plain HystrixBadRequestException. */ if (e instanceof HystrixBadRequestException) { eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey); return Observable.error(e); }
 return handleFailureViaFallback(e); } } };

这里我们拿线程池跑满的逻辑来进行分析

线程池跑满

 private Observable<R> handleThreadPoolRejectionViaFallback(Exception underlying) { eventNotifier.markEvent(HystrixEventType.THREAD_POOL_REJECTED, commandKey); threadPool.markThreadRejection(); // use a fallback instead (or throw exception if not implemented) return getFallbackOrThrowException(this, HystrixEventType.THREAD_POOL_REJECTED, FailureType.REJECTED_THREAD_EXECUTION, "could not be queued for execution", underlying); }

1. 第一行发布了一个线程池拒绝的事件

2. 第二行记录了线程池拒绝的次数

3. 获取Fallback方法

获取Fallback方法

 final HystrixRequestContext requestContext = HystrixRequestContext.getContextForCurrentThread(); long latency = System.currentTimeMillis() - executionResult.getStartTimestamp(); // record the executionResult // do this before executing fallback so it can be queried from within getFallback (see See https://github.com/Netflix/Hystrix/pull/144) executionResult = executionResult.addEvent((int) latency, eventType);
 if (isUnrecoverable(originalException)) { logger.error("Unrecoverable Error for HystrixCommand so will throw HystrixRuntimeException and not apply fallback. ", originalException);
 /* executionHook for all errors */ Exception e = wrapWithOnErrorHook(failureType, originalException); return Observable.error(new HystrixRuntimeException(failureType, this.getClass(), getLogMessagePrefix() + " " + message + " and encountered unrecoverable error.", e, null)); } else { if (isRecoverableError(originalException)) { logger.warn("Recovered from java.lang.Error by serving Hystrix fallback", originalException); }
 if (properties.fallbackEnabled().get()) { /* fallback behavior is permitted so attempt */
 final Action1<Notification<? super R>> setRequestContext = new Action1<Notification<? super R>>() { @Override public void call(Notification<? super R> rNotification) { setRequestContextIfNeeded(requestContext); } };
 final Action1<R> markFallbackEmit = new Action1<R>() { @Override public void call(R r) { if (shouldOutputOnNextEvents()) { executionResult = executionResult.addEvent(HystrixEventType.FALLBACK_EMIT); eventNotifier.markEvent(HystrixEventType.FALLBACK_EMIT, commandKey); } } };
 final Action0 markFallbackCompleted = new Action0() { @Override public void call() { long latency = System.currentTimeMillis() - executionResult.getStartTimestamp(); eventNotifier.markEvent(HystrixEventType.FALLBACK_SUCCESS, commandKey); executionResult = executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_SUCCESS); } };
 final Func1<Throwable, Observable<R>> handleFallbackError = new Func1<Throwable, Observable<R>>() { @Override public Observable<R> call(Throwable t) { /* executionHook for all errors */ Exception e = wrapWithOnErrorHook(failureType, originalException); Exception fe = getExceptionFromThrowable(t);
 long latency = System.currentTimeMillis() - executionResult.getStartTimestamp(); Exception toEmit;
 if (fe instanceof UnsupportedOperationException) { logger.debug("No fallback for HystrixCommand. ", fe); // debug only since we're throwing the exception and someone higher will do something with it eventNotifier.markEvent(HystrixEventType.FALLBACK_MISSING, commandKey); executionResult = executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_MISSING);
 toEmit = new HystrixRuntimeException(failureType, _cmd.getClass(), getLogMessagePrefix() + " " + message + " and no fallback available.", e, fe); } else { logger.debug("HystrixCommand execution " + failureType.name() + " and fallback failed.", fe); eventNotifier.markEvent(HystrixEventType.FALLBACK_FAILURE, commandKey); executionResult = executionResult.addEvent((int) latency, HystrixEventType.FALLBACK_FAILURE);
 toEmit = new HystrixRuntimeException(failureType, _cmd.getClass(), getLogMessagePrefix() + " " + message + " and fallback failed.", e, fe); }
 // NOTE: we're suppressing fallback exception here if (shouldNotBeWrapped(originalException)) { return Observable.error(e); }
 return Observable.error(toEmit); } };
 final TryableSemaphore fallbackSemaphore = getFallbackSemaphore(); final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false); final Action0 singleSemaphoreRelease = new Action0() { @Override public void call() { if (semaphoreHasBeenReleased.compareAndSet(false, true)) { fallbackSemaphore.release(); } } };
 Observable<R> fallbackExecutionChain;
 // acquire a permit if (fallbackSemaphore.tryAcquire()) { try { if (isFallbackUserDefined()) { executionHook.onFallbackStart(this); fallbackExecutionChain = getFallbackObservable(); } else { //same logic as above without the hook invocation fallbackExecutionChain = getFallbackObservable(); } } catch (Throwable ex) { //If hook or user-fallback throws, then use that as the result of the fallback lookup fallbackExecutionChain = Observable.error(ex); }
 return fallbackExecutionChain .doOnEach(setRequestContext) .lift(new FallbackHookApplication(_cmd)) .lift(new DeprecatedOnFallbackHookApplication(_cmd)) .doOnNext(markFallbackEmit) .doOnCompleted(markFallbackCompleted) .onErrorResumeNext(handleFallbackError) .doOnTerminate(singleSemaphoreRelease) .doOnUnsubscribe(singleSemaphoreRelease); } else { return handleFallbackRejectionByEmittingError(); } } else { return handleFallbackDisabledByEmittingError(originalException, failureType, message); } } }

方法比较长，主要做了以下事情：

1. 直接看isUnrecoverable方法，判断异常是否为不可恢复异常，如果不可恢复则直接返回失败

2. 如果是可恢复异常则打印日志

3. 判断是否开启执行回退方法，如果开启进入步骤4

4. 创建开始和完成需要发送的两个事件：FALLBACK_EMIT、FALLBACK_SUCCESS

5. 创建调用回退方法出现异常时的处理逻辑：handleFallbackError，而这种场景发生的异常只有两种情况：

1. UnsupportedOperationException异常：未实现getFallback抽象方法

2. 其他异常

6. 创建释放信号量的Action：singleSemaphoreRelease

7. 获取信号量，如果成功执行回退逻辑，也就是调用用户实现的getFallback方法

8. final protected Observable<R> getFallbackObservable() { return Observable.defer(new Func0<Observable<R>>() { @Override public Observable<R> call() { try { return Observable.just(getFallback()); } catch (Throwable ex) { return Observable.error(ex); } } }); }

对于其他异常类型的处理感兴趣的同学可以继续基于Func1 handleFallback研究