二十二.SpringCloud源码剖析-Hystrix降级

小咪咪 2022-10-15 12:56 2阅读 0赞

#### 前言 ####

这篇文章是接上一篇《[SpringCloud源码剖析-Hystrix初始化][SpringCloud_-Hystrix]》，继续完成Hystrix未完成的执行流程

#### HystrixCommand ####

上回说到，Hystrix通过HystrixCommandApsect 环绕通知@Around切到@HystrixCommand注解的方法。然后会创建 HystrixCommand 去执行Hystrix,在创建HystrixCommand的时候会初始化熔断器和线程池。HystrixCommand创建好之后，如果没有配置observable模式，默认会交给CommandExecutor去执行，我们接着上一张继续看Hystrix的执行流程

HystrixCommandAspect\#methodsAnnotatedWithHystrixCommand：

...省略...
     try { 
     		//目标方法是否是采用异步回调，默认false
            if (!metaHolder.isObservable()) { 
                  result = CommandExecutor.execute(invokable, executionType, metaHolder);
              } else { 
                  result = executeObservable(invokable, executionType, metaHolder);
              }
            }

跟下去.HystrixCommandAspect\#methodsAnnotatedWithHystrixCommand

public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException { 
            Validate.notNull(invokable);
            Validate.notNull(metaHolder);
    		//ExecutionType有三种类型，同步执行，异步执行，OBSERVABLE异步回调，默认走SYNCHRONOUS
            switch (executionType) { 
            	//同步
                case SYNCHRONOUS: { 
                	//把 HystrixInvokable 转成 HystrixExecutable 后执行execute
                    return castToExecutable(invokable, executionType).execute();
                }
                //异步执行
                case ASYNCHRONOUS: { 
                    HystrixExecutable executable = castToExecutable(invokable, executionType);
                    if (metaHolder.hasFallbackMethodCommand()
                            && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) { 
                        return new FutureDecorator(executable.queue());
                    }
                    return executable.queue();
                }
                //OBSERVABLE 模式
                case OBSERVABLE: { 
                    HystrixObservable observable = castToObservable(invokable);
                    return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
                }
                default:
                    throw new RuntimeException("unsupported execution type: " + executionType);
            }
        }
    	//转发了一下类型HystrixInvokable 转成 HystrixExecutable
        private static HystrixExecutable castToExecutable(HystrixInvokable invokable, ExecutionType executionType) { 
            if (invokable instanceof HystrixExecutable) { 
                return (HystrixExecutable) invokable;
            }
            throw new RuntimeException("Command should implement " + HystrixExecutable.class.getCanonicalName() + " interface to execute in: " + executionType + " mode");
        }

该方法会根据传入的ExecutionType判断执行的方式，ExecutionType中有三种类型

*  ASYNCHRONOUS：异步执行，
 *  SYNCHRONOUS：同步执行，
 *  OBSERVABLE：异步回调，你可去百度一下Java的Observer,它其实是观察者的实现

默认走同步执行SYNCHRONOUS，先把 HystrixInvokable 转成 HystrixExecutable 后，执行execute 继续跟下去，代码来到：com.netflix.hystrix.HystrixCommand\#execute

public R execute() { 
            try { 
                return queue().get();
            } catch (Exception e) { 
                throw Exceptions.sneakyThrow(decomposeException(e));
            }
        }
    
     public Future<R> queue() { 
            /* * The Future returned by Observable.toBlocking().toFuture() does not implement the * interruption of the execution thread when the "mayInterrupt" flag of Future.cancel(boolean) is set to true; * thus, to comply with the contract of Future, we must wrap around it. */
            final Future<R> delegate = toObservable().toBlocking().toFuture();
        	
           ...省略...
            return f;
        }

