Hystrix

在分布式系统中，各个服务之间会进行依赖调用，因为网络原因或者是服务本身出现故障，可能会导致调用失败或延迟，这就有可能造成调用方出现延迟的情况，造成线程阻塞。若此时调用方的请求不断增加，就会因为等待故障方响应形成任务积压，使得线程资源消耗殆尽，最后造成自身服务的瘫痪，并且有可能造成整个系统瘫痪，即雪崩效应。

举个例子：例如在电商网站中，可能存在用户、订单、库存等很多服务。当用户创建订单时，会调用订单服务的接口，然后订单服务会调用库存服务接口判断库存量等操作，如果此时库存服务因为网络原因或者出现了故障，就会导致创建订单服务的线程被挂起，等待库存服务的响应。在并发量极大的情况下，这些挂起的线程因为未释放导致线程资源用尽，会使得后面所有订单服务的请求阻塞，最后造成订单服务的瘫痪。

在分布式系统中，服务依赖关系错综复杂，如果一个服务出现了故障，很可能导致整个系统的瘫痪，为了解决这个问题，熔断器等服务保护机制应运而生。当某个服务发生故障时，通过熔断器的故障监控，会向调用方迅速返回失败信息，而不是长时间的等待响应，这样就不会因为线程因调用故障服务被长时间占用不释放，从而造成阻塞，避免了故障在系统中的蔓延。

Spring Cloud Hystrix是基于Netflix的Hystrix实现的熔断器，具备服务降级、服务熔断、线程和信号隔离、服务监控等强大功能，从而对系统提供更强大的容错能力。

入门案例

首先，我们需要启动以下工程，使用我们上节的例子即可

服务注册中心：Eureka-server
启动两个服务提供者实例，端口号分别为9998和9999,Eureka-Client
启动服务消费者，端口号为8888,Eureka-Consumer

此时我们还未加入熔断器，我们关闭一个端口号为9998的服务提供者实例。
我们多次访问http://localhost:8888/test/aaa
由于Ribbon默认使用轮询的负载均衡策略，所以当访问到9998的服务实例时，会出现如下出错页面。

接下来我们引入Spring Cloud Hystrix，我们首先在服务消费者的pom文件中添加依赖。

       <dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-hystrix</artifactId><version>2.0.3.RELEASE</version></dependency>

然后在启动类上使用@EnableCircuitBreaker注解开启熔断器功能

@EnableCircuitBreaker
@EnableEurekaClient
@SpringBootApplication
public class Application {public static void main(String[] args) {SpringApplication.run(Application.class);}@Bean@LoadBalancedRestTemplate restTemplate(){return new RestTemplate();}
}

我们发现启动类上已经有三个注解，我们可以使用@SpringCloudApplication这个注解代替这三个注解，我们看一下@SpringCloudApplication源码。可以看到该注解包含了我们上面的三个注解，这也意味着标准Spring Cloud应用应该包含服务发现和熔断器功能。

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@SpringBootApplication
@EnableDiscoveryClient
@EnableCircuitBreaker
public @interface SpringCloudApplication {
}

接下来呢，我们得去改造我们的服务消费接口。在服务消费接口上增加 @HystrixCommand接口指定回调函数。编写回调函数。回调函数的返回值和参数要和消费接口一样，否则会报错。

@RestController
public class ConsumerController {@AutowiredRestTemplate restTemplate;@HystrixCommand(fallbackMethod = "testFallback")@RequestMapping("/test/{msg}")public  String  test(@PathVariable String msg){return restTemplate.getForObject("http://eureka-client01/hello/"+msg,String.class);}public String testFallback(String msg){return "Hystrix Error";}
}

接下来我们启动刚才关闭的9998服务，确保能够正常访问并且返回正常结果。
然后我们关闭9998服务，继续访问，然后发现轮询到9998时，错误信息已经变为了我们回调函数的返回结果，说明Hystrix生效。

原理分析

首先，先上一张官方的工作流程图，可能不是很清楚，可以去Github上看大图。
https://github.com/Netflix/Hystrix/wiki/How-it-Works

(1)首先当客户端发送一个请求的时候，创建一个HystrixCommand或HystrixObservableCommand对象，用来表示对依赖服务的操作请求，同时传递所有需要的参数。从其命名我们就可以了解到它采用了“命令模式”来实现对服务调用操作的封装。

HystrixCommand：用在依赖的服务返回单个操作结果的时候。
HystrixObservableCommand：用在依赖的服务返回多个操作结果的时候。

(2)从图中可以看出有四种方法可以执行命令，而Hystrix在执行时会根据创建的Command对象以及具体的情况来选择一个执行。

HystrixCommand支持四种执行方式：

execute():同步执行，从依赖的服务返回一个单一的结果对象，或是在发生错误的时候抛出异常。
queue():异步执行，直接返回一个Future对象，其中包含了服务执行结束时要返回的单一结果对象。

R  value=command.execute();
Future<R> value=comman.queue();

而HystrixObservableCommand只支持以下两种执行方式：

observe():返回Observable对象，它代表了操作的多个结果，它是一个Hot Observable
toObservable():同样会返回Observable对象，代表了操作的多个结果，它是一个Cold Observable

Observable<R> hvalue=command.observe();
Observable<R> cvalue=command.toObservable();

这儿的Hot Cold是什么意思呢？，其实在Hystrix底层大量使用了RxJava,这是个开源库。RxJava的核心就是观察者模式，接下来我们来简单理解下观察者模式。

