引子
在大量微服务所构成的分布式系统中，某个基础服务的不可用，可能导致服务雪崩效应，即：依赖该基础服务的所有其它基础服务及级联的上游服务的级联性不可用故障。
熔断机制是防止服务雪崩的基本技术手段。通过检查依赖服务的失败状况并封装熔断逻辑，阻止在依赖服务暂时出现故障期间的错误反复不断地向上传播。基本思路是快速失败和 Fallback 机制。
熔断的主要目的：防止服务雪崩效应；防止局部次要失败影响整体可用性。
设计目标
熔断机制的设计目标是：在未达到熔断要求时，正常调用依赖服务；在达到熔断要求时，调用指定的降级方法或抛出异常。这要求定义熔断要求，做一个熔断器设计。
熔断机制的设计挑战是：上一次服务调用的结果，会决定下一次调用的策略（降级、抛出异常或恢复服务调用）。原来相互独立的服务调用被整合成有密切关联的。若某个依赖服务配置了熔断要求，则针对该依赖服务的所有调用必须整合成一个连续不中断的处理流。若处理流因为超时或异常中断，则无法正确决定下一次调用，也就起不到熔断和恢复的作用。
针对这种需求，常用的一种处理方式是采用事件机制。将每一次依赖服务调用或者降级调用转换成一次事件，建立事件监听器进行事件统计，将事件统计结果传给熔断器，熔断器来决定下一次调用的走向。
实现思路
熔断机制的实现思路主要包括：
- 熔断配置：失败事件的阈值、执行器配置等；失败事件包括超时、异常事件；
- 断路器：在失败事件达到指定阈值时，将依赖服务的调用熔断，采取降级策略；采取某种规则从熔断状态恢复到正常状态；
- 事件统计：在特定时间窗口内，统计依赖服务调用成功事件、失败事件（失败次数、失败比率等）、异常事件等；
- 事件机制：连接事件统计、断路器状态机、服务调用与熔断降级，串联成完整的流程。
重点是断路器和事件机制的设计实现。
断路器
熔断情形主要有：强制直接熔断；断路器开启、调用次数达到总数阈值且失败事件达到指定阈值（失败百分比、失败绝对次数等）。主要是断路状态机的设计实现。断路器状态机如图所示：
断路状态机有三个状态： CLOSE （关闭），HALF-OPEN （半开），OPEN （开启）。断路器默认是 CLOSE 状态，此时，正常调用依赖服务。
当调用次数达到总数阈值且失败事件的阈值达到指定值时，进入 OPEN 状态，开启降级逻辑；当断路器位于 OPEN 状态时，将进入一段断路时间窗期，这个时间窗内的请求将不会转发给依赖服务，而是转发给指定的降级逻辑；当断路器位于 OPEN 状态，且过了断路时间窗期，就会进入 HALF-OPEN 状态。
断路器使用称为 HALF-OPEN 状态的监视和反馈机制来了解依赖服务是否以及何时恢复。在 HALF-OPEN 状态，根据规则将部分请求转发给依赖服务（默认是只重试第一次请求），若调用成功则进入 CLOSE 状态，恢复调用依赖服务，若对依赖服务的调用超时或失败，则断路器保持在 OPEN 状态。
事件统计
可以划分为事件窗口和若干个数据桶分别统计，再进行数据聚合。
事件机制
将依赖服务的调用结果转化为事件，触发事件监听器，对成功及失败事件进行统计，判断是否触发断路器的熔断，以决定下一次调用是调用依赖服务还是降级策略或者直接抛出异常。
事件机制通常基于观察者模式。
Hystrix
Hystrix 是业界采用的熔断机制的主要实现之一。
Hystrix 使用起来比较简单：使用 @EnableCircuitBreaker 开启熔断功能，使用 @HystrixCommand 为方法添加熔断降级配置。下面先给出代码示例，再讲讲原理的实现和部分源码的理解。
使用
步骤一：引入 POM 依赖（SpringBoot 版本为 1.5.9.RELEASE ）
<!-- 引入 hystrix 熔断 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
<version>1.4.7.RELEASE</version>
</dependency>
步骤二：在应用启动类 Application.java 上开启熔断
package cc.lovesq;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
@EnableCircuitBreaker // 开启熔断机制
@SpringBootApplication
@Configuration
@ComponentScan(basePackages = {"cc.lovesq.*"})
@MapperScan(basePackages = {"cc.lovesq.dao"})
@ImportResource(locations={"classpath:spring.xml"})
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}
步骤三：为方法配置熔断设置。通常是在 Service 层做熔断。
package cc.lovesq.service.impl;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixProperty;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.stereotype.Component;
import java.util.Random;
/**
* @Description 返回随机值
* @Date 2021/1/28 4:37 上午
* @Created by qinshu
*/
@Component
public class RandomValueService {
private static Log log = LogFactory.getLog(RandomValueService.class);
Random rand = new Random(47);
@HystrixCommand(commandKey = "randInt", fallbackMethod = "randIntDowngrade",
commandProperties = {
@HystrixProperty(name="metrics.rollingStats.timeInMilliseconds", value="5000"),
@HystrixProperty(name="circuitBreaker.requestVolumeThreshold", value="10"),
@HystrixProperty(name="circuitBreaker.errorThresholdPercentage", value="50")
})
public Integer randInt() {
int v = rand.nextInt(100);
if (v == 0) {
throw new RuntimeException("Invalid number");
}
return v;
}
public Integer randIntDowngrade() {
log.info("randIntDowngrade");
return 0;
}
}
步骤四：测试熔断。
package cc.lovesq.experiments;
import cc.lovesq.service.impl.RandomValueService;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @Description 熔断实现
* @Date 2021/1/28 4:39 上午
* @Created by qinshu
*/
@Component
public class HystrixExperiment implements IExperiment {
private static Log log = LogFactory.getLog(HystrixExperiment.class);
@Resource
private RandomValueService randomValueService;
@Override
public void test() {
log.info("----HystrixExperiment-start----");
for (int i=0; i < 100000; i++) {
try {
randomValueService.randInt();
} catch (Exception ex) {
log.error("CatchedException: " + ex.getMessage());
}
}
log.info("----HystrixExperiment-end----");
}
}
原理及实现
Hystrix 将服务调用封装成一个 HystrixCommand 来执行。HystrixCommand 的熔断流程如下图所示：
Hystrix 的整个流程如下图所示：
Hystrix 的基本实现如下：
- 最主要的 API 是 @HystrixCommand 注解。@HystrixCommand 注解的处理器是 hystrix-javanica 包里的 HystrixCommandAspect.methodsAnnotatedWithHystrixCommand 方法。这里使用了切面编程来封装和隔离依赖服务调用：从切面 joinPoint 提取目标方法 targetMethod 并转换为可执行的 HystrixInvokable 实例（ 实现类是 GenericCommand ）;
- GenericCommand 实例的创建使用了 Builder 模式；GenericCommand 实例的执行使用了 Command 模式；
- 断路器实现类是 HystrixCircuitBreakerImpl。circuitOpened 表示断路器开启的时间，默认值 -1 表示熔断未开启，大于 0 表示熔断已开启；circuitBreakerSleepWindowInMilliseconds 表示断路器开启到重试请求的“断路-休眠时间窗口期”。断路器的实现并不复杂。
源码解析
Hystrix 借用了 RxJava 库的响应式编程机制，代码有点绕，需要细细多看几遍。总的来说，就是 Observable 对象产生的事件导致的回调和转换，以及回调和转换的级联处理。
要理解 Hystrix 源码，需要有一定的响应式编程基础，可参阅 “响应式编程库RxJava初探”。为什么要采用响应式编程模型呢？因为 Hystrix 要处理的回调比较复杂：
- 同步执行或异步执行服务调用的命令；
- 执行服务调用的命令的回调处理：成功时、异常时、超时时、取消时、终止时的情形（回调的多种情形处理）；
- 当执行服务调用失败后，采取降级策略，降级方法执行同样有：成功时、失败时的情形（回调的嵌套与级联）；
- 连续执行多次服务调用的结果，要源源不断送往事件统计模块（回调的连续性）；
- 连续执行多次服务调用，若其中某些发生异常，整个执行流不能中断，要继续发送给事件统计模块（回调的连续性）。
可以看到，这需要一个连续不断的回调流，连续不断地处理各种事件。响应式编程模型正好能够对付这种编程需求。
GenericCommand 的执行入口代码如下：
public R execute() {
try {
return queue().get();
} catch (Exception e) {
throw Exceptions.sneakyThrow(decomposeException(e));
}
}
其中 queue 方法的代码如下：
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(); // 这句是重点
final Future<R> f = new Future<R>() {
// 为 delegate 对象封装线程中断功能，暂时跳过
}
return f;
重点是 toObservable().toBlocking().toFuture(); 这句，主要干的事情是：首先将 GenericCommand 的执行转换成一个 Observable 对象，从而能够变成可监听的事件，连接后面的事件统计、断路器状态机及熔断功能，最后再转换成一个 Future 对象，来获取服务调用命令的结果。
先说说 toBlocking().toFuture() 这部分，toBlocking() 使用了装饰器模式，将 Observable 对象装饰成一个可阻塞的 BlockingObservable 对象，阻塞并等待被装饰的 Observable 对象的执行完成事件或发生异常事件；toFuture 方法将 BlockingObservable 的执行转换成一个 Future 对象，使用 CountDownLatch 锁来实现阻塞功能（that 就是被装饰的 Observable 对象）：
public static <T> Future<T> toFuture(Observable<? extends T> that) {
final CountDownLatch finished = new CountDownLatch(1);
final AtomicReference<T> value = new AtomicReference<T>();
final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
@SuppressWarnings("unchecked")
final Subscription s = ((Observable<T>)that).single().subscribe(new Subscriber<T>() {
@Override
public void onCompleted() {
finished.countDown();
}
@Override
public void onError(Throwable e) {
error.compareAndSet(null, e);
finished.countDown();
}
@Override
public void onNext(T v) {
// "single" guarantees there is only one "onNext"
value.set(v);
}
});
return new Future<T>() {
// 根据 finished 的状态、value 及 error 封装一个 Future 的实现，等待 that 的订阅执行完成之后，获取结果；暂时跳过
};
}
接下来就是重点的 toObservable 方法了。 这个方法首先定义了一些回调函数：
- terminateCommandCleanup : Command 执行终止后的清理函数；
- unsubscribeCommandCleanup : Command 取消执行时的清理函数；
- applyHystrixSemantics : 执行 Hystrix 熔断语义的主要函数；
- wrapWithAllOnNextHooks : 将一个 R 对象转换成另一个 R 对象，这里 R 通常是 Observable 对象；
- fireOnCompletedHook : 命令执行完成后的钩子函数。
然后创建了一个装配了这些回调函数的带缓存功能的命令执行的 Observable 对象，在命令执行的不同阶段或发生异常时，就会执行对应的方法。
return Observable.defer(new Func0<Observable<R>>() {
@Override
public Observable<R> call() {
// 命令执行状态判断
// 命令执行日志记录
// 带缓存的命令执行
Observable<R> hystrixObservable =
Observable.defer(applyHystrixSemantics)
.map(wrapWithAllOnNextHooks);
Observable<R> afterCache;
// put in cache
if (requestCacheEnabled && cacheKey != null) {
// 从缓存里取 afterCache
} else {
afterCache = hystrixObservable;
}
return afterCache
.doOnTerminate(terminateCommandCleanup) // perform cleanup once (either on normal terminal state (this line), or unsubscribe (next line))
.doOnUnsubscribe(unsubscribeCommandCleanup) // perform cleanup once
.doOnCompleted(fireOnCompletedHook);
}
});
Observable.defer 将一个 Func0[Observable] 对象包装成一个 Observable 对象。
接着看主要方法 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);
/* 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());
return executeCommandAndObserve(_cmd)
.doOnError(markExceptionThrown) // 设置异常回调
.doOnTerminate(singleSemaphoreRelease) // 设置终止时回调，归还执行调用服务的许可
.doOnUnsubscribe(singleSemaphoreRelease); // 设置取消回调，归还执行调用服务的许可
} catch (RuntimeException e) {
return Observable.error(e);
}
} else { // 没有足够线程来执行服务调用，采取降级策略
return handleSemaphoreRejectionViaFallback();
}
} else { // 断路器不允许请求服务调用的情形，采取降级策略
return handleShortCircuitViaFallback();
}
}
断路器执行逻辑如下：
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)) { // 如果熔断状态可以从 OPEN 转换为 HALF-OPEN，可以执行一次依赖服务调用
//only the first request after sleep window should execute
return true;
} else { // 熔断状态已经为 HALF-OPEN, 已经执行过一次服务调用，后续请求暂时不能执行依赖服务调用
return false;
}
} else { // // 熔断开启过，仍然处于熔断时间窗口期，不调用依赖服务
return false;
}
}
}
最后看一下executeCommandAndObserve 方法。这个方法也是一样的套路：先定义若干回调函数，然后创建一个装配了这些回调函数的 Observable 对象，以便在适当的时候被触发调用。这里面还封装了事件的通知，比如 eventNotifier.markEvent(HystrixEventType.XXX, commandKey); 以及执行结果的处理，比如 executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS); 。ExecutionResult 对象使用了 Immutable Variable 模式，简化了对执行结果的并发处理。
private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
final Action1<R> markEmits = new Action1<R>() {
@Override
public void call(R r) {
if (shouldOutputOnNextEvents()) {
executionResult = executionResult.addEvent(HystrixEventType.EMIT);
eventNotifier.markEvent(HystrixEventType.EMIT, commandKey);
}
if (commandIsScalar()) {
long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
circuitBreaker.markSuccess();
}
}
};
final Action0 markOnCompleted = new Action0() {
@Override
public void call() {
if (!commandIsScalar()) {
long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
circuitBreaker.markSuccess();
}
}
};
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);
}
}
};
final Action1<Notification<? super R>> setRequestContext = new Action1<Notification<? super R>>() {
@Override
public void call(Notification<? super R> rNotification) {
setRequestContextIfNeeded(currentRequestContext);
}
};
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)
.onErrorResumeNext(handleFallback)
.doOnEach(setRequestContext);
}
降级处理的主要逻辑如下：
private Observable<R> getFallbackOrThrowException(final AbstractCommand<R> _cmd, final HystrixEventType eventType, final FailureType failureType, final String message, final Exception originalException) {
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)) {
// 不可恢复的异常处理，调用 onError 方法；
} 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()) { // 若持有降级逻辑的执行许可，则生成用于降级逻辑处理的 Observable 对象
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);
}
}
}
Hystrix 的代码处理流程先分析到这里。整个代码套路基本明了：
- Observable 对象 + 各种回调函数 => 回调处理后，生成新的 Observable 对象，实现 Observable 对象的级联和嵌套执行；
- Observable 对象的行为产生各种事件，然后标记事件类型，进行事件统计，并生成不可变的执行结果对象 executionResult ；
- 封装了服务调用的 Command 对象或可执行的 Action0[Observable]（无返回值） 或 Func0[Observable] （有返回值） 对象通过 Observable.defer 方法转换为 Observable 对象；
- Observable 对象的预先指定的各种回调函数的调用会返回 Observable 对象，比如 Observable.doOnComplete(onCompleted)，先将回调函数对象 onCompleted 行为封装成 ActionObserver ，再包装成 Observable 对象，从而使整个回调链始终处于观察者模式驱动的事件处理流中；这里使用了OnSubscribeDoOnEach 对象来保证观察者订阅调用的级联；
- Command 对象的执行需要熔断状态判断及熔断逻辑线程池的执行许可；同样，降级逻辑的执行需要降级逻辑线程池的执行许可；
- Command 对象执行后产生事件，通过预先指定的各种回调函数处理后产生新的 Observable 对象，新的 Observable 对象的方法执行进一步产生新事件，新事件又需要预先指定的各种回调函数处理，往复循环，不厌其烦；简单点说，就是把各种非 Observable 的对象、行为和结果设法再转化成 Observable 对象，以维持庞大的 Observable 系统的周而复始的运转；
- Command 对象执行失败后的降级逻辑处理，也是遵循类似的套路。
具体回调流程的细节，恐怕需要再单步调试加仔细梳理了。在这里，我们只需要知道这些代码要实现的目标，至于代码怎么写（某种代码逻辑组织方式），倒是其次的事情。我觉得 Hystrix 的开发人员再过三个月也不认得自己都写得些啥了。总的来说， Hystrix 的代码实现还是比较优雅的，不过优雅的代码并不一定易懂。这个也与熔断机制的设计挑战有关。
最后说下事件统计，主要有两种主要的滑动窗口计数机制，其实现分别在对象 OperatorWindowWithTime（在指定时间范围内聚合计数） 和 OperatorWindowWithSize（在指定数量内聚合计数） 里。比如 BucketedCounterStream 的计数，使用 OperatorWindowWithTime 来实现：
this.bucketedStream = Observable.defer(new Func0<Observable<Bucket>>() {
@Override
public Observable<Bucket> call() {
return inputEventStream
.observe()
.window(bucketSizeInMs, TimeUnit.MILLISECONDS) //bucket it by the counter window so we can emit to the next operator in time chunks, not on every OnNext
.flatMap(reduceBucketToSummary) //for a given bucket, turn it into a long array containing counts of event types
.startWith(emptyEventCountsToStart); //start it with empty arrays to make consumer logic as generic as possible (windows are always full)
}
});
小结
熔断机制是分布式的微服务体系中必不可少的技术手段，用来防止服务雪崩。本文总结了熔断机制的实现原理及 Hystrix 的使用和基本的源码解析。
参考文献
- Hystrix源码(1.5.12版本)
- “服务容错保护断路器Hystrix之二：Hystrix工作流程解析”
- “Hystrix核心基础 - 滑动窗口创建过程及demo”