execute方法中调用了queue方法，queue方法中用到了观察者模式，在toObservable方法中可以看到hystrix的降级触发流程，在toFuture方法中最终会通过GennericCommand调用CommandAction去执行我们的方法，这里我关注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);
                }
            };
    		
    		return Observable.defer(new Func0<Observable<R>>() { 
                @Override
                public Observable<R> call() { 
                     ...省略...
    				//这里走缓存
                    /* try from cache first */
                    if (requestCacheEnabled) { 
                        HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);
                        if (fromCache != null) { 
                            isResponseFromCache = true;
                            return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
                        }
                    }
    				//【第一步】
                    Observable<R> hystrixObservable =
                            Observable.defer(applyHystrixSemantics)
                                    .map(wrapWithAllOnNextHooks);
                                    ...省略...

Hystrix大量使用到了类似于JS的回调函数式编程，applyHystrixSemantics 就是一个回调，通过 Observable.defer(applyHystrixSemantics) 去执行。applyHystrixSemantics 中又调用了 applyHystrixSemantics(\_cmd);跟进com.netflix.hystrix.AbstractCommand\#applyHystrixSemantics方法

private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) { 
            // mark that we're starting execution on the ExecutionHook
            // if this hook throws an exception, then a fast-fail occurs with no fallback. No state is left inconsistent
            executionHook.onStart(_cmd);
    		//是否允许执行熔断器，方法中判断电路是否打开，方法中会
    		//判断circuitBreakerForceOpen强制打开熔断器 和 circuitBreakerForceClosed强制关闭熔断器
    		
            /* determine if we're allowed to execute */
            if (circuitBreaker.attemptExecution()) { 
            	//信号量
                final TryableSemaphore executionSemaphore = getExecutionSemaphore();
                final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
                //信号量释放的回调
                final Action0 singleSemaphoreRelease = new Action0() { 
                    @Override
                    public void call() { 
                        if (semaphoreHasBeenReleased.compareAndSet(false, true)) { 
                            executionSemaphore.release();
                        }
                    }
                };
    
                final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() { 
                    @Override
                    public void call(Throwable t) { 
                        eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, commandKey);
                    }
                };
    			//尝试获取信号量
                if (executionSemaphore.tryAcquire()) { 
                    try { 
                        /* used to track userThreadExecutionTime */
                        //创建一个执行结果对象
                        executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
                        //执行command和监听
                        return executeCommandAndObserve(_cmd)
                        		//当出现错误
                                .doOnError(markExceptionThrown)
                                .doOnTerminate(singleSemaphoreRelease)
                                .doOnUnsubscribe(singleSemaphoreRelease);
                    } catch (RuntimeException e) { 
                        return Observable.error(e);
                    }
                } else { 
                    return handleSemaphoreRejectionViaFallback();
                }
            } else { 
                return handleShortCircuitViaFallback();
            }
        }

这里确定电路未打开的情况下，获取信号量，然后调用com.netflix.hystrix.AbstractCommand\#executeCommandAndObserve方法，继续跟下去：

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) { 
            final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
    
            ... 省略部分代码...
    		//【重要】这个是处理降级的回调，处理降级的核心流程
            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);
                    }
                }
            };
            
      return execution.doOnNext(markEmits)
                    .doOnCompleted(markOnCompleted)
                    //【第一步】这里执行错误，触发降级回调
                    .onErrorResumeNext(handleFallback)
                    .doOnEach(setRequestContext);

该方法中 Observable 去执行，如果出现错误会触发handleFallback降级回调，handleFallback\#call方法中拿到异常后，进行处理，我们继续跟踪一下：com.netflix.hystrix.AbstractCommand\#handleFailureViaFallback方法

private Observable<R> handleFailureViaFallback(Exception underlying) { 
            /** * All other error handling */
            logger.debug("Error executing HystrixCommand.run(). Proceeding to fallback logic ...", underlying);
    
            // report failure
            eventNotifier.markEvent(HystrixEventType.FAILURE, commandKey);
    
            // record the exception
            executionResult = executionResult.setException(underlying);
    		//获取fallback或者抛出异常
            return getFallbackOrThrowException(this, HystrixEventType.FAILURE, FailureType.COMMAND_EXCEPTION, "failed", underlying);
        }

继续跟进 com.netflix.hystrix.AbstractCommand\#getFallbackOrThrowException

...省略部分代码...
     // 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);
                }

这里通过 getFallbackObservable()方法拿到降级执行链，getFallbackObservable方法最终调用com.netflix.hystrix.contrib.javanica.command.GenericCommand\#getFallback获取降级

@Override
        protected Object getFallback() { 
        //降级方法的CommandAction
            final CommandAction commandAction = getFallbackAction();
            if (commandAction != null) { 
                try { 
                    return process(new Action() { 
                        @Override
                        Object execute() { 
                            MetaHolder metaHolder = commandAction.getMetaHolder();
                            Object[] args = createArgsForFallback(metaHolder, getExecutionException());
                            //执行降级方法
                            return commandAction.executeWithArgs(metaHolder.getFallbackExecutionType(), args);
                        }
                    });
                } catch (Throwable e) { 
                    LOGGER.error(FallbackErrorMessageBuilder.create()
                            .append(commandAction, e).build());
                    throw new FallbackInvocationException(unwrapCause(e));
                }
            } else { 
                return super.getFallback();
            }
        }

最终通过 commandAction.executeWithArgs 执行降级方法，底层使用反射调用降级方法，然后把结果返回。

文章结束，最后想吐槽一下，Hystrix真的是我看过最难看的源码，到处用到回调编程，代码可读性是真的差，看来停更是有原因的呀。。。。。。。最后。。。喜欢的话给个好评吧。

[SpringCloud_-Hystrix]: https://blog.csdn.net/u014494148/article/details/117221635?spm=1001.2014.3001.5501