上面的Observable对象可以理解为"事件源"或者"被观察者"，与其对应的是Subscriber对象，可以理解为"观察者"或者"订阅者"。

Observable用来向订阅者Subscriber对象发布事件，而Subscriber对象则在接收到事件后对其进行处理，而在这里指的事件通常就是对依赖服务的调用。

一个Observable可以发出多个事件，直到结束或者发生异常。
Observable对象每发出一个事件，就会调用对应观察者Subscriber对象的onNext()方法。

每一个Observable的执行，最后一定会通过调用Subscriber.onCompleted()或者Subscriber.onError()来结束该事件的操作流。

我们对事件源Observable提到了两个不同的概念，Hot Observable和Cold Observable。
Hot Observable，不论事件源是否有订阅者，都会在创建后对事件进行发布，所以对于Hot Observable的每一个订阅者都有可能是中途开始的，可能只是看到了整个操作流程的局部过程。
Cold Observable在没有订阅者的时候不会发布事件，而是进行等待，直到有订阅者之后才发布事件，所以对于Cold Observable的订阅者，可以保证看到整个操作的完整过程。

其实不只是HystrixObservableCommand使用了RxJava，实际上execute()、queue()也都使用了RxJava来实现。我们来看下源码。

public R execute() {try {return queue().get();} catch (Exception e) {throw Exceptions.sneakyThrow(decomposeException(e));}}public Future<R> queue() {final Future<R> delegate = toObservable().toBlocking().toFuture();final Future<R> f = new Future<R>() {@Overridepublic boolean cancel(boolean mayInterruptIfRunning) {if (delegate.isCancelled()) {return false;}if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) {interruptOnFutureCancel.compareAndSet(false, mayInterruptIfRunning);}final boolean res = delegate.cancel(interruptOnFutureCancel.get());if (!isExecutionComplete() && interruptOnFutureCancel.get()) {final Thread t = executionThread.get();if (t != null && !t.equals(Thread.currentThread())) {t.interrupt();}}return res;}@Overridepublic boolean isCancelled() {return delegate.isCancelled();}@Overridepublic boolean isDone() {return delegate.isDone();}@Overridepublic R get() throws InterruptedException, ExecutionException {return delegate.get();}@Overridepublic R get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {return delegate.get(timeout, unit);}};........return f;}

从源码中可以看到，execute()是通过queue()返回的Future< R >的get()方法来实现同步执行的。该方法会等待任务执行结束，然后获得R类型的结果进行返回。

queue()则是通过toObservable()来获得一个Cold Observable，并且通过toBlocking()将该Observable转换成BlockingObservable，它可以把数据以阻塞的方式发射出来。而toFuture()则是把BlockingObservable转换成一个Future，该方法只是创建一个Future返回，并不会阻塞，这使得消费者可以自己决定如何处理异步操作。而execute()就是直接使用Future中的阻塞方法get()来实现同步操作的，同时通过这种方式转换的Future要求Observable只发射一个数据，所以这两个实现都只能返回单一结果。

(3)判断响应是否被缓存，如果当前命令的请求缓存功能已经开启，并且本次请求的响应已经存在于缓存中，那么就会立即返回一个包含缓存响应的Observable对象。

(4)判断断路器是否打开，当命令执行前，Hystrix会检查断路器是否被打开。
如果断路器被打开，那么Hystrix就不会再执行这个命令，而是获取fallback方法，并执行fallback逻辑，也就是下面的第八步。

如果断路器关闭，那么将进入第5步，检查是否有可用资源来执行任务。
断路器的具体细节，我们后面再讲。

(5)判断线程池/请求队列/信号量是否已满？如果与命令相关的线程池和请求队列（或者信号量，不使用线程池的时候）已经被占满，那么Hystrix也不会执行命令，而是立即跳到第八步,执行fallback逻辑。

这里Hystrix判断的线程池不是容器的线程池，而是每个依赖服务的专有线程池。Hystrix为了保证不会因为某个依赖服务的问题影响到其他服务采用了“舱壁模式”(Bulkhead Pattern)来隔离每个依赖的服务，后面会细说。

(6)Hystrix会根据我们编写的方法来决定采用什么方式去请求依赖服务

HystrixCommand.run():返回单一结果，或者抛出异常
HystrixObservableCommand.construct()：返回一个Observable对象来发射多个结果，或通过onError发送错误通知。

如果run()或者construct()方法的执行时间超过了命令设置的超时阀值，当前处理线程将会抛出一个TimeoutException(如果该命令不在其自身的线程中执行，会通过单独的计时线程来抛出)。在这种情况下，Hystrix会转到fallback处理逻辑，同时，如果当前命令没有取消或者中断，那么它最终会忽略run()或者construct()方法的返回。
如果命令没有出现异常并返回了结果，那么Hystrix在记录一些日志并采集监控报告之后将该结果返回，在使用run的情况下，Hystrix会返回一个Observable，它发射单个结果并产生onCompleted的结束通知，而使用construct()的情况下，会直接返回该方法产生的Observable。

(7)计算断路器的健康
Hystrix会将"成功"、“失败”、“拒绝”、“超时"等信息报告给断路器，断路器会维护一组计数器来统计这些数据。
断路器会使用这些统计数据来决定是否要将断路器打开，来对某个依赖服务的请求进行"熔断/短路”,直到恢复期结束，如果在恢复期结束以后，根据统计结果判断还是未达到健康状态，就再次"熔断/短路"。

(8)fallback处理。当命令执行失败的时候，Hystrix会进入fallback尝试回退处理，我们通常也称该操作为"服务降级"，上面的第4、5、6步都能引起服务降级。

