Hystrix失败处理逻辑解析

￡神魔★判官ぃ 2023-06-12 06:30 141阅读 0赞

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方法
4.  ######  ######

######  ######

###### 获取Fallback方法 ######

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

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`研究