支持哪些功能
命令模式
使用命令模式，把对远程的调用都封装成command，具体命令包括HystrixCommand和HystrixObservableCommand。
- HystrixCommand：依赖的服务返回单个操作结果。封装了两种执行方法：
- execute:同步阻塞，调用了queue().get()方法。调用execute()后，hystrix先创建一个新线程运行run()，接着调用程序要在execute()调用处一直堵塞着，直到run()运行完成。
- queue:异步非阻塞，它调用了toObservable().toBlocking().toFuture()方法。以异步非堵塞方式执行run()。调用queue()就直接返回一个Future对象，同时hystrix创建一个新线程运行run()，调用程序通过Future.get()拿到run()的返回结果，而Future.get()是堵塞执行的。
- HystrixObservableCommand：依赖的服务返回多个操作结果。也封装了两种执行方法：
- observe：返回的是Hot Observe对象。当调用obeseve()方法，内部其实调用的是toObservable().subscribe(subject)，直接就触发了run方法，如果run()/construct()执行成功则触发onNext()和onCompleted()，如果执行异常则触发onError()。
- toObserve()：返回的是cold observe对象。该方法不会主动创建线程运行run()，只有当调用了toBlocking().single()或subscribe()时，才会去创建线程运行run()。
隔离策略
- Hystrix为每个远程调用都维护一个小线程池（或信号量）;如果它们达到设定值（触发隔离），则发往该依赖项的请求将立即被拒绝，执行失败回退逻辑（Fallback），而不是排队。
- 隔离分为线程池和信号量：
- 线程池
- 每个远程调用都从线程池获取线程来执行，将业务请求和远程调用很好隔离，互不影响
- 在线程池层面可以在错误比例，延迟，超时，拒绝上面做文章
- 请求线程和执行线程不是同一个线程，涉及到了线程的切换，增加性能负担
- 信号量
- 信号量主要是控制的并发数，远程调用和业务请求是同一个线程，所以谨慎使用，但是不涉及线程切换
- 线程池
- 有三个参数可以用来设置隔离单位：
- CommandKey/CommandName：每个command都有对应的commandKey可以认为是command的名字，默认情况下（注解未指定），命令名称来源于切入方法的方法名（method.getName()）。自定义使用HystrixCommandKey.Factory.asKey("HelloWorld")，说实话感觉没啥用，不需要自定义。
- GroupKey：Hystrix使用命令分组将一起的命令进行管理，比如报告、警报、仪表盘或组/库。默认情况下（注解未指定），名称来源于切入方法所属类的类名名（obj.getClass().getSimpleName()），Hystrix使用 HystrixCommandGroupKey 来定义命令线程池，除非单独定义线程池。
- ThreadPoolKey：线程池主要体现是用于监测、指标发布、缓存和其他此类用途的HystrixThreadPool。当隔离方式为线程时，有隔离的作用.具有相同的threadPoolKey的使用同一个线程池，如果注解未指定threadPoolKey，则使用groupKey作为HystrixThreadPoolKey
- 线程池存储在一个静态的ConcurrentHashMap中
观察者模式
- Hystrix通过观察者模式对服务进行状态监听
- 每个远程服务都包含有一个对应的Metrics，所有Metrics都由一个ConcurrentHashMap来进行维护，Key是CommandKey.name()
- 在远程调用的不同阶段会往Metrics中写入不同的信息，Metrics会对统计到的历史信息进行统计汇总，供熔断器以及Dashboard监控时使用
熔断机制
- 熔断机制是一种保护性机制，当系统中某个服务失败率过高时，将开启熔断器，对该服务的后续调用，直接拒绝，进行Fallback操作。
- 熔断所依靠的数据即是Metrics中的HealthCount所统计的错误率。
- 休眠时间过后，Hystrix会将熔断器状态改为半开状态，然后尝试性的执行一次command，如果成功，则关闭熔断器，如果失败，继续打开熔断器，执行新的熔断周期；
- 熔断器打开后，熔断器的健康检查指标会重置，重新开始计算。
- 熔断器可设置的参数：
- 至少有10个请求，熔断器才开始进行错误率的计算
- 熔断器中断请求5秒后会进入半打开状态
- 错误率达到50开启熔断保护
缓存机制
同一对象的不同HystrixCommand实例，只执行一次底层的run()方法，并将第一个响应结果缓存起来，其后的请求都会从缓存返回相同的数据。
HystrixCommandAspect
@EnableHystrix注解开启断路器，该注解会初始化HystrixCommandAspect类。
@Aspect
public class HystrixCommandAspect {
@Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)")
public void hystrixCommandAnnotationPointcut() {
}
// 请求的合并
@Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
public void hystrixCollapserAnnotationPointcut() {
}
@Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
public Object methodsAnnotatedWithHystrixCommand(ProceedingJoinPoint joinPoint) throws Throwable {
Method method = AopUtils.getMethodFromTarget(joinPoint);
Validate.notNull(method, "failed to get method from joinPoint: %s", new Object[]{joinPoint});
if (method.isAnnotationPresent(HystrixCommand.class) && method.isAnnotationPresent(HystrixCollapser.class)) {
throw new IllegalStateException("method cannot be annotated with HystrixCommand and HystrixCollapser annotations at the same time");
} else {
HystrixCommandAspect.MetaHolderFactory metaHolderFactory = (HystrixCommandAspect.MetaHolderFactory)META_HOLDER_FACTORY_MAP.get(HystrixCommandAspect.HystrixPointcutType.of(method));
//如果是@HystrixCommand，最终调用CommandMetaHolderFactory.create创建metaHolder
MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
// 创建HystrixInvokable,只是一个空接口，没有任何方法，只是用来标记具备可执行的能力
HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ? metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();
try {
Object result;
if (!metaHolder.isObservable()) {
// 利用工具CommandExecutor来执行
result = CommandExecutor.execute(invokable, executionType, metaHolder);
} else {
result = this.executeObservable(invokable, executionType, metaHolder);
}
return result;
} catch (HystrixBadRequestException var9) {
throw var9.getCause();
} catch (HystrixRuntimeException var10) {
throw this.getCauseOrDefault(var10, var10);
}
}
}
}
//创建Command
public class HystrixCommandFactory {
private static final HystrixCommandFactory INSTANCE = new HystrixCommandFactory();
private HystrixCommandFactory() {
}
public static HystrixCommandFactory getInstance() {
return INSTANCE;
}
public HystrixInvokable create(MetaHolder metaHolder) {
Object executable;
if (metaHolder.isCollapserAnnotationPresent()) {
executable = new CommandCollapser(metaHolder);
} else if (metaHolder.isObservable()) {//如果切入的方法是Observable类型
executable = new GenericObservableCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder));
} else {//比如：public String hello()方法
executable = new GenericCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder));
}
return (HystrixInvokable)executable;
}
}
//新建GenericCommand：有点熟悉了吧
public class GenericCommand extends AbstractHystrixCommand<Object> {
private static final Logger LOGGER = LoggerFactory.getLogger(GenericCommand.class);
public GenericCommand(HystrixCommandBuilder builder) {
super(builder);
}
// 执行具体的方法，如：serviceName的hello
protected Object run() throws Exception {
LOGGER.debug("execute command: {}", this.getCommandKey().name());
return this.process(new AbstractHystrixCommand<Object>.Action() {
Object execute() {
return GenericCommand.this.getCommandAction().execute(GenericCommand.this.getExecutionType());
}
});
}
// 执行fallback方法，如：serviceName的error()
protected Object getFallback() {
final CommandAction commandAction = this.getFallbackAction();
if (commandAction != null) {
try {
return this.process(new AbstractHystrixCommand<Object>.Action() {
Object execute() {
MetaHolder metaHolder = commandAction.getMetaHolder();
Object[] args = CommonUtils.createArgsForFallback(metaHolder, GenericCommand.this.getExecutionException());
return commandAction.executeWithArgs(metaHolder.getFallbackExecutionType(), args);
}
});
} catch (Throwable var3) {
LOGGER.error(FallbackErrorMessageBuilder.create().append(commandAction, var3).build());
throw new FallbackInvocationException(var3.getCause());
}
} else {
return super.getFallback();
}
}
//实现了HystrixExecutable的方法
public R execute() {
try {
return this.queue().get();
} catch (Exception var2) {
throw Exceptions.sneakyThrow(this.decomposeException(var2));
}
}
}
//HystrixInvokable（GenericCommand）委托CommandExecutor 来执行
public class CommandExecutor {
// 全文的关键方法
public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException {
Validate.notNull(invokable);
Validate.notNull(metaHolder);
switch(executionType) {
case SYNCHRONOUS:
// 首先将 invokable 转换为 HystrixExecutable，再执行 HystrixCommand的execute() 方法
return castToExecutable(invokable, executionType).execute();
case ASYNCHRONOUS://如果切入的目标方法是Future返回类型时
HystrixExecutable executable = castToExecutable(invokable, executionType);
if (metaHolder.hasFallbackMethodCommand() && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) {
return new FutureDecorator(executable.queue());
}
return executable.queue();
case OBSERVABLE:
HystrixObservable observable = castToObservable(invokable);
return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
default:
throw new RuntimeException("unsupported execution type: " + executionType);
}
}
// HystrixExecutable 的 execute() 方法由 HystrixCommand.execute() 实现
private static HystrixExecutable castToExecutable(HystrixInvokable invokable, ExecutionType executionType) {
if (invokable instanceof HystrixExecutable) {
return (HystrixExecutable)invokable;
} else {
throw new RuntimeException("Command should implement " + HystrixExecutable.class.getCanonicalName() + " interface to execute in: " + executionType + " mode");
}
}
}
public interface HystrixExecutable<R> extends HystrixInvokable<R> {
R execute();
Future<R> queue();
Observable<R> observe();
}
- 启动注解初始化一个切面类HystrixCommandAspect，该切面定义个切点会拦截HystrixCommand注解的方法，然后声明一个环绕通知
- 通知中调用HystrixCommandFactory.create创建HystrixInvokable
- HystrixInvokable委托CommandExecutor.execute来执行请求
- 内部通还是过HystrixInvokable.execute()实现。
- HystrixInvokable.execute()通过queue().get()执行
继续看queue()方法
public Future<R> queue() {
//调用的toObservable
final Future<R> delegate = this.toObservable().toBlocking().toFuture();//入口
Future<R> f = new Future<R>() {
public boolean cancel(boolean mayInterruptIfRunning) {
if (delegate.isCancelled()) {
return false;
....省略代码
}
public static <T> Future<T> toFuture(Observable<? extends T> that) {
final CountDownLatch finished = new CountDownLatch(1);
final AtomicReference<T> value = new AtomicReference();
final AtomicReference<Throwable> error = new AtomicReference();
//这里发起订阅，才真正触发前边构建的Observable对象，发射数据
final Subscription s = that.single().subscribe(new Subscriber<T>() {
public void onCompleted() {
finished.countDown();
}
public void onError(Throwable e) {
error.compareAndSet((Object)null, e);
finished.countDown();
}
public void onNext(T v) {
value.set(v);
}
});
return new Future<T>() {
private volatile boolean cancelled;
public boolean cancel(boolean mayInterruptIfRunning) {
if (finished.getCount() > 0L) {
this.cancelled = true;
s.unsubscribe();
finished.countDown();
return true;
} else {
return false;
}
}
public boolean isCancelled() {
return this.cancelled;
}
public boolean isDone() {
return finished.getCount() == 0L;
}
public T get() throws InterruptedException, ExecutionException {
finished.await();
return this.getValue();
}
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
if (finished.await(timeout, unit)) {
return this.getValue();
} else {
throw new TimeoutException("Timed out after " + unit.toMillis(timeout) + "ms waiting for underlying Observable.");
}
}
private T getValue() throws ExecutionException {
Throwable throwable = (Throwable)error.get();
if (throwable != null) {
throw new ExecutionException("Observable onError", throwable);
} else if (this.cancelled) {
throw new CancellationException("Subscription unsubscribed");
} else {
return value.get();
}
}
};
}
- queue方法内部调用的是this.toObservable().toBlocking().toFuture();
- 所以所有方法最终都是调用的toObservable。
- 整个调用函数链为： toObservable().toBlocking().toFuture().get()
- toObservable负责创建Observable对象，实际并没有真正的创建
- toFuture负责执行任务，内部通过.subscribe触发最终的订阅、构建Observable、发射数据
- get获取执行结果
其中toObservable方法是最复杂的部分，接下来重点分析这个方法。这个方法也是在HystrixInvokable这系列继承关系里边，所以看下HystrixInvokable。
HystrixInvokable
HystrixInvokable 很关键，它的实现类是 GenericCommand 。三个抽象父类 AbstractHystrixCommand、HystrixCommand、AbstractCommand 帮助 GenericCommand 做了不少公共的事情，而 GenericCommand 负责执行具体的方法和fallback时的方法.
AbstractCommand
//验证逻辑
abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {
private static HystrixThreadPoolKey initThreadPoolKey(HystrixThreadPoolKey threadPoolKey, HystrixCommandGroupKey groupKey, String threadPoolKeyOverride) {
if (threadPoolKeyOverride == null) {
// we don't have a property overriding the value so use either HystrixThreadPoolKey or HystrixCommandGroup
if (threadPoolKey == null) {
/* use HystrixCommandGroup if HystrixThreadPoolKey is null */
return HystrixThreadPoolKey.Factory.asKey(groupKey.name());
} else {
return threadPoolKey;
}
} else {
// we have a property defining the thread-pool so use it instead
return HystrixThreadPoolKey.Factory.asKey(threadPoolKeyOverride);
}
}
//线程池启作用
private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {
...
if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.STARTED)) {
HystrixCounters.incrementGlobalConcurrentThreads();
threadPool.markThreadExecution();
...
}
}
创建线程池的作用
RxJava
如图就是rxjava在hystrix中的实现，定义了一堆action和一个func，可以把他们都理解成rxjava中的订阅者。其中action是没有返回值的，func是由返回值的。
如上图所示：Obeserve对象有几个doOnXxx方法，hystrix实现的几个功能比如缓存、统计、断路器等等需要的数据支持就来之这几个doOnXxx方法，所以就需要定义一些函数代表这些功能，然后将相关的doOnXxx方法订阅。接下来一系列非常饶的代码最终就是实现这个的
下面还会大量使用defer来构建Observable对象，先看下defer。
defer
- defer是将Func1的实例当做参数去构建OnSubscribeDefer对象，在发生订阅关系时，再去回调func1.call()。相当于懒加载，只有等observable 与observer建立了订阅关系时，observable才会建立
- defer可以动态创建Observable ，比如缓存功能无法在创建Observable阶段就知道是否有缓存，可以通过defer声明动态Observable
- 为每个Observer创建一个新的Observable。它等待观察者订阅它，然后它产生一个Observable，虽然每个用户可能会认为它正在订阅同一个Observable，实际上每个用户都有自己的Observable。
- defer与create、just、from、future都是用来构建observable的，他们的区别是什么：
- create：操作在call方法结束之后，需要手动调用subscriber.next()或subscriber.complete()方法。
- just，from或defer：都是在producer中调用的onComplete、onNext、onError方法。
- future：首先ObSubscribeToObservableFuture.call方法，在该方法中，当future取到callable的返回值后，将回调onNext、onComplete、OnError方法。
toObservable
通过hystrix发起远程服务调用的时候，无论是execute还是queue还是observe，最终都是调用的toObservable方法。所以接下来看toObservable。
abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {
public Observable<R> toObservable() {
final AbstractCommand<R> _cmd = this;
// 命令执行结束后的清理者
final Action0 terminateCommandCleanup = new Action0() {...};
// 取消订阅时处理者
final Action0 unsubscribeCommandCleanup = new Action0() {...};
// 重点：Hystrix 核心逻辑: 断路器、隔离
final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {...};
// 发射数据(OnNext表示发射数据)时的Hook
final Func1<R, R> wrapWithAllOnNextHooks = new Func1<R, R>() {...};
// 命令执行完成的Hook
final Action0 fireOnCompletedHook = new Action0() {...};
// 通过Observable.defer()创建一个Observable
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
final boolean requestCacheEnabled = isRequestCachingEnabled();
final String cacheKey = getCacheKey();
// 首先尝试从请求缓存中获取结果
if (requestCacheEnabled) {
HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);
if (fromCache != null) {
isResponseFromCache = true;
return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
}
}
// 使用上面的Func0：applyHystrixSemantics 来创建Observable
Observable<R> hystrixObservable =
Observable.defer(applyHystrixSemantics)
.map(wrapWithAllOnNextHooks);
Observable<R> afterCache;
// 如果启用请求缓存，将Observable包装成HystrixCachedObservable并进行相关处理
if (requestCacheEnabled && cacheKey != null) {
HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
...
} else {
afterCache = hystrixObservable;
}
// 返回Observable
return afterCache
.doOnTerminate(terminateCommandCleanup)
.doOnUnsubscribe(unsubscribeCommandCleanup)
.doOnCompleted(fireOnCompletedHook);
}
});
}
}
- 调用defer，传入func如下：
- 首先尝试从请求缓存中获取结果
- 调用defer，传入func：applyHystrixSemantics，通过这里获取obsrveable对象。
- 如果启用缓存，将Observable包装成HystrixCachedObservable并进行相关处理
继续看applyHystrixSemantics
// applyHystrixSemantics 是一个Func0(理解为执行实体或处理者),表示没有参数,返回值是Observable。
final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
// Func0 做的事情如下
@Override
public Observable<R> call() {
// 如果未订阅,返回一个"哑炮" Observable, 即一个不会发射任何数据的Observable
if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
return Observable.never();
}
// 调用applyHystrixSemantics()来创建Observable
return applyHystrixSemantics(_cmd);
}
};
// Semantics 译为语义, 应用Hystrix语义很拗口，其实就是应用Hystrix的断路器、隔离特性
private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
// 源码中有很多executionHook、eventNotifier的操作，这是Hystrix拓展性的一种体现。这里面啥事也没做，留了个口子，开发人员可以拓展
executionHook.onStart(_cmd);
// 判断断路器是否开启
if (circuitBreaker.attemptExecution()) {
// 获取执行信号
final TryableSemaphore executionSemaphore = getExecutionSemaphore();
final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
final Action0 singleSemaphoreRelease = new Action0() {...};
final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {...};
// 判断是否信号量拒绝
if (executionSemaphore.tryAcquire()) {
try {
// 重点：处理隔离策略和Fallback策略
return executeCommandAndObserve(_cmd)
.doOnError(markExceptionThrown)
.doOnTerminate(singleSemaphoreRelease)
.doOnUnsubscribe(singleSemaphoreRelease);
} catch (RuntimeException e) {
return Observable.error(e);
}
} else {
return handleSemaphoreRejectionViaFallback();
}
}
// 开启了断路器,执行Fallback
else {
return handleShortCircuitViaFallback();
}
}
- 判断断路器是否开启：
- 开启断路器：执行Fallback，handleShortCircuitViaFallback
- 未开启断路器，获取信号量，无论策略是线程池还是信号量都先获取一下信号量（这块没有关系，当策略是线程池的时候，获取信号量永远返回true，判断永远可以过去，在后面还有一个分支判断信号量还是线程池）：
- 信号量，如果拒绝，执行信号量Fallback，handleSemaphoreRejectionViaFallback
- 信号量，如果通过，执行executeCommandAndObserve
继续看executeCommandAndObserve
// 处理隔离策略和各种Fallback
private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
final Action1<R> markEmits = new Action1<R>() {...};
final Action0 markOnCompleted = new Action0() {...};
// 利用Func1获取处理Fallback的 Observable
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 {
...
return handleFailureViaFallback(e);
}
}
};
final Action1<Notification<? super R>> setRequestContext ...
Observable<R> execution;
// 利用特定的隔离策略来处理
if (properties.executionTimeoutEnabled().get()) {
execution = executeCommandWithSpecifiedIsolation(_cmd)
.lift(new HystrixObservableTimeoutOperator<R>(_cmd));
} else {
execution = executeCommandWithSpecifiedIsolation(_cmd);//入口
}
return execution.doOnNext(markEmits)
.doOnCompleted(markOnCompleted)
// 绑定Fallback的处理者
.onErrorResumeNext(handleFallback)
.doOnEach(setRequestContext);
}
private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {
// 线程池隔离
if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
// 再次使用 Observable.defer(), 通过执行Func0来得到Observable
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
// 收集metric信息
metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.THREAD);
...
try {
... // 获取真正的用户Task
return getUserExecutionObservable(_cmd);
} catch (Throwable ex) {
return Observable.error(ex);
}
...
}
// 绑定各种处理者
}).doOnTerminate(new Action0() {...})
.doOnUnsubscribe(new Action0() {...})
// 绑定超时处理者
.subscribeOn(threadPool.getScheduler(new Func0<Boolean>() {
@Override
public Boolean call() {
return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT;
}
}));
}
// 信号量隔离，和线程池大同小异，全部省略了
else {
return Observable.defer(new Func0<Observable<R>>() {...}
}
}
- 分线程池隔离和信号量隔离：
- 线程池隔离
- 开启defer获取observeable对象，入参func逻辑：
- 通过方法getUserExecutionObservable获取真正的用户Task
- 通过subscribeOn指定线程池运行模型
- 开启defer获取observeable对象，入参func逻辑：
- 信号量隔离
- 线程池隔离
最后看getUserExecutionObservable
private Observable<R> getUserExecutionObservable(final AbstractCommand<R> _cmd) {
Observable<R> userObservable;
try {
userObservable = getExecutionObservable();
} catch (Throwable ex) {
// the run() method is a user provided implementation so can throw instead of using Observable.onError
// so we catch it here and turn it into Observable.error
userObservable = Observable.error(ex);
}
return userObservable
.lift(new ExecutionHookApplication(_cmd))
.lift(new DeprecatedOnRunHookApplication(_cmd));
}
@Override
final protected Observable<R> getExecutionObservable() {
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
try {
//run()是GenericCommand.run，执行切入的目标方法，前面讲过的
return Observable.just(run());
} catch (Throwable ex) {
return Observable.error(ex);
}
}
}).doOnSubscribe(new Action0() {
@Override
public void call() {
// Save thread on which we get subscribed so that we can interrupt it later if needed
executionThread.set(Thread.currentThread());
}
});
}
- 通过GenericCommand.run来执行真正用户逻辑
- 将执行真正用户逻辑的返回值，通过just射为Observable类，返回
到这里一个懒加载的observable对象就构建完毕，并且为其doOnXxx绑定了各种函数。各种函数的逻辑就支撑了hystrix的各种功能。
请求缓存
replaySubject
Hystrix 的请求缓存用的就是RxJava 中的 ReplaySubject 这个特性。replay 译为重放，Subject 是个合体工具，既可以做数据发射器(被观察者、Observable)，也可以做数据消费者(观察者、Observer)。如果请求相同数据，就把原先的结果发你一份.
无论是 replaySubject 多久前发射的数据，新的订阅者都可以收到所有数据
@Test
public void replaySubject() {
ReplaySubject<Integer> replaySubject = ReplaySubject.create();
replaySubject.subscribe(v -> System.out.println("订阅者1:" + v));
replaySubject.onNext(1);//订阅者1:1
replaySubject.onNext(2);//订阅者1:2
replaySubject.subscribe(v -> System.out.println("订阅者2:" + v));//订阅者2:1 订阅者2:2
replaySubject.onNext(3);//订阅者1:3 订阅者2:3
replaySubject.subscribe(v -> System.out.println("订阅者3:" + v)); //订阅者3:1 订阅者3:2 订阅者3:3
}
在toObserve方法里有一段代码，如果开启缓存，则将查询结果包装成HystrixCachedObservable对象，看一下
public class HystrixCachedObservable<R> {
protected final Subscription originalSubscription;
protected final Observable<R> cachedObservable;
private volatile int outstandingSubscriptions = 0;
protected HystrixCachedObservable(final Observable<R> originalObservable) {
ReplaySubject<R> replaySubject = ReplaySubject.create();
// 订阅普通的Observable, 拿到其中的数据
this.originalSubscription = originalObservable
.subscribe(replaySubject);
this.cachedObservable = replaySubject...
}
...
// 将cachedObservable作为数据源提供出去, 完成普通Observable向ReplaySubject的转换
public Observable<R> toObservable() {
return cachedObservable;
}
}
可以看到内部是使用的ReplaySubject。
生命周期
public class HystrixCommandCacheTest extends HystrixCommand<String> {
private final String value;
public HystrixCommandCacheTest(String value) {
super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
this.value = value;
}
// 重要：要重写getCacheKey()方法
@Override
protected String getCacheKey() {
return value;
}
@Override
protected String run() throws Exception {
return "hello," + value;
}
public static void main(String[] args) {
// 第一个请求环境
HystrixRequestContext context1 = HystrixRequestContext.initializeContext();
HystrixCommandCacheTest cmd1 = new HystrixCommandCacheTest("kitty");
System.out.println("cmd1结果：" + cmd1.execute() + ";使用缓存：" + cmd1.isResponseFromCache());
// 模拟10个相同请求参数的命令执行
for (int i = 0; i < 10; i++) {
HystrixCommandCacheTest temp2 = new HystrixCommandCacheTest("kitty");
System.out.println("第" + i + " 次执行:" + temp2.execute() + ";使用缓存：" + temp2.isResponseFromCache());
}
try {
new Thread(() -> {
HystrixCommandCacheTest tempCmd = new HystrixCommandCacheTest("kitty");
System.out.println("线程2执行:" + tempCmd.execute() + ";使用缓存：" + tempCmd.isResponseFromCache());
}).start();
} catch (Exception ex) {//报错：Request caching is not available. Maybe you need to initialize the HystrixRequestContext?
ex.printStackTrace();
}
context1.shutdown();
// 第二个请求环境
HystrixRequestContext context2 = HystrixRequestContext.initializeContext();
HystrixCommandCacheTest cmd2 = new HystrixCommandCacheTest("kitty");
System.out.println("cmd2结果:" + cmd2.execute() + ";使用缓存：" + cmd2.isResponseFromCache());
}
}
/**
cmd1结果：hello,kitty;使用缓存：false
第0 次执行:hello,kitty;使用缓存：true
第1 次执行:hello,kitty;使用缓存：true
第2 次执行:hello,kitty;使用缓存：true
第3 次执行:hello,kitty;使用缓存：true
第4 次执行:hello,kitty;使用缓存：true
第5 次执行:hello,kitty;使用缓存：true
第6 次执行:hello,kitty;使用缓存：true
第7 次执行:hello,kitty;使用缓存：true
第8 次执行:hello,kitty;使用缓存：true
第9 次执行:hello,kitty;使用缓存：true
cmd2结果:hello,kitty;使用缓存：false
**/
可以看出来，缓存的生命周期是request
hystrix跨线程传递变量
对于ThreadLocal类型的变量，如何从servlet线程传递到hystrix线程池的线程里。
hystrix是基于InheritableThreadLocal的思想来解决这个问题的。
InheritableThreadLocal是在主线程中创建子线程的时候，将内部的ThreadLocalMap复制一份到子线程中，就相当于完成了传递。
但是servlet线程和hystrix线程池的线程并没有父子关系，那怎么做的？
基于两个类HystrixRequestContext、HystrixRequestVariableDefault来完成的。
HystrixRequestVariableDefault可以看成是ThreadLocalMap。
- 那么在servlet线程里将ThreadLocalMap的数据封装到HystrixRequestVariableDefault里，然后在把HystrixRequestVariableDefault封装到HystrixRequestContext里。在把HystrixRequestContext封装到runnable对象里。
- 在servlet线程到hystrix线程池线程传递的是一个可执行的runnable对象。那么到hystrix线程池线程里自然就可以获取到了ThreadLocal变量了。
其实很简单，就是把变量封装一下，通过runnable任务传递过去。只不过多了两个封装的类。
Metrics
在 Hystrix Command 执行过程(开始执行、结束执行、异常、超时)时会不断发出各类事件，通过收集这些数据，提供给消费者。如断路器、Hystrix Dashboard可以统计分析这些数据，从而完成特定的功能。
Metrics就是对各种事件进行数据的收集统计。
我们以一个doOnTerminate事件数据收集为例子分析一下。
入口在toObserve方法
abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {
public Observable<R> toObservable() {
final AbstractCommand<R> _cmd = this;
// 命令执行结束后的清理者
final Action0 terminateCommandCleanup = new Action0() {...};
// 取消订阅时处理者
final Action0 unsubscribeCommandCleanup = new Action0() {...};
// 重点：Hystrix 核心逻辑: 断路器、隔离
final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {...};
// 发射数据(OnNext表示发射数据)时的Hook
final Func1<R, R> wrapWithAllOnNextHooks = new Func1<R, R>() {...};
// 命令执行完成的Hook
final Action0 fireOnCompletedHook = new Action0() {...};
// 通过Observable.defer()创建一个Observable
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
final boolean requestCacheEnabled = isRequestCachingEnabled();
final String cacheKey = getCacheKey();
// 首先尝试从请求缓存中获取结果
if (requestCacheEnabled) {
HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);
if (fromCache != null) {
isResponseFromCache = true;
return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
}
}
// 使用上面的Func0：applyHystrixSemantics 来创建Observable
Observable<R> hystrixObservable =
Observable.defer(applyHystrixSemantics)
.map(wrapWithAllOnNextHooks);
Observable<R> afterCache;
// 如果启用请求缓存，将Observable包装成HystrixCachedObservable并进行相关处理
if (requestCacheEnabled && cacheKey != null) {
HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
...
} else {
afterCache = hystrixObservable;
}
// 返回Observable
return afterCache
.doOnTerminate(terminateCommandCleanup)//入口
.doOnUnsubscribe(unsubscribeCommandCleanup)
.doOnCompleted(fireOnCompletedHook);
}
});
}
}
可以看到，在doOnTerminate绑定了一个函数terminateCommandCleanup。
继续看terminateCommandCleanup
final Action0 unsubscribeCommandCleanup = new Action0() {
@Override
public void call() {
if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.UNSUBSCRIBED)) {
.......省略干扰代码...........
handleCommandEnd(false); //user code never ran
} else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.UNSUBSCRIBED)) {
.......省略干扰代码...........
handleCommandEnd(true); //user code did run
}
}
};
该方法调用了handleCommandEnd
private void handleCommandEnd(boolean commandExecutionStarted) {
........省略干扰代码..........
executionResult = executionResult.markUserThreadCompletion((int) userThreadLatency);
if (executionResultAtTimeOfCancellation == null) {
metrics.markCommandDone(executionResult, commandKey, threadPoolKey, commandExecutionStarted);
} else {
metrics.markCommandDone(executionResultAtTimeOfCancellation, commandKey, threadPoolKey, commandExecutionStarted);
}
........省略干扰代码..........
}
- 调用了HystrixCommandMetrics的markCommandDone方法，把一个ExecutionResult传入了进来
- executionResult：HystrixCommand的执行结果记录，也包含了各种状态以及出现的异常
继续看markCommandDone
void markCommandDone(ExecutionResult executionResult, HystrixCommandKey commandKey, HystrixThreadPoolKey threadPoolKey, boolean executionStarted) {
HystrixThreadEventStream.getInstance().executionDone(executionResult, commandKey, threadPoolKey);
if (executionStarted) {
concurrentExecutionCount.decrementAndGet();
}
}
public void executionDone(ExecutionResult executionResult, HystrixCommandKey commandKey, HystrixThreadPoolKey threadPoolKey) {
HystrixCommandCompletion event = HystrixCommandCompletion.from(executionResult, commandKey, threadPoolKey);
writeOnlyCommandCompletionSubject.onNext(event);
}
- 调用的是HystrixThreadEventStream的executionDone
- HystrixThreadEventStream是ThreadLocal类型的，线程级别的
- 根据executionResult, threadpoolkey,comandKey，生成一个HystrixCommandCompletion然后通过writeOnlyCommandCompletionSubject写入
- HystrixCommandCompletion：包含了ExecutionResult（执行结果）和HystrixRequestContext（上下文），它是一种HystrixEvent，标识着command执行完成的一个事件，该事件是当前这个点HystrixCommand的请求信息，执行结果，状态等数据的封装
- writeOnlyCommandCompletionSubject.onNext(event)：该操作又会触发writeCommandCompletionsToShardedStreams的call方法
看一下writeCommandCompletionsToShardedStreams的call方法
private static final Action1<HystrixCommandCompletion> writeCommandCompletionsToShardedStreams = new Action1<HystrixCommandCompletion>() {
@Override
public void call(HystrixCommandCompletion commandCompletion) {
HystrixCommandCompletionStream commandStream = HystrixCommandCompletionStream.getInstance(commandCompletion.getCommandKey());
commandStream.write(commandCompletion);
if (commandCompletion.isExecutedInThread() || commandCompletion.isResponseThreadPoolRejected()) {
HystrixThreadPoolCompletionStream threadPoolStream = HystrixThreadPoolCompletionStream.getInstance(commandCompletion.getThreadPoolKey());
threadPoolStream.write(commandCompletion);
}
}
};
- 调用HystrixCommandCompletionStream的write方法
- 内部实际调用的是writeOnlySubject的方法
先总结一下流程：
- Command直接使用HystrixCommandMetrics来记录命令开始、结束等事件
- HystrixCommandMetrics利用线程级别的HystrixThreadEventStream的来接收数据
- HystrixThreadEventStream完成各种事件的封装(如将结束事件封装成HystrixCommandCompletion)，
- 再利用command级别的HystrixEventStream来接收数据(HystrixEventStream有不同的子类来处理不同的事件)
- 最终消费者通过HystrixEventStream的observe()方法，拿到这个数据源，然后订阅它，从而源源不断的拿到Command发射出的各种数据
这套过程中出现了好几个类，有必须要认识一下
HystrixCommandMetrics
- 每个Command的构造器中会获取一个HystrixCommandMetrics工具，用来记录metrics，也就是说，每个CommandKey会拥有一个对应的HystrixCommandMetrics工具
HystrixThreadEventStream
- HystrixCommandMetrics主要是通过HystrixThreadEventStream工作的
- 每个线程拥有自己的HystrixThreadEventStream（从ThreadLocal中获取对象），它包含了很多Subject<事件,事件>，用来接收和发射数据。
HystrixEventStream
- HystrixThreadEventStream的事件下一个接收者。
- 实现类包括：
- HystrixCommandCompletionStream
- HystrixCommandStartStream
- HystrixThreadPoolCompletionStream
- HystrixThreadPoolStartStream
- HystrixCollapserEventStream
BucketedCounterStream
- 一堆stream，不同的stream实现了不同的相关统计功能
- BucketedCounterStream：实现了基本的桶计数器，下层有两个子类。第三层就是对特定指标的具体实现
- BucketedCumulativeCounterStream基于父类实现了累计计数
- BucketedRollingCounterStream基于父类实现了滑动窗口计数
- BucketedCounterStream：实现了基本的桶计数器，下层有两个子类。第三层就是对特定指标的具体实现
- HystrixEventStream事件的下一个接收者
- 一个事件从HystrixThreadEventStream传递到HystrixEventStream，然后传递到BucketedCounterStream。
Rxjava
- Observable.share：属于连接操作，组合操作
- Observable.publish( )：将一个Observable转换为一个ConnectableObservable（可连接的Observable）
- ConnectableObservable.refCount( )：让一个ConnectableObservable表现得像一个普通的Observable
- ConnectableObservable与普通的Observable差不多，ConnectableObservable在被订阅时并不开始发射数据，只有在它的connect()被调用时才开始。用这种方法，你可以等所有的潜在订阅者都订阅了这个Observable之后才开始发射数据
HystrixEventNotifier
在执行 HystrixCommand和 HystrixObservableCommand期间发生的事件在 HystrixEventNotifier上触发，以提供警报和统计信息收集的机会。
BucketedCounterStream基本桶的实现
请求经过各种stream，最终到达的是BucketedCounterStream。所以先看一下BucketedCounterStream基本桶的实现。
先看下其构造器
protected BucketedCounterStream(final HystrixEventStream<Event> inputEventStream, final int numBuckets, final int bucketSizeInMs,
final Func2<Bucket, Event, Bucket> appendRawEventToBucket) {
this.numBuckets = numBuckets;
// 每个桶如何计算聚合的数据
this.reduceBucketToSummary = new Func1<Observable<Event>, Observable<Bucket>>() {
@Override
public Observable<Bucket> call(Observable<Event> eventBucket) {
return eventBucket.reduce(getEmptyBucketSummary(), appendRawEventToBucket);
}
};
//第一个桶的定义
final List<Bucket> emptyEventCountsToStart = new ArrayList<Bucket>();
for (int i = 0; i < numBuckets; i++) {
emptyEventCountsToStart.add(getEmptyBucketSummary());
}
this.bucketedStream = Observable.defer(new Func0<Observable<Bucket>>() {
@Override
public Observable<Bucket> call() {
return inputEventStream
.observe()
//window窗口的定义，这里第一个参数就是每个桶的时长，
// 第二个参数时间的单位。利用RxJava的window帮我们做聚合数据。
.window(bucketSizeInMs, TimeUnit.MILLISECONDS)
.flatMap(reduceBucketToSummary)
.startWith(emptyEventCountsToStart);
}
});
}
- reduceBucketToSummary每个桶如何计算聚合的数据，具体由其子类实现，比如HealthCountsStream的实现是将次数做累加
- emptyEventCountsToStart第一个桶的定义
- window窗口的定义，这里第一个参数就是每个桶的时长，第二个参数时间的单位。利用RxJava的window帮我们做聚合数据。
- window操作符会在时间间隔内缓存结果
- 像滑动窗口时长、桶时长、桶个数，这些都是可配置的，不桶的BucketedCounterStream的实现也不一样。
熔断器
经过Metrics的学习，我们假设滑动窗口的统计数据已经准备完毕。
- 熔断器是个开关，有三个状态：
- closed：当关闭的时候请求直接进入fallback
- open：可以正常执行请求
- halfOpen：关闭状态到一定事件，会进入该状态，尝试处理一个请求，如果该请求成功则进入open状态，失败进入closed状态
- 核心类是HystrixCircuitBreaker，它有两个实现类
- NoOpCircuitBreaker：什么也每做，代表未开启熔断器
- HystrixCircuitBreakerImpl：封装熔断器主要逻辑
- Hystrix为每个commandKey都维护了一个熔断器
开启熔断器
protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
this.properties = properties;
this.metrics = metrics;
//On a timer, this will set the circuit between OPEN/CLOSED as command executions occur
Subscription s = subscribeToStream();
activeSubscription.set(s);
}
private Subscription subscribeToStream() {
/*
* This stream will recalculate the OPEN/CLOSED status on every onNext from the health stream
*/
return metrics.getHealthCountsStream()
.observe()
.subscribe(new Subscriber<HealthCounts>() {
//.....................省略干扰代码......................
@Override
public void onNext(HealthCounts hc) {
// check if we are past the statisticalWindowVolumeThreshold
if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
} else {
if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
} else {
if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
circuitOpened.set(System.currentTimeMillis());
}
}
}
}
});
}
- HystrixBreaker启动的时候会订阅HystrixCommandMetrics的HealthCountsStream，每当HealthCountsStream搜集到数据，都会触发上面的onNext方法，然后该方法做下面几个判断：
- 当前请求量是否达到设定水位（请求量太小不做阀值控制）
- 当前的请求错误量是否达到阀值，达到后会将熔断器状态置为OPEN, circuitOpened设置为当前时间戳表示开启的时间。
熔断器的使用
入口在toObServe方法中，有一步通过attemptExecution获取当前是否开启熔断器
public boolean attemptExecution() {
if (properties.circuitBreakerForceOpen().get()) {
return false;
}
if (properties.circuitBreakerForceClosed().get()) {
return true;
}
if (circuitOpened.get() == -1) {
return true;
} else {
if (isAfterSleepWindow()) {
if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
//only the first request after sleep window should execute
return true;
} else {
return false;
}
} else {
return false;
}
}
}
- 判断是否强制开启熔断器，是则command不能执行
- 判断是否强制关闭熔断器，是则command可执行
- 判断熔断器是否开启,没有开启，command可执行。
- 开启了熔断器，那么判断是否可尝试请求，如果可以同时会把熔断器的状态改为HALF_OPEN
Sentinel
Sentinel是阿里中间件团队研发的面向分布式服务架构的轻量级高可用流量控制组件