在服务降级逻辑中，我们需要实现一个通用的响应结果，并且该结果的处理逻辑应该是
从缓存或是根据一些静态逻辑来获取，而不是依赖网络请求获取。如果一定要在降级逻辑中包含网络请求，那么该请求也必须被包装在HystrixCommand或HystrixObservableCommand中，从而形成级联的降级策略，而最终的降级逻辑一定不是一个依赖网络请求的处理，而是一个能够稳定返回结果的处理逻辑。

当使用HystrixCommand的时候，通过实现HystrixCommand.getFallback()来实现服务降级逻辑。
当使用HystrixObservableCommand的时候，通过HystrixObservableCommand.resumeWithFallback()实现服务降级逻辑，该方法会返回一个Observable对象来发射一个或多个降级结果。

当命令的降级逻辑返回结果后，Hystrix就将该结果返回给调用者，当使用HystrixCommand.getFallback()的时候，会返回一个Observable对象，该对象会发射getFallback()的处理结果。当使用HystrixObservableCommand.resumeWithFallback()实现的时候，会将Observable对象直接返回。

如果我们没有为命令实现降级逻辑或降级逻辑中出现了异常，Hystrix依然会返回一个Observable对象，但是它不会发射任何结果数据。而是通过onError方法通知命令立即中断请求，并通过onError()方法将引起命令失败的异常发送给调用者。

我们应该在实现降级策略的时候尽可能避免失败的情况。如果降级执行发生失败，Hystrix会根据不同的执行方法做出不同的处理

execute():抛出异常
queue():正常返回Future对象，但是当调用get()来获取结果时会抛出异常。
observe():正常返回Observable对象，当订阅它的时候，将立即通过调用订阅者的onError方法来通知中止请求。
toObservable():正常返回Observable对象，当订阅它的时候，将通过调用订阅者的onError方法来通知中止请求。

(9)返回成功的响应。当Hystrix命令执行成功后，它会将处理结果直接返回或是以Observable的形式返回。具体以哪种方式返回取决于第二步的4种执行方式。

toObservable():返回最原始的Observable，必须通过订阅它才会真正触发命令的执行流程
observe():在toObservable()产生原始Observable之后立即订阅它，让命令能够马上开始异步执行，并返回一个Observable对象，当调用它的subscribe时，将重新产生结果和通知给订阅者。
queue():将toObservable()产生的原始Observable通过toBlocking()方法转换成BlockingObservable对象，并调用它的toFuture()方法返回异步的Future对象。
execute():在queue()返回异步的Future对象之后，通过调用get()方法阻塞并等待结果的返回。

断路器原理

断路器在HystrixCommand或HystrixObservableCommand执行过程中起到了很重要的作用，它是Hystrix的核心部件。
我们先来看看断路器的接口HystrixCircuitBreaker。

public interface HystrixCircuitBreaker {public boolean allowRequest();public boolean isOpen();void markSuccess();public static class Factory {.........}static class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {.........}static class NoOpCircuitBreaker implements HystrixCircuitBreaker {.........}}

该接口中定义的有三个抽象方法

//每个Hystrix的命令都通过它判断是否被执行
public boolean allowRequest();//当前断路器是否打开
public boolean isOpen();//用来闭合断路器
void markSuccess();

另外还有三个静态类

(1)静态类Factory中维护了一个Hystrix命令和HystrixCircuitBreaker的关系集合，其中String类型的key通过HystrixCommandKey定义，每一个Hystrix命令需要有一个key来标识，同时一个Hystrix命令也会在该集合中找到它对应的断路器HystrixCircuitBreaker实例。

public static class Factory {private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap<String, HystrixCircuitBreaker>();public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name());if (previouslyCached != null) {return previouslyCached;}HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreakerImpl(key, group, properties, metrics));if (cbForCommand == null) {return circuitBreakersByCommand.get(key.name());} else {return cbForCommand;}}public static HystrixCircuitBreaker getInstance(HystrixCommandKey key) {return circuitBreakersByCommand.get(key.name());}static void reset() {circuitBreakersByCommand.clear();}}

(2)静态类NoOpCircuitBreaker做了一个很简单的断路器实现，允许所有的请求，并且断路器状态始终闭合。

static class NoOpCircuitBreaker implements HystrixCircuitBreaker {@Overridepublic boolean allowRequest() {return true;}@Overridepublic boolean isOpen() {return false;}@Overridepublic void markSuccess() {}}

（3）静态类HystrixCircuitBreakerImpl是断路器HystrixCircuitBreaker的实现类，该类中定义了断路器的四个核心对象。

HystrixCommandProperties properties：断路器对应HystrixCommand实例的属性对象。
HystrixCommandMetrics metrics：用来让HystrixCommand记录各类度量指标的对象。
AtomicBoolean circuitOpen：断路器是否打开的标志，默认为false。
AtomicLong circuitOpenedOrLastTestedTime：断路器打开或者是上一次测试的时间戳。

     private final HystrixCommandProperties properties;private final HystrixCommandMetrics metrics;private AtomicBoolean circuitOpen = new AtomicBoolean(false);private AtomicLong circuitOpenedOrLastTestedTime = new AtomicLong();

HystrixCircuitBreakerImpl对断路器接口的实现方法如下。

isOpen():判断断路器的打开关闭状态。如果断路器打开标识为true，则直接返回true，表示断路器处于打开状态。否则，就从度量指标对象metrics获取HealthCounts统计对象做进一步判断（该对象记录了一个滚动时间窗内的请求信息快照，默认时间窗为10s）

如果它的请求总数（QPS）在预设的阀值范围内就返回false，表示断路器处于未打开状态，该阀值的配置参数为circuitBreakerRequestVolumeThreshold，默认为20.

如果错误百分比在阀值范围内就返回false，表示断路器处于未打开状态。该阀值的配置参数为circuitBreakerErrorThresholdPercentage，默认为50.

如果上面两个条件都不满足，则将断路器设置为打开状态(熔断)，同时，如果是从关闭状态切换到打开状态的话，就将当前时间记录到circuitOpenedOrLastTestedTime对象中。

 public boolean isOpen() {if (circuitOpen.get()) {return true;}HealthCounts health = metrics.getHealthCounts();if (health.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {return false;}if (health.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {return false;} else {if (circuitOpen.compareAndSet(false, true)) {circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());return true;} else {return true;}}}

allowRequest()：判断请求是否被允许。先根据配置对象properties中的断路器判断强制打开或关闭属性是否被设置。如果强制打开被设置，就直接返回false，拒绝请求。如果强制关闭被设置，它会允许所有请求，但是同时也会调用isOpen()来执行断路器的计算逻辑。默认情况下，断路器并不会进入这两个分支，而是通过!isOpen() || allowSingleTest()来判断是否允许请求访问，

public boolean allowRequest() {if (properties.circuitBreakerForceOpen().get()) {// properties have asked us to force the circuit open so we will allow NO requestsreturn false;}if (properties.circuitBreakerForceClosed().get()) {// we still want to allow isOpen() to perform it's calculations so we simulate normal behaviorisOpen();// properties have asked us to ignore errors so we will ignore the results of isOpen and just allow all traffic throughreturn true;}return !isOpen() || allowSingleTest();}

我们看下allowSingleTest()方法。

allowSingleTest()：先获取当断路器从闭合到打开时记录的时间戳，当断路器在打开的状态的时候，这里会判断断开时的时间戳+配置中的circuitBreakerSleepWindowInMilliseconds时间是否小于当前时间，是的话，就将当前时间更新到记录断路器打开的时间对象circuitOpenedOrLastTestedTime中，并且允许此次请求。简单的说，通过circuitBreakerSleepWindowInMilliseconds属性设置了一个断路器打开之后的休眠时间(默认为5秒)，在该休眠时间到达之后，将再次允许请求尝试访问，此时断路器处于半开状态，若此时请求继续失败，断路器又进入打开状态，并继续等待下一个休眠窗口过去之后再次尝试，若请求成功，则将断路器重新置于关闭状态。
所以通过!isOpen() || allowSingleTest()配合，实现了打开关闭的切换。

public boolean allowSingleTest() {long timeCircuitOpenedOrWasLastTested = circuitOpenedOrLastTestedTime.get();// 1) if the circuit is open// 2) and it's been longer than 'sleepWindow' since we opened the circuitif (circuitOpen.get() && System.currentTimeMillis() > timeCircuitOpenedOrWasLastTested + properties.circuitBreakerSleepWindowInMilliseconds().get()) {// We push the 'circuitOpenedTime' ahead by 'sleepWindow' since we have allowed one request to try.// If it succeeds the circuit will be closed, otherwise another singleTest will be allowed at the end of the 'sleepWindow'.if (circuitOpenedOrLastTestedTime.compareAndSet(timeCircuitOpenedOrWasLastTested, System.currentTimeMillis())) {// if this returns true that means we set the time so we'll return true to allow the singleTest// if it returned false it means another thread raced us and allowed the singleTest before we didreturn true;}}return false;}

markSuccess()：该函数用来在"半开"的状态时使用，若Hystrix命令调用成功，通过调用它将打开的断路器关闭，并重置度量指标对象。

   public void markSuccess() {if (circuitOpen.get()) {if (circuitOpen.compareAndSet(true, false)) {//win the thread race to reset metrics//Unsubscribe from the current stream to reset the health counts stream.  This only affects the health counts view,//and all other metric consumers are unaffected by the resetmetrics.resetStream();}}}

下面是官方给的流程图

依赖隔离

Hystrix通过线程池来实现依赖隔离，它会为每一个依赖服务创建一个独立的线程池，这样就算某个依赖服务出现延迟过高的情况，也只是对该依赖服务的调用产生影响，而不会出现耗尽所有的线程资源，从而导致整个系统崩溃的情况。

实现对依赖服务的线程池隔离，有如下优势：

应用自身得到完全保护，不会受不可控的依赖服务影响，即便给依赖服务的线程池被占尽，也不会影响应用的其他服务。
可以有效降低接入新服务的风险。如果接入新服务后出现问题，也不会对其他服务请求产生影响。
当依赖的服务从失效恢复正常后，它的线程会被清理并且能够马上恢复健康的服务，相比之下，容器级别的清理恢复速度要慢得多。
当依赖的服务出现配置错误的时候，线程池会快速反映出此问题（通过失败次数、延迟、超时、拒绝等指标的增加情况）。当依赖的服务因实现机制等原因造成其性能出现很大变化的时候，线程池的监控指标信息会反映出这样的变化。

总之，通过对依赖服务实现线程隔离，可以让我们的应用更加健壮，不会因为个别依赖服务出现问题而引起其他服务的异常。

很多使用者可能会担心为每一个依赖服务都分配一个线程池是否会过多地增加系统地负载。Netflix在设计此系统时，决定接受此开销的成本，以换取它所提供的好处，并认为它很小，不会对成本或性能产生重大影响。

但是对于小部分延迟本身就比较小的请求，这些开销可能是承受不起的。Hystrix为此设计了：信号量。

在Hystrix中除了可使用线程池之外，还可以使用信号量来控制单个依赖服务的并发度，信号量的开销远比线程池小，但是它不能设置超时和实现异步访问。所以，只有在依赖服务是足够可靠的情况下才使用信号量。在HystrixCommand和HystrixObservableCommand中有两处支持信号量的使用。

命令执行：如果将隔离策略参数设置为execution.isolation.strategy设置为SEMAPHORE ，Hystrix会使用信号量代替线程池来控制依赖服务的并发。
降级逻辑：当Hystrix尝试降级逻辑时，它会在调用线程中使用信号量。

信号量仅访问内存数据性能可以达到5000rps(每秒请求数)，信号量仅为1或2，但默认值为10，我们可以按照该标准来设置信号量。

Hystrix使用详解

创建请求命令
前面我们的入门案例中，使用了@HystrixCommand注解来创建请求命令。HystrixCommand用来封装具体的依赖服务调用。接下来我们详细的讲一下。

（1）HystrixCommand可以通过继承的方式实现。并且我们的command必须要实现至少一个构造方法。因为Hystrix要求你构造方法中，必须指明command的一些附属配置。不论我们实现哪个构造方法，最后都是调用AbstractCommand中的构造方法。

public class ConsumerCommand extends HystrixCommand<String> {private RestTemplate restTemplate;private String msg;public ConsumerCommand(RestTemplate restTemplate,String msg) {//这个我们需要传入组名super(HystrixCommandGroupKey.Factory.asKey("ExampleConsumer"));this.restTemplate=restTemplate;this.msg=msg;}@Overrideprotected String run() throws Exception {return restTemplate.getForObject("http://eureka-client01/hello/"+msg,String.class);}
}

    @RequestMapping("/test/{msg}")public  String  test(@PathVariable String msg){ConsumerCommand consumerCommand=new ConsumerCommand(restTemplate,msg);String result = null;try {//同步执行//result=consumerCommand.execute();//异步执行，可以通过Future的get方法获取返回结果Future<String> queue = consumerCommand.queue();result = queue.get();} catch (Exception e) {e.printStackTrace();}return  result;}

（2）HystrixCommand也可以通过我们前面入门案例中的注解来更优雅地实现。

 @HystrixCommand@RequestMapping("/test/{msg}")public  String  test(@PathVariable String msg){return restTemplate.getForObject("http://eureka-client01/hello/"+msg,String.class);}

但是仅仅通过这个注解实现地方法只会同步执行，如果想要异步执行则还需要另外定义。AsyncResult可以被转换成Futrue对象，我们直接调用get方法返回其中的结果，但是这里虽然没有要求我们实现get方法，但我们必须实现，因为它的默认实现是抛出一个异常。

    @HystrixCommand@RequestMapping("/test/{msg}")public  String  test(@PathVariable String msg){return new AsyncResult<String>(){@Overridepublic String invoke() {return restTemplate.getForObject("http://eureka-client01/hello/"+msg,String.class);}@Overridepublic String get() {return invoke();}}.get();}

除了同步异步执行之外，我们还可以通过Observable来实现响应式编程。
我们先来看observe()方法，我们前面讲过，observe()返回的是一个Hot Observable，无论有没有订阅会在observe()被调用的时候立即执行，每次被订阅都会再次执行。
那我们来测试一下是不是这样。

首先我们改造一下服务提供者，打印一条信息，以便看到它被调用。

    @RequestMapping("/hello/{msg}")public String hello(@PathVariable String msg){//打印一条信息System.out.println(msg);return "hello,SpringCloud"+msg+":"+request.getServerPort();}

然后我们使用observe()创建Observable对象，但是不订阅，ConsumerCommand类不变。

    @RequestMapping("/test/{msg}")public  void  test(@PathVariable String msg){Observable<String> observe = new ConsumerCommand(restTemplate, msg).observe();}

我们访问接口http://localhost:8888/test/observe
当我们访问这个接口的时候我们会发现，即使没有订阅者，也执行了run方法。

然后呢我们创建一个订阅者来接收事件源发出的信息。我们前面讲过事件源每发出一次事件，都会调用订阅者的onNext方法，并且在结束的时候调用onCompleted方法。

@RequestMapping("/test/{msg}")public  void  test(@PathVariable String msg){Observable<String> observe = new ConsumerCommand(restTemplate, msg).observe();observe.subscribe(new Observer<String>() {@Overridepublic void onCompleted() {System.out.println("end");}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(String s) {System.out.println("订阅者看到的返回结果"+s);}});}

再次访问接口http://localhost:8888/test/observe

接下来我们再来看看返回Cold Observable的toObservable()方法。
我们还是先在没有订阅者的情况下看看是不是和我们前面说的一样。

    @RequestMapping("/test/{msg}")public  void  test(@PathVariable String msg){Observable<String> observe = new ConsumerCommand(restTemplate, msg).toObservable();}

访问http://localhost:8888/test/toobservable
发现空空如也，果然没有执行。

至于添加订阅者后的测试这里就不再演示了，无非是把上面的订阅者代码复制下来再执行一次。

(3) 前面的HystrixCommand的observe()和toObservable()有个局限性就是只能返回单个操作结果，这个前面我们有讲到。而HystrixObservableCommand则能返回多个操作结果的Observable对象。
我们先来通过继承的方式来实现HystrixObservableCommand，使用HystrixObservableCommand最后都是调用的construct方法。

public class ConsumerObserableCommand extends HystrixObservableCommand {private RestTemplate restTemplate;private String[] ids;public ConsumerObserableCommand(RestTemplate restTemplate, String[] ids) {super(HystrixCommandGroupKey.Factory.asKey("ExampleConsumer"));this.restTemplate=restTemplate;this.ids=ids;}@Overrideprotected Observable construct() {//这里用的全都是RxJava的东西,创建一个Observable对象return Observable.create(new Observable.OnSubscribe<String>(){@Overridepublic void call(Subscriber<? super String> subscriber) {try {// 写业务逻辑，注意try-catch//判断是否无人订阅if (!subscriber.isUnsubscribed()) {//执行多次for (String id : ids) {String s = restTemplate.getForObject("http://eureka-client01/hello/" + id, String.class);subscriber.onNext(s);}subscriber.onCompleted();}} catch (Exception e) {subscriber.onError(e);}}});}
}

编写订阅者

    @RequestMapping("/test/{msg}")public  void  test(@PathVariable String msg){Observable<String> observe = new ConsumerObserableCommand(restTemplate, new String[]{"1","2","3","4","5'"}).observe();observe.subscribe(new Observer<String>() {@Overridepublic void onCompleted() {System.out.println("end");}@Overridepublic void onError(Throwable e) {}@Overridepublic void onNext(String s) {System.out.println("订阅者看到的返回结果"+s);}});}

访问结果看到如下返回结果。而这个正是HystrixCommand无法做到的。

而此类的注解实现依然是@HystrixCommand，但是需要一些变化。
大致和上面的相同。
ObservableExecutionMode.EAGER表示使用observe().
ObservableExecutionMode.LAZY表示使用toObservable()

 @HystrixCommand(observableExecutionMode = ObservableExecutionMode.EAGER)@RequestMapping("/test/{ids}")public  Observable<String>  test(@PathVariable String[] ids){return Observable.create(new Observable.OnSubscribe<String>(){@Overridepublic void call(Subscriber<? super String> subscriber) {try {// 写业务逻辑，注意try-catchif (!subscriber.isUnsubscribed()) {for (String id : ids) {String s = restTemplate.getForObject("http://eureka-client01/hello/" + id, String.class);subscriber.onNext(s);}subscriber.onCompleted();}} catch (Exception e) {subscriber.onError(e);}}});}

定义服务降级
fallback是Hystrix命令执行失败后使用的方法，用来实现服务的降级处理逻辑。
在HystrixCommand中可以通过重载getFallback()方法来实现服务降级逻辑，Hystrix会在run()执行过程中出现错误、超时、线程池拒绝、断路器熔断的情况下，执行该方法。

在HystrixObservableCommand中可以通过重载resumeWithFallback方法来实现服务降级逻辑，该方法会返回一个Observable对象，当命令执行失败,Hystrix会将Observable的结果通知给订阅者。

当然我们也可以直接使用注解。

异常处理

在HystrixCommand实现的run方法中抛出异常时，除了HystrixBadRequestException之外，其他异常均会被认为命令执行失败并触发服务降级的处理逻辑，所以如果我们希望不触发服务降级时就要抛出HystrixBadRequestException。

当我们使用注解时，可以配置参数忽略指定异常。

 @HystrixCommand(ignoreExceptions = {MyException.class})

原理是当我们的方法抛出了MyException异常，Hystrix会将其包装在HystrixBadRequestException中抛出。

当Hystrix命令因为异常进入服务降级逻辑之后，往往需要对不同异常做不同的处理。
在以继承方式实现的HystrixCommand中，可以通过在getFallback()方法里调用getExecutionException()方法获取具体的异常。
在注解方式中，只需要在我们的fallback方法的参数中添加一个Throwable e参数即可在方法内部获取异常。

public String testFallback( String msg,Throwable e){return "Hystrix Error";}

命令名称、分组及线程池划分

以继承方式实现的HystrixCommand使用类名作为默认的命令名称。我们可以在构造函数中设置。对于每个命令，GroupKey是必需的，我们前面的例子中在构造函数中就只设置了组名。

 public ConsumerCommand(){super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleConsumer")).andCommandKey(HystrixCommandKey.Factory.asKey("MyKeyName")));}

通过设置组名，Hystrix会根据组来统计命令的告警、仪表盘等信息。为什么一定要设置组名呢？因为Hystrix命令默认的线程划分也是根据组名来实现的，Hystrix会让同一个组名的命令使用同一个线程池。

同时，Hystrix还提供了HystrixThreadPoolKey来对线程池进行划分。

 public ConsumerCommand(){super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleConsumer")).andCommandKey(HystrixCommandKey.Factory.asKey("MyKeyName")).andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("test")));}

当我们通过注解时，可以这样设置。

@HystrixCommand(commandKey = "testKey",groupKey = "testGroup",threadPoolKey = "test")

请求缓存

Hystrix提供了请求缓存的功能，我们可以开启请求缓存，达到减轻高并发时的请求线程消耗、降低请求响应时间的效果。

在继承方式实现Hystrix命令时，我们只需在HystrixCommand或HystrixObservableCommand中重写getCacheKey()方法即可开启请求缓存。当不同的外部请求调用同一个依赖服务时，Hystrix会根据getCacheKey返回值来区分是否是重复的请求，如果getCacheKey返回值相同，那么该依赖服务只会在第一个请求时被调用一次，另外的请求直接从缓存中返回结果。

@Overrideprotected String getCacheKey(){return "1";}

其实这时候如果直接调用的话，会报错，说我们需要初始化Hystrix请求上下文。
通常这个初始化请求上下文的操作要在过滤器filter中完成。
我在测试的时候一直纳闷为什么缓存不生效，后来才忽然领悟每一次调用方法都会重新初始化上下文。
Hystrix的请求缓存是这样的，在一次请求上下文中，也就是一次请求中，然后多次调用同一个依赖服务，才会有缓存的效果。所以我觉得这个功能不怎么实用。

HystrixRequestContext.initializeContext();

如果我们的请求命令中有写的操作，那么缓存中的数据就需要我们在进行写操作时进行处理，以防止后面的读操作获取到了失效的数据。
可以通过HystrixRequestCache的clear方法清除缓存。第一个参数为命令名称,clear的参数为缓存时的key。我们上面返回的缓存key为1，我们就对缓存key为1的缓存内容进行清理。

        HystrixRequestCache.getInstance(HystrixCommandKey.Factory.asKey("commandKey"), HystrixConcurrencyStrategyDefault.getInstance()).clear("1");

我们来看一下缓存的原理，我们来看一下AbstractCommand这个类。

 protected String getCacheKey() {return null;}protected boolean isRequestCachingEnabled() {return properties.requestCacheEnabled().get() && getCacheKey() != null;}public Observable<R> toObservable() {final boolean requestCacheEnabled = isRequestCachingEnabled();final String cacheKey = getCacheKey();/* 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);Observable<R> afterCache;// put in cacheif (requestCacheEnabled && cacheKey != null) {// wrap it for cachingHystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.putIfAbsent(cacheKey, toCache);if (fromCache != null) {// another thread beat us so we'll use the cached value insteadtoCache.unsubscribe();isResponseFromCache = true;return handleRequestCacheHitAndEmitValues(fromCache, _cmd);} else {// we just created an ObservableCommand so we cast and return itafterCache = toCache.toObservable();}} else {afterCache = hystrixObservable;}
}

我们可以看到getCacheKey默认返回值为null，并且从isRequestCachingEnabled方法的实现逻辑我们知道如果我们不重写getCacheKey方法并返回一个不为null的值，那么缓存是不会开启的。requestCacheEnabled属性默认是为true的。

然后我们看看toObservable中的缓存的执行步骤。首先会根据requestCacheEnabled判断是否开启了请求缓存，如果开启了并且getCacheKey返回了一个非空值，那么就使用getCacheKey方法返回的key值去调用HystrixRequestCache的get方法去获取缓存中的对象。
接下来也是判断是否开启了请求缓存，如果开启了并且getCacheKey返回了一个非空值，就将hystrixObservable对象包装成缓存结果的HystrixCachedObservable对象，然后将其放入当前命令的缓存结果中。

Hystrix提供了三个注解支持请求缓存。

@CacheResult：该注解用来标记请求命令返回的结果应该被缓存，必须与@HystrixCommand搭配使用。属性有cacheKeyMethod。
@CacheRemove：该注解用来让请求命令的缓存失效，失效的缓存根据定义的key决定。属性有cacheKeyMethod和commandKey。
@CacheKey：该注解用来在请求命令的参数上标记，使其作为缓存的key值，如果没有标注则会使用所有参数。如果同时还使用了@CacheResult和@CacheRemove注解的cacheKeyMethod属性指定了缓存key的生成方法，该注解将失效。

看下面的例子。

 //这样就开启了缓存，当User对象返回时，会将结果置入请求缓存中。//因为缓存key会使用所有的参数，所以这里的缓存key即为id。@HystrixCommand@CacheResultpublic User getUsserById(Long id){return restTemplate.getForObject("http://eureka-client01/getUser/{1}",User.class,id);}

 //我们可以通过cacheKeyMethod 指定生成缓存key的方法@HystrixCommand@CacheResult(cacheKeyMethod = "getCacheKey")public User getUsserById(Long id){return restTemplate.getForObject("http://eureka-client01/getUser/{1}",User.class,id);}public Long getCacheKey(Long id){return id;}

 //也可以通过CacheKey指定缓存key//CacheKey还允许对象的内部参数作为缓存key，例如参数为User对象，我们可以指定其中的id为缓存key@HystrixCommand@CacheResultpublic User getUsserById(@CacheKey("id") Long id){return restTemplate.getForObject("http://eureka-client01/getUser/{1}",User.class,id);}

 //我们可以通过CacheRemove在写方法上清除缓存，防止造成读数据有误。@CacheRemove(commandKey ="getUsserById" )@HystrixCommandpublic String update(@CacheKey("id") User user){return "更新写入操作";}

Hystrix Dashboard仪表盘

在断路器原理的介绍中，我们提到关于请求命令的度量指标的判断。这些度量指标都是命令执行过程中重要信息，除了在断路器中使用之外，对系统运维也有很大帮助。这些指标信息会以滚动时间窗和桶结合的方式进行汇总，并在内存中驻留一段时间，供查询使用。Hystrix仪表盘就可以用来查询这些信息。Hystrix Dashboard用来实时监控Hystrix的各项指标信息，我们来搭建一下。

我们先把我们的服务注册中心、服务提供者、服务消费者启动起来。

(1)首先我们要创建一个项目，命名为hystrix-dashboard。添加依赖。

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-hystrix</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-hystrix-dashboard</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

（2）为启动类添加@EnableHystrixDashboard注解，启用仪表盘功能。

@EnableHystrixDashboard
@SpringBootApplication
public class HystrixDashboardApplication {public static void main(String[] args) {SpringApplication.run(HystrixDashboardApplication.class, args);}}

（3）第三步，修改配置文件，指定端口号等。

spring.application.name=hystrix-dashboard
server.port=2001

（4）然后启动项目，访问http://localhost:2001/hystrix
通过页面的英文我们知道，Hystrix提供了三种监控方式。

默认的集群监控
指定的集群监控
对单体应用的监控

对集群的监控需要整合Turbine才能实现，也很简单，可以自己百度去进行尝试。
这里我们实现对单个服务的监控。

（5）修改服务消费者的pom文件，添加依赖。

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency>

（6）修改启动类，一定要注入这个Bean，并且确保启用了EnableCircuitBreaker注解。

@EnableCircuitBreaker
@EnableEurekaClient
@SpringBootApplication
public class Application {public static void main(String[] args) {SpringApplication.run(Application.class);}@Bean@LoadBalancedRestTemplate restTemplate(){return new RestTemplate();}@Beanpublic ServletRegistrationBean getServlet(){HystrixMetricsStreamServlet streamServlet = new HystrixMetricsStreamServlet();ServletRegistrationBean registrationBean = new ServletRegistrationBean(streamServlet);registrationBean.setLoadOnStartup(1);registrationBean.addUrlMappings("/actuator/hystrix.stream");registrationBean.setName("HystrixMetricsStreamServlet");return registrationBean;}
}

(7)我们在Hystrix Dashboard首页输入http://localhost:8888/actuator/hystrix.stream
点击Monitor Stream，我们会发现一直在Loadding，这时候我们访问服务消费者。可以看到如下画面。

（8）首页中的Delay参数用来控制服务器上轮询监控信息的延迟时间，默认2000毫秒。title用来配置仪表盘的标题，默认为访问的url。
仪表盘界面有绿、黄、蓝、红等颜色的0，用来对请求成功、延迟、失败的请求进行计数。还有一个实心圆，流量越大该实心圆越大。

Hystrix还有许多参数的配置可以去Api中查看，这里就不再一一细说。
Hystrix里面有很多我似懂非懂的地方，主要集中在RxJava和线程隔离方面。
主要的思路都是跟着翟永超大神的书学下来的，当然它的版本比较老旧，有的方面与现在最新版本是不合的。

参考书籍：翟永超《SpringCloud微服务实战》

Spring Cloud从入门到精通（四）：熔断器 Hystrix相关推荐

浅谈服务治理、微服务与Service Mesh（二） Spring Cloud从入门到精通到放弃
引言作为本系列文章的第二篇,本文主要为大家介绍下微服务概念中非常火热的Spring Cloud开发框架.由于网上关于Spring Cloud的文章多如牛毛,为了让大家阅读后能有不一样的收获,因此本文 ...
Spring Boot+Spring Cloud基础入门（五）断路器——Hystrix
Hystrix Hystrix翻译成中文是"豪猪",豪猪周身长满了刺,能保护自己不受天敌的伤害,代表了一种防御机制,这与Hystrix本身的功能不谋而合,因此Netflix团队将该 ...
Spring Boot从入门到精通（超详细）
Spring Boot从入门到精通(超详细) _kayden_ 2020-07-20 15:19:22 9491 正在上传-重新上传取消收藏 184 分类专栏: springboot 文章标签: ...
Spring Cloud Eureka 入门（三）服务消费者详解
2019独角兽企业重金招聘Python工程师标准>>> 摘要: 原创出处:www.bysocket.com 泥瓦匠BYSocket 希望转载,保留摘要,谢谢! "真正的进步 ...
Spring Cloud Eureka 入门（二）服务提供者详解
2019独角兽企业重金招聘Python工程师标准>>> 摘要: 原创出处:www.bysocket.com 泥瓦匠BYSocket 希望转载,保留摘要,谢谢! "优秀不是过 ...
微服务技术方案：Spring Cloud 从入门到实战
随着互联网技术的发展与不断创新,以及用户流量的不断增大,越来越多的企业项目面临大数据.高并发等问题,随之而来的就是通过分布式模型组建架构,微服务思想就集中体现了应用价值,2020 年的你还没有掌握微服 ...
Spring cloud eureka 入门使用及指导说明（单机篇）
Spring cloud eureka 入门使用及指导说明(单机篇) 基本概念和方案 Eureka是基于REST(Representational State Transfer,代表性状态传输)的服务 ...
Spring Cloud 微服务开发：入门、进阶与源码剖析 —— 10.2 Spring Cloud Sleuth 入门
10.2 Spring Cloud Sleuth 入门 10.2.1 Spring Cloud Sleuth 入门案例先简单介绍一下案例目标,在本案例中,我们将会使用Feign.RestTempla ...
C4D致富经典入门到精通(四)
C4D中的Nurbs建模 C4D基础界面的介绍与常用快捷键: C4D致富经典入门到精通(一) C4D父子关系的理解与创建参数几何体与可编辑对象: C4D致富经典入门到精通(二) C4D样条曲线创建 ...
视频教程-Spring框架快速入门到精通-Java
Spring框架快速入门到精通十年项目开发经验,主要从事java相关的开发,熟悉各种mvc开发框架. 王振伟 ¥18.00 立即订阅扫码下载「CSDN程序员学院APP」,1000+技术好课免费看 ...

Spring Cloud从入门到精通（四）：熔断器 Hystrix