Spring Cloud Hystrix 断路器

简介

Hystrix 具有服务降级、服务熔断、线程和信号隔离、请求缓存、请求合并以及服务监控等强大功能。

为项目引入Spring Cloud Hystrix

1.pom.xml配置：

<dependency>
	<groupId>org.springframework.cloud</groupId>
	<artifactId>spring-cloud-starter-hystrix</artifactId>
</dependency>

2 添加注解，开启断路器

@EnableCircuitBreaker

注意：@SpringCloudApplication注解

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

3.在方法添加@HystrixCommand

@HystrixCommand(fallbackMethod = "helloFallback")
public String helloConsumer() {
    long start = System.currentTimeMillis();
    String result = restTemplate.getForEntity("http://EUREKA-CLIENT/demo/test", String.class).getBody();
    long end = System.currentTimeMillis();
    logger.info(end - start);
    return result;
}

public String  helloFallback() {
    return "error";
}

原理分析

1.工作流程

当你发出请求后，hystrix是这么运行的

Spring Cloud Hystrix 断路器-LMLPHP
详细解释个步骤

step1: 创建 HystrixCommand or HystrixObservableCommand Object

HystrixCommand 用于返回单一的响应
HystrixObservableCommand 用于返回多个可自定义的响应

step2:Execute the Command（命令执行）

对于HystrixCommand有4个执行方法
对于HystrixObservableCommand只有后两个

HystrixCommand:

//同步阻塞方法，其实就是调用了queue().get()

execute() — blocks, then returns the single response received from the dependency (or throws an exception in case of an error)

//异步非阻塞方法，直接返回Future，可以先做自己的事情，做完再.get()

queue() — returns a Future with which you can obtain the single response from the dependency

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

HystrixObservableCommand:
//热观察（Hot Observable），可以被立即执行，如果订阅了那么会重新通知，其实就是调用了toObservable()并内置ReplaySubject，详细可以参考RxJava

observe() — subscribes to the Observable that represents the response(s) from the dependency and returns an Observable that replicates that source Observable

//冷观察(Cold Observable)，返回一个Observable对象，当调用此接口，还需要自己加入订阅者，才能接受到信息，详细可以参考RxJava

toObservable() — returns an Observable that, when you subscribe to it, will execute the Hystrix command and emit its responses

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

注：由于Hystrix底层采用了RxJava框架开发，所以没接触过的可能会一脸懵逼，需要再去对RxJava有所了解。

step3. 结果是否被缓存

若当前命令的请求缓存功能是被启动的，并且该命令缓存命中，那么缓存的结果会立即以Observable对象的形式返回。

step4. 断路器是否打开

在命令结果没有缓存命中的时候，Hystrix在执行命令前需要检查断路器是否为打开状态

如果断路器是打开的，那么Hystrix不会执行命令，而是转接到fallback处理逻辑（对应下面的第8步）
如果断路器是关闭的，那么Hystrix调到第5步，检查是否有可用资源来执行命令。

step5.线程池/请求队列/信号量是否已经占满

如果与命令相关的线程池和请求队列，或者信号量（不适用线程池的时候）已经被占满，那么Hystrix也不会执行命令，而是转接到fallback处理逻辑（对应下面的第8步）

需要注意的是，这里的Hystrix所判断的线程池并发容器的线程池，而是每个依赖服务的专有线程池。Hystrix为了保证不会因为某个依赖服务的问题影响到其他依赖服务而采用了“舱壁模式”来隔离每个依赖的服务。

step6.HystrixObservableCommand.construct() or HystrixCommand.run()）

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

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

如果run()或construct()的执行时间超过了命令设置的超时阈值，当前处理线程将会抛出一个TimeoutException(如果该命令不在其自身的线程中执行，则会通过单独的计时线程来抛出)。在这种情况下，Hystrix会转接到fallback处理逻辑（对应下面的第8步）。同时，如果当前命令没有被取消或中断，那么它最终会忽略run()或者construct()方法的返回。

如果命令没有抛出异常并返回了结果，那么Hystrix在记录一些日志并采集监控报告之后将该结果返回。在使用run()的情况下，Hystrix会返回一个Observable，它发射单个结果并产生onCompleted的结束通知；而在使用construct()的情况下，Hystrix会直接返回该方法产生的Observable对象。

step7.计算断路器的健康度

Hystrix会将成功、失败、拒绝、超时等信息报告给断路器，而断路器会维护一组计数器来统计这些数据

断路器使用这些统计数据来决定是否要将断路器打开，来对某个依赖服务的请求进行“熔断/短路”，直到恢复期结束后，根据统计数据判断如果还是未达到健康指标，就再次“熔断/短路”。

step8. fallback处理

当命令执行失败的时候，Hystrix进入fallback尝试回退处理，我们通常也称该操作为**“服务降级”**。而能够引起服务降级的情况有以下几种：

第4步，当前命令处于“熔断/短路”状态，断路器是打开的时候。
第5步，当前命令的线程池、请求队列或信号量被占满的时候。
第6步，HystrixObservableCommand.construct()或HystrixCommand.run()抛出异常的时候。

step9. 成功返回

当Hystrix命令执行成功后，它会将处理结果直接返回或是以Observable的形式返回。而具体以哪种方式返回取决于之前第2步中我们所提到的对命令的4种不同执行方式，下图总结了这4种调用方式之间的依赖关系。
Spring Cloud Hystrix 断路器-LMLPHP

断路器原理

通过ConcurrentHashMap<String, HystrixCircuitBreaker>来维护每一个HystrixCommandKey和它对应的断路器实例HystrixCircuitBreaker。

下面是断路器的详细执行逻辑图
Spring Cloud Hystrix 断路器-LMLPHP

依赖隔离

Hystrix使用“舱壁模式”实现线程池的隔离，它会为每一个依赖服务创建一个独立的线程池，这样就算某个依赖服务出现延迟过高的情况，也只是对该依赖服务的调用产生影响，而不会拖慢其他的依赖服务。

线程池是否会过多的增加系统的负载和开销？Netflix在设计Hystrix的时候，认为线程池上的开销相对于隔离所带来的好处是无法比拟的。同时Netflix也针对线程池的开销做了相关的测试，以用结果打消Hystrix的实现对性能影响的顾虑。

如果觉得在使用线程池时会带来延迟影响，在Hystrix中还可以使用信号量来控制单个依赖服务的并发读，信号量的开销远比线程池的开销小。但是它不能设置超时或实现异步访问。所以，只有在依赖服务是足够可靠的情况下才使用信号量。咋HystrixCommand和HystrixObservableCommand中有两处支持信号量的使用。

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

信号量的默认值为10，我们也可以通过动态刷新配置的方式来控制并发线程的数量。

使用详解

1.创建请求命令

Hystrix命令就是之前所说的HystrixCommand，它用来封装具体的依赖服务调用逻辑

我们可以通过继承的方式实现

import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import org.springframework.web.client.RestTemplate;

public class UserCommand extends HystrixCommand<User> {

    private RestTemplate restTemplate;

    private Long id;
    public UserCommand(Setter setter, RestTemplate restTemplate, Long id) {
        super(setter);
        this.restTemplate = restTemplate;
        this.id = id;
    }

    @Override
    protected User run() throws Exception {
        return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
    }
}

通过上面实现的UserCommand，我们既可以实现请求的同步执行也可以实现请求的异步执行

同步执行 User u = new UserCommand(restTemplate, 1L).execute();
异步执行 Future<User> futureUser = new UserCommand(restTemplate, 1L).queue()；。异步执行的时候，可以通过返回的futureUser 调用get方法获取结果。

另外，也可以通过@HystrixCommand注解来更为优雅的试下Hystrix命令的定义，比如：

import com.example.ribbonconsumer.command.User;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.client.RestTemplate;

public class UserService {

    @Autowired
    private RestTemplate restTemplate;

    @HystrixCommand
    public User getUserById(Long id ) {
        return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
    }
}

虽然@HystrixCommand可以优雅的定义Hystrix命令的实现。但是如上定义的getUserById方式只是同步执行的实现，若要实现异步执行则还需另外定义，比如：

@HystrixCommand
    public Future<User> getUserByIdAsync(final String id) {
        return new AsyncResult<User>() {
            @Override
            public User invoke() {
                return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
            }
        };
    }

除了传统的同步执行与异步执行外，我们还可以将HystrixCommand通过Observable来实现响应式执行方式。通过调用observe()和toObservable()方法可以返回Observable对象，比如：

Observable<User> ho=new UserCommand(com.netflix.hystrix.HystrixCommand.Setter.withGroupKey(
        HystrixCommandGroupKey.Factory.asKey("")),new RestTemplate(),0L).observe();
List<User> list=new ArrayList<>();
//注意：因为执行是异步的，所以要想看到输出结果这里就要阻塞一下
Thread.sleep(3000);
//订阅
ho.subscribe(new Observer<User>() {
    @Override
    public void onCompleted() {

        System.out.println(list.toString());
    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
    }

    @Override
    public void onNext(User user) {
        list.add(user);
    }
});

对于第二种toObservable()返回Observable对象，前面介绍execute()、queue()也都使用了RxJava来实现，并且queue()是通过toObservable()来获得一个Cold Observable，并且通过toBlocking()将该Observable转换成BlockingObservable，它可以把数据以阻塞的方式发出来，而toFuture方法则是把BlockingObservable转换成一个Future，该方法只是创建一个Future返回，并不会阻塞，这使得消费者可以自己决定如何处理异步操作。所以第二种可以这样取得User对象：

Observable<User> co=new UserCommand(com.netflix.hystrix.HystrixCommand.Setter.withGroupKey(
        HystrixCommandGroupKey.Factory.asKey("")),new RestTemplate(),0L).toObservable();
BlockingObservable<User> blockingObservable=co.toBlocking();
Future<User> future=blockingObservable.toFuture();
User user=future.get();//注意捕获异常

observe()和toObservable()虽然都返回了Observable对象，但是observe()返回的是Hot Observable，所以我们用ho作为引用名，该命令会在observe()调用的时候立即执行，当Observable每次被订阅的时候会重放他的行为；而toObservable()返回的是Cold Observable，我们用co作为引用名，toObservable()执行之后，命令不会被立即执行，只有当所有订阅者都订阅它之后才会执行。

虽然HystrixCommand具备了observe()和toObservable()的功能，但是它的实现由一定的局限性，它返回的Observable只能发射一次数据，所以Hystrix还提供了另外一个特殊命令封装HystrixObservableCommand,通过它实现的命令可以获取能发射多次的Observable。

如果使用HystrixObservableCommand来实现命令封装，需要将命令的执行逻辑在construct方法重载，这样，hystrix才能将具体逻辑包装到Observable。如下：

import com.netflix.hystrix.HystrixObservableCommand;
import org.springframework.web.client.RestTemplate;
import rx.Observable;
import rx.Subscriber;

public class UserObservableCommand extends HystrixObservableCommand<User> {

    private RestTemplate restTemplate;

    private Long id;

    public UserObservableCommand(Setter setter,  RestTemplate restTemplate, Long id) {
        super(setter);
        this.restTemplate = restTemplate;
        this.id = id;
    }

    @Override
    protected Observable<User> construct() {
        return Observable.create(new Observable.OnSubscribe<User>() {
            @Override
            public void call(Subscriber<? super User> subscriber) {
                try {
                    if (!subscriber.isUnsubscribed()) {
                        User user = restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
                        subscriber.onNext(user);
                        subscriber.onCompleted();
                    }
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }
        });
    }
}

而对此的注解实现依然是使用@HystrixCommand，只是方法定义需要做一些变化，具体内容与construct()的实现类似，如下：

@HystrixCommand
 public Observable<User> getUserById(final String id) {
     return Observable.create(new Observable.OnSubscribe<User>() {
         @Override
         public void call(Subscriber<? super User> subscriber) {
             try {
                 if (!subscriber.isUnsubscribed()) {
                     User user = restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
                     subscriber.onNext(user);
                     subscriber.onCompleted();
                 }
             } catch (Exception e) {
                 subscriber.onError(e);
             }
         }
     });
 }

在使用@HystrixCommand注解实现响应式命令时，可以通过observableExecutionMode参数来控制时使用observe()还是toObservable()的执行方式，该参数有下面两种设置方式

@HystrixCommand(observableExecutionMode = ObservableExecutionMode.EAGER): EAGER是该参数的模式值，表示使用observe()执行方式
@HystrixCommand(observableExecutionMode = ObservableExecutionMode.LAZY)：表示使用toObservable()方式执行。

定义服务降级

在HystrixCommand中可以通过重载getFallback()方法来实现服务降级逻辑，Hystrix会在run()执行过程中出现错误、超时、线程池拒绝、断路器熔断等情况下，执行getFallback()方法内的逻辑。

import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import org.springframework.web.client.RestTemplate;

public class UserCommand extends HystrixCommand<User> {

    private RestTemplate restTemplate;

    private Long id;
    public UserCommand(Setter setter, RestTemplate restTemplate, Long id) {
        super(setter);
        this.restTemplate = restTemplate;
        this.id = id;
    }

    @Override
    protected User run() throws Exception {
        return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
    }

	@Override
    protected User getFallback() {
        return new User();
    }
}

在HystrixObservableCommand实现的Hystrix命令中，重载resumeWithFallback方法来实现服务降级逻辑。

@Override
    protected Observable<User> resumeWithFallback() {
        return Observable.create(new Observable.OnSubscribe<User>() {
            @Override
            public void call(Subscriber<? super User> subscriber) {
                try {
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onNext(new User());
                        subscriber.onCompleted();
                    }
                } catch (Exception e) {
                    subscriber.onError(e);
                }
            }
        }).subscribeOn(Schedulers.io());
    }

如要通过注解实现服务降级只需要使用@HystrixCommand中的fallbackMethod参数来指定具体的服务降级实现方法，如下：

import com.example.ribbonconsumer.command.User;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import com.netflix.hystrix.contrib.javanica.command.AsyncResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.client.RestTemplate;

import java.util.concurrent.Future;

public class UserService {

    @Autowired
    private RestTemplate restTemplate;

    @HystrixCommand
    public User getUserById(Long id ) {
        return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
    }

    @HystrixCommand(fallbackMethod = "defaultUser")
    public Future<User> getUserByIdAsync(final String id) {
        return new AsyncResult<User>() {
            @Override
            public User invoke() {
                return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
            }
        };
    }

    public User defaultUser() {
        return new User();
    }
}

在使用注解来定义服务降级逻辑时，我们需要将具体的Hystrix命令与fallback实现函数定义在同一个类中，并且fallbackMethod的值必须与实现fallback方法的名字相同。由于必须在同一个类中，所以对于fallback的访问修饰符没有特定的要求，定义为private、protected、public均可。

若defaultUser方法实现的并不是一个稳定的逻辑，它依然可能会发生异常，那么我们也可以为它添加@HystrixCommand注解以生成Hystrix命令，同时使用fallbackMethod来指定服务降级逻辑，比如

import com.example.ribbonconsumer.command.User;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import com.netflix.hystrix.contrib.javanica.command.AsyncResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.client.RestTemplate;

import java.util.concurrent.Future;

public class UserService {

    @Autowired
    private RestTemplate restTemplate;

    @HystrixCommand
    public User getUserById(Long id ) {
        return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
    }

    @HystrixCommand(fallbackMethod = "defaultUser")
    public Future<User> getUserByIdAsync(final String id) {
        return new AsyncResult<User>() {
            @Override
            public User invoke() {
                return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
            }
        };
    }

    @HystrixCommand(fallbackMethod = "defaultUserSec")
    public User defaultUser() {
        // 此处可能有另外一个网络请求来获取，所以也有可能失败
        return new User();
    }

    public User defaultUserSec() {
        return new User();
    }
}

在实际使用时，我们需要为大多数执行过程中可能会失败的Hystrix命令实现服务降级逻辑，但是也有一些情况下不去实现降级逻辑

执行写操作的命令 当Hystrix命令是用来执行写操作而不是返回一些信息的时候，通常情况下这类操作的返回类型是void或是空的Observable，实现服务降级的意义不是很大。当写入操作失败的时候，我们通常只需要通知调用者即可。
执行批处理或离线计算的命令 这时通常这些操作只需要将错误传播给调用者，然后让调用者稍后重试而不是发送给调用者一个静默的降级处理响应。

异常处理

1. 异常传播

在调用服务执行HsytrixCommand实现的run()方法抛出异常时，除HystrixBadRequestException之外，其他异常都会认为是Hystrix命令执行失败并触发服务降级处理逻辑，所以当需要在命令执行中抛出不触发服务降级的异常时来使用它。

而在使用注册配置实现Hystrix命令时，它还支持忽略指定异常类型功能，只需要通过设置@HystrixCommand注解的ignoreException参数，比如：

@HystrixCommand(ignoreExceptions = {BadRequestException.class})
    public User getUserById(Long id ) {
        return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
    }

2.异常获取

注解方式获取异常（除HystrixBadRequestException外）来做针对性处理。只需在fallback实现方法的参数中增加Throwable e对象的定义，这样在方法内部就可以触发服务降级的具体异常内容了，比如：

@HystrixCommand(fallbackMethod = "fallback1")
public User getUserById(Long id ) {
     return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
 }

 User fallback1(Long id, Throwable e) {
     assert "getUserById command failed".equals(e.getMessage());
     return new User();
 }

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

以继承的方式实现Hystrix命令使用类名作为默认的命令名称，我们也可以在构造函数中通过Setter静态类来设置

public CommandHelloWorld(String name) {
    super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
            .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld")));
    this.name = name;
}

先调用了withGroupKey来设置命令组名，然后再调用andCommandKey来设置命令名。为什么这么麻烦？因为在Setter的定义中，只有withGroupKey静态函数可以创建Setter的实例，基于这个原因，所以GroupKey是每个Setter必需的参数，而CommandKey则是可选参数。

通过设置命令组，Hystrix会根据组来组织和统计命令的告警、仪表盘等信息。那么为什么一定要设置命令组呢？因为除了根据组能实现统计外，Hystrix命令默认的线程划分也是根据命令组的。默认情况下，Hystrix会让相同命令组名的命令使用同一个线程池，所以需要在创建Hystrix命令时为其指定命令组名来实现默认的线程池的划分。

如果Hystrix的线程池分配仅仅只能依靠命令组来划分，那么就显得不够灵活了，所以Hystrix还提供了HystrixThreadPoolKey来对线程池进行设置，通过它可以实现更细粒度的线程池的划分。

public CommandHelloWorld(String name) {
    super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
            .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld"))
            .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("HelloWorldPool")));
    this.name = name;
}

如果没有特别指定HystrixThreadPoolKey的情况下，依然会使用命令组的方式来划分线程池。通常情况下，尽量使用HystrixThreadPoolKey的方式来指定线程池的划分，而不是通过组名的默认方式实现划分，因为多个不同的命令可能从业务逻辑上看是属于同一个组，但是往往实现本身需要跟其他命令进行隔离。

当我们使用@HystrixCommand注解的时候，只需要设置commandKey、groupKey以及threadPoolKey属性就可以设置命令名称、分组以及线程划分。

@HystrixCommand(commandKey="HelloWorld",groupKey="ExampleGroup",threadPoolKey="HelloWorldPool")
public CommandHelloWorld(String name) {
    ......
    this.name = name;
}

请求缓存

在高并发的场景之下，Hystrix中提供了请求缓存的功能，可以方便地开启和使用请求缓存来优化系统，达到减轻高并发时请求线程的消耗、降低请求响应时间的效果

1. 开启请求缓存功能

Hystrix请求缓存的使用非常简单，我们只需要在实现HystrixCommand或HystrixObservableCommand时，通过重载getCacheKey()方法来开启请求缓存，比如：


import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import org.springframework.web.client.RestTemplate;

public class UserCommand extends HystrixCommand<User> {

    private RestTemplate restTemplate;

    private Long id;
    public UserCommand(Setter setter, RestTemplate restTemplate, Long id) {
        super(setter);
        this.restTemplate = restTemplate;
        this.id = id;
    }

    @Override
    protected User run() throws Exception {
        return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
    }

    @Override
    protected User getFallback() {
        return new User();
    }

    @Override
    protected String getCacheKey() {
        return String.valueOf(id);
    }
}

在上面的例子中，我们通过在getCacheKey方法中返回的请求缓存key值（使用了传入的获取User对象的id值），就能让该请求命令具备缓存功能。此时，当不同的外部请求处理逻辑调用了同一个依赖服务时，Hystrix会根据getCacheKey方法返回的值来区分是否是重复的请求，如果它们的cacheKey相同，那么该依赖服务只会在第一个请求到达时被真实地调用一次，另外一个请求则是直接从缓存中返回结果。所以通过开启请求缓存可以让我们实现的Hystrix命令具备下面几项好处：

减少重复的请求数，降低依赖服务的并发度
在同一个用户请求的上下文中，相同依赖服务的返回数据始终保持一致
请求缓存在run()和construct()执行之前生效，所以可以有效减少不必要的线程开销

2.清理失效缓存功能

使用请求缓存时，如果只是读操作，那么不需要考虑缓存内容是否正确的问题，但是如果请求命令中还有更新数据的操作，那么缓存中的数据就需要我们在进行写操作时进行及时处理，以防止读操作的请求命令获取到失效的数据。

在Hystrix中，可以通过HystrixRequestCache.clear（）方法来进行缓存的清理。

import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
import com.netflix.hystrix.HystrixRequestCache;
import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategyDefault;
import org.springframework.web.client.RestTemplate;

public class UserGetCommand extends HystrixCommand<User> {

    private RestTemplate restTemplate;

    private static final HystrixCommandKey GETTER_KEY = HystrixCommandKey.Factory.asKey("CommandKey");

    private Long id;
    public UserGetCommand(RestTemplate restTemplate, Long id) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("GetSetGet")).andCommandKey(GETTER_KEY));
        this.restTemplate = restTemplate;
        this.id = id;
    }

    @Override
    protected User run() {
        return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, id);
    }

    @Override
    protected String getCacheKey() {
        return String.valueOf(id);
    }

    public static void flushCache(Long id) {
        // 刷新缓存，根据id进行清理
        HystrixRequestCache.getInstance(GETTER_KEY, HystrixConcurrencyStrategyDefault.getInstance()).clear(String.valueOf(id));
    }
}


import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
import com.netflix.hystrix.HystrixRequestCache;
import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategyDefault;
import org.springframework.web.client.RestTemplate;

public class UserPostCommand extends HystrixCommand<User> {

    private RestTemplate restTemplate;

    private User  user;

    public UserPostCommand(RestTemplate restTemplate, User  user) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("GetSetGet")));
        this.restTemplate = restTemplate;
        this.user = user;
    }

    @Override
    protected User run() {
        User r = restTemplate.postForObject("http://USER_SERVICE/users", user, User.class);
		// 刷新缓存，清理缓存中失效的User
        UserGetCommand.flushCache(Long.valueOf(user.getId()));
		return r;
    }
}

该示例中主要有两个请求命令：UserGetCommand用于根据id获取User对象、而UserPostCommand用于更新User对象。当我们队UserGetCommand命令实现了请求缓存后，那么势必需要为UserPostCommand命令实现缓存的清理，以保证User被更新之后，Hystrix请求缓存中相同缓存Key的结果被移除，这样在下一次获取User的时候不会从缓存中获取到未更新的结果。

3.使用注解实现请求缓存
Spring Cloud Hystrix 断路器-LMLPHP

/**
* 使用注解请求缓存 方式1
 * @CacheResult  标记这是一个缓存方法，结果会被缓存
 */
@CacheResult(cacheKeyMethod = "getCacheKey")
@HystrixCommand(commandKey = "commandKey1")
public Integer openCacheByAnnotation1(Long id){
    //此次结果会被缓存
    return restTemplate.getForObject("http://eureka-service/hystrix/cache", Integer.class);
}

/**
 * 使用注解清除缓存 方式1
 * @CacheRemove 必须指定commandKey才能进行清除指定缓存
 */
@CacheRemove(commandKey = "commandKey1", cacheKeyMethod = "getCacheKey")
@HystrixCommand
public void flushCacheByAnnotation1(Long id){
    LOGGER.info("请求缓存已清空！");
    //这个@CacheRemove注解直接用在更新方法上效果更好
}

/**
 * 第一种方法没有使用@CacheKey注解，而是使用这个方法进行生成cacheKey的替换办法
 * 这里有两点要特别注意：
 * 1、这个方法的入参的类型必须与缓存方法的入参类型相同，如果不同被调用会报这个方法找不到的异常
 * 2、这个方法的返回值一定是String类型
 */
public String getCacheKey(Long id){
    return String.valueOf(id);
}

使用@CacheResult和@CacheKey实现缓存功能

 @CacheResult
 @HystrixCommand(commandKey = "findUserById", groupKey = "UserService", threadPoolKey = "userServiceThreadPool")
 public UserVO findById2(@CacheKey("id") Long id) {
     ResponseEntity<UserVO> user = restTemplate.getForEntity("http://users-service/user?id={id}", UserVO.class, id);
     return user.getBody();
 }

注意： @CacheKey的优先级比cacheKeyMethod的优先级低。

@CacheKey注解除了可以指定方法参数作为缓存key之外，它还允许访问参数对象的内部属性作为缓存Key。比如，下面，它指定了User对象的id属性作为缓存key。

@CacheResult
@HystrixCommand
public User getUserById(@CacheKey("id") User user) {
	return restTemplate.getForObject("http://USER_SERVICE/user/{1}", User.class, user.getId());
}

3.1缓存清理
我们已经通过@CacheResult注解将请求结果置入Hystrix的请求缓存中。若该内容调用了update操作进行了更新，那么此时请求缓存中的结果与实际结果就会产生不一致，所以我们需要在update类型的操作上对失效的缓存进行清理。在Hystrix的注解配置中，可以通过@CacheRemove注解来实现失效缓存的清理。

@CacheRemove(commandKey = "findUserById")
@HystrixCommand(commandKey = "updateUser",groupKey = "UserService",threadPoolKey = "userServiceThreadPool")
public void updateUser(@CacheKey("id")UserVO user){
    restTemplate.postForObject("http://users-service/user",user,UserVO.class);
}

需要注意的是，@CacheRemove注解的commandKey属性（注意不是cacheKey）时必须指定的，它用来指明需要使用请求缓存的请求命令，因为只有通过该属性的配置，Hystrix才能找到正确的请求命令缓存位置。

请求合并

通常微服务架构中的依赖通过远程调用实现，而远程调用中最常见的问题就是通信消耗与连接数占用。在高并发的情况之下，因通信次数的增加，总的通信时间消耗将会变的不那么理想。同时，因为对依赖服务的线程池资源有限，将出现排队等待与响应延迟的情况。为了优化这两个问题，Hystrix提供了HystrixCollapser来实现请求的合并，以减少通信消耗和线程数的占用。

HystrixCollapser实现了在HystrixCommand之前放置一个合并处理器，它将处于一个很短时间窗（默认10毫秒）内对同一依赖服务的多个请求进行整合并以批量方式发起请求的功能（服务提供方也需要提供相应的批量实现接口）。通过HystrixCollapser的封装，开发者不需要去关注线程合并的细节过程，只需要关注批量化服务和处理。下面我们从HystrixCollapser的使用实例，对其合并请求的过程一探究竟。

1.Hystrix的请求合并示例

public abstract class HystrixCollapser<BatchReturnType, ResponseType, RequestArgumentType> implements
        HystrixExecutable<ResponseType>, HystrixObservable<ResponseType> {
    ...
    public abstract RequestArgumentType getRequestArgument();

    protected abstract HystrixCommand<BatchReturnType> createCommand(Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests);

    protected abstract void mapResponseToRequests(BatchReturnType batchResponse, Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests);
    ...
}

从HystrixCollapser抽象类的定义中可以看到，它指定了三个不同的类型：

BatchReturnType：合并后批量请求的返回类型
ResponseType：单个请求返回的类型
RequestArgumentType：请求参数类型

而对于这三个类型的使用可以在它的三个抽象方法中看到：

RequestArgumentType getRequestArgument()：该函数用来定义获取请求参数的方法。
HystrixCommand createCommand(Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests)：合并请求产生批量命令的具体实现方法。
mapResponseToRequests(BatchReturnType batchResponse, Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests)：批量命令结果返回后的处理，这里需要实现将批量结果拆分并传递给合并前的各个原子请求命令的逻辑。

接下来，我们通过一个简单的示例来直观的理解实现请求合并的过程。

假设，当前微服务USER-SERVICE提供了两个获取User的接口：

/users/{id}：根据id返回User对象的GET请求接口。
/users?ids={ids}：根据ids参数返回User对象列表的GET请求接口，其中ids为以逗号分割的id集合。

而在服务消费端，为这两个远程接口已经通过RestTemplate实现了简单的调用，具体如下：

@Service
public class UserServiceImpl implements UserService {

    @Autowired
    private RestTemplate restTemplate;

    @Override
    public User find(Long id) {
        return restTemplate.getForObject("http://USER-SERVICE/users/{1}", User.class, id);
    }

    @Override
    public List<User> findAll(List<Long> ids) {
        return restTemplate.getForObject("http://USER-SERVICE/users?ids={1}", List.class, StringUtils.join(ids, ","));
    }

}

接着，我们来实现将短时间内多个获取单一User对象的请求命令进行合并的实现：

第一步：为请求合并的实现准备一个批量请求命令的实现，具体如下：

public class UserBatchCommand extends HystrixCommand<List<User>> {

    UserService userService;
    List<Long> userIds;

    public UserBatchCommand(UserService userService, List<Long> userIds) {
        super(Setter.withGroupKey(asKey("userServiceCommand")));
        this.userIds = userIds;
        this.userService = userService;
    }

    @Override
    protected List<User> run() throws Exception {
        return userService.findAll(userIds);
    }

}

批量请求命令实际上就是一个简单的HystrixCommand实现，从上面的实现中可以看到它通过调用userService.findAll方法来访问/users?ids={ids}接口以返回User的列表结果。

第二步，通过继承HystrixCollapser实现请求合并器：

public class UserCollapseCommand extends HystrixCollapser<List<User>, User, Long> {

    private UserService userService;
    private Long userId;

    public UserCollapseCommand(UserService userService, Long userId) {
        super(Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey("userCollapseCommand")).andCollapserPropertiesDefaults(
                HystrixCollapserProperties.Setter().withTimerDelayInMilliseconds(100)));
        this.userService = userService;
        this.userId = userId;
    }

    @Override
    public Long getRequestArgument() {
        return userId;
    }

    @Override
    protected HystrixCommand<List<User>> createCommand(Collection<CollapsedRequest<User, Long>> collapsedRequests) {
        List<Long> userIds = new ArrayList<>(collapsedRequests.size());
        userIds.addAll(collapsedRequests.stream().map(CollapsedRequest::getArgument).collect(Collectors.toList()));
        return new UserBatchCommand(userService, userIds);
    }

    @Override
    protected void mapResponseToRequests(List<User> batchResponse, Collection<CollapsedRequest<User, Long>> collapsedRequests) {
        int count = 0;
        for (CollapsedRequest<User, Long> collapsedRequest : collapsedRequests) {
            User user = batchResponse.get(count++);
            collapsedRequest.setResponse(user);
        }
    }

}

在上面的构造函数中，我们为请求合并器设置了时间延迟属性，合并器会在该时间窗内收集获取单个User的请求并在时间窗结束时进行合并组装成单个批量请求。下面getRequestArgument方法返回给定的单个请求参数userId，而createCommand和mapResponseToRequests是请求合并器的两个核心：

createCommand：该方法的collapsedRequests参数中保存了延迟时间窗中收集到的所有获取单个User的请求。通过获取这些请求的参数来组织上面我们准备的批量请求命令UserBatchCommand实例。
mapResponseToRequests：在批量命令UserBatchCommand实例被触发执行完成之后，该方法开始执行，其中batchResponse参数保存了createCommand中组织的批量请求命令的返回结果，而collapsedRequests参数则代表了每个被合并的请求。在这里我们通过遍历批量结果batchResponse对象，为collapsedRequests中每个合并前的单个请求设置返回结果，以此完成批量结果到单个请求结果的转换。

2.使用注解实现请求合并器
在快速入门的例子中，我们使用@HystrixCommand注解优雅地实现了HystrixCommand的定义，那么对于请求合并器是否也可以通过注解来定义呢？答案是肯定！

以上面实现的请求合并器为例，也可以通过如下方式实现：

@Service
public class UserService {

    @Autowired
    private RestTemplate restTemplate;

    @HystrixCollapser(batchMethod = "findAll", collapserProperties = {
            @HystrixProperty(name="timerDelayInMilliseconds", value = "100")
    })
    public User find(Long id) {
        return null;
    }

    @HystrixCommand
    public List<User> findAll(List<Long> ids) {
        return restTemplate.getForObject("http://USER-SERVICE/users?ids={1}", List.class, StringUtils.join(ids, ","));
    }
}

@HystrixCommand我们之前已经介绍过了，可以看到这里通过它定义了两个Hystrix命令，一个用于请求/users/{id}接口，一个用于请求/users?ids={ids}接口。而在请求/users/{id}接口的方法上通过@HystrixCollapser注解为其创建了合并请求器，通过batchMethod属性指定了批量请求的实现方法为findAll方法（即：请求/users?ids={ids}接口的命令），同时通过collapserProperties属性为合并请求器设置相关属性，这里使用@HystrixProperty(name=“timerDelayInMilliseconds”, value = “100”)将合并时间窗设置为100毫秒。这样通过@HystrixCollapser注解简单而又优雅地实现了在/users/{id}依赖服务之前设置了一个批量请求合并器。

3.请求合并的额外开销

虽然通过请求合并可以减少请求的数量以缓解依赖服务线程池的资源，但是在使用的时候也需要注意它所带来的额外开销：用于请求合并的延迟时间窗会使得依赖服务的请求延迟增高。比如：某个请求在不通过请求合并器访问的平均耗时为5ms，请求合并的延迟时间窗为10ms（默认值），那么当该请求的设置了请求合并器之后，最坏情况下（在延迟时间窗结束时才发起请求）该请求需要15ms才能完成。

由于请求合并器的延迟时间窗会带来额外开销，所以我们是否使用请求合并器需要根据依赖服务调用的实际情况来选择，主要考虑下面两个方面：

请求命令本身的延迟。如果依赖服务的请求命令本身是一个高延迟的命令，那么可以使用请求合并器，因为延迟时间窗的时间消耗就显得莫不足道了。
延迟时间窗内的并发量。如果一个时间窗内只有1-2个请求，那么这样的依赖服务不适合使用请求合并器，这种情况下不但不能提升系统性能，反而会成为系统瓶颈，因为每个请求都需要多消耗一个时间窗才响应。相反，如果一个时间窗内具有很高的并发量，并且服务提供方也实现了批量处理接口，那么使用请求合并器可以有效的减少网络连接数量并极大地提升系统吞吐量，此时延迟时间窗所增加的消耗就可以忽略不计了。

属性详解

在之前介绍Hystrix的使用方法时，已经涉及过一下Hystrix属性的配置，我们可以根据实现HystrixCommand的不同方式将配制方法分为如下两类。

①通过继承的方式实现，可用Setter对象来对请求命令的属性进行设置，比如下面的例子：

public UserCommand() {
	super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("GroupName"))
            .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
            .withExecutionTimeoutInMilliseconds(5000))
	    .andCommandKey(HystrixCommandKey.Factory.asKey("CommandName"))
	    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("ThreadPoolKey")));
	}

②通过注解的方法实现，只需要使用@HystrixCommand中的commandProperties属性来设置，比如：

@HystrixCommand(commandKey="hello",commandProperties={
	@HystrixProperty(name="execution.isolation.thread.timeoutInMilliseconds",value="5000")
})
public User findById(String name,int age){
	return restTemplate.getForObject("http://HELLO-SERVICE/hystrix/getUser?name={1}&age={2}", User.class, name, age);
}

Hystrix为我们提供的属性都存在四种优先级别的配置（优先级由低到高）。

全局默认值：如果没有设置下面三个优先级的属性，那么这个属性就是默认值。由于该属性通过代码定义，所以对这个级别，我们需要关注它在代码中定义的默认值即可。
全局配置属性：通过在配置文件中定义全局属性值，在应用启动时或在与Spring Cloud Config和Spring Cloud Bus实现的动态刷新配置功能的配合下，可以实现对“全局默认值”的覆盖已经在运行期对“全局默认值”的动态调整。
实例默认值：通过代码为实例定义的默认值。通过代码的方式为实例设置属性值来覆盖默认的全局配置
实例配置属性：通过配置文件来为指定的实例进行属性配置，以覆盖前面的三个默认值。它也可用Spring Cloud Config 和Spring Cloud Bus实现的动态刷新配置功能实现对具体实例配置的动态调整。

关于配置的写法，这里作简要说明：

属性配置可以是在properties文件中，也可以是在方法注解的属性里配置，两处配置的属性名称有区别，在properties里配置的属性是以 hystrix.command.default. 、 hystrix.threadpool.default 、 hystrix.collapser.default 开头，其中default表示默认值，如需要配置指定commandKey的值，将default换成commandKey即可。如果是在方法注解的属性里配置，则不需要这个前缀。下面我们来具体看看Hystrix有哪些具体的属性配置，且详细说明了属性配置在properties里配置和在方法注解里配置的写法。

1.Command属性

Command属性主要用来控制HystrixCommand命令的行为。它主要有下面5种不同类型的属性配置。

A. execution配置

execution配置控制的是HystrixCommand.run()的执行。

execution.isolation.strategy：该属性用来设置HystrixCommand.run()执行的隔离策略，它有如下两个选项。
- THREAD：通过线程池隔离的策略。它在独立的线程上执行，并且它的并发限制受线程池中线程数量的限制。
- SEMAPHORE：通过信号量隔离的策略。它在调用线程上执行，并且它的并发限制受信号量计数的限制。
execution.isolation.thread.timeoutMilliseconds: 该属性用来配置HystrixCommand执行的超时时间，单位为毫秒。当HystrixCommand执行时间超过该配置值后，Hystrix会将该命令标记为TIMEOUT并进入服务降级处理逻辑
execution.timeout.enabled：该属性用来设置HystrixCommand.run()的执行是否启用超时时间。默认为true，如果设置为false，那么execution.isolation.thread.timeoutMilliseconds属性将不起作用
execution.isolation.thread.interruptOnTimeout：该属性用来配置当HystrixCommand.run()执行超时的时候是否要将它中断。

Spring Cloud Hystrix 断路器-LMLPHP

execution.isolation.semaphore.maxConcurrentRequests：当HystrixCommand的隔离策略使用信号量时，该属性用来配置信号量的大小（并发请求数）。当最大并发请求数达到该设置值时，后续的请求将会被拒绝。

Spring Cloud Hystrix 断路器-LMLPHP

B. fallback配置

下面这些属性用来控制HystrixCommand.getFallback()的执行。这些属性同时适用于线程池的信号量的隔离策略。

fallback.isolation.semaphore.maxConcurrentRequests：该属性用来设置从调用线程中允许HystrixCommand.fallback()方法执行的最大并发请求数。当达到最大并发请求数时，后续的请求将会被拒绝并抛出异常（因为它已经没有后续的fallback可以被调用了）
fallback.enabled：该属性用来设置服务降级策略是否启用，如果设置为false，那么当请求失败或拒绝发生时，将不会调用HystrixCommand.getFallback()来执行服务降级逻辑。

C. circuitBreaker配置

下面这些是断路器的属性位置，用来控制HystrixCircuitBreaker的行为。

circuitBreaker.enabled：该属性用来确定当服务请求命令失败时，是否使用断路器来跟踪其健康指标和熔断请求
circuitBreaker.requestVolumeThreshold：该属性用来设置在滚动时间窗中，断路器熔断的最小请求数。例如：默认值为20的时候，如果滚动时间窗（默认10s）内收到了19个请求，即使这19个请求都失败了，断路器也不会打开。
circuitBreaker.sleepWindowInMilliseconds：该属性用来设置当断路器打开之后的休眠时间窗。休眠时间窗结束后，会将断路器置为“半开”状态，尝试熔断的请求命令，如果依然失败就将断路器继续设置为“打开”状态，如果成功就设置为“关闭”状态。
circuitBreaker.errorThresholdPercentage：该属性用来设置断路器打开的错误百分比条件。比如，默认值为50的情况下，表示在滚动时间窗中，在请求数量超过circuitBreaker.requestVolumeThreshold阈值的前提下，如果错误请求数的百分比超过50，就把断路器设置为“打开”状态，否则就设置为“关闭”状态。
circuitBreaker.forceOpen：如果将该属性设置为true，断路器将强制进入“打开”状态，它会拒绝所有请求。该属性优先于circuitBreaker.forceClosed属性
circuitBreaker.forceClosed：如果将该属性设置为true，断路器将强制进入“关闭”状态，它会接收所有请求。如果circuitBreaker.forceOpen属性为true，该属性不会生效。

D. metrics配置

下面的属性均与HystrixCommand和HystrixObservableCommand执行中捕获的指标信息有关。

metrics.rollingStats.timeInMilliseconds：该属性用来设置滚动时间窗的长度，单位为毫秒。该时间同于断路器判断健康度时需要收集信息的持续时间。断路器在收集指标信息的时候会根据设置的时间窗长度拆分成多个“桶”来累计各度量值，每个“桶”记录了一段时间内的采集指标。例如，当采用默认值10000毫秒时，断路器默认将其拆分成10个桶（桶的数量也可通过metrics.rollingStats.numBuckets参数设置），每个桶记录1000毫秒内的指标信息。

Spring Cloud Hystrix 断路器-LMLPHP
注意：该属性从Hystrix1.4.12版本开始只有在应用初始化时生效，通过动态刷新不会产生效果，避免运行期监测数据丢失。

metrics.rollingStats.numBuckets：该属性用来设置滚动时间窗统计指标信息时划分“桶”的数量。

注意：metrics.rollingStats.timeInMilliseconds参数的设置必须能被metrics.rollingStats.numBuckets参数整除，否则抛出异常，且1.4.12版本开始只有在应用初始化时生效，通过动态刷新不会产生效果，避免运行期监测数据丢失。
metrics.rollingPercentile.enabled：该属性用来设置对命令执行的延迟是否使用百分位数来跟踪和计算。如果为false，那么所有的概要统计都将返回-1。
metrics.rollingPercentile.timeInMilliseconds：该属性用来设置百分位统计的滚动窗口的持续时间，单位毫秒

注意：该属性从1.4.12版本开始只有在应用初始化时生效，通过动态刷新不会产生效果，避免运行期监测数据丢失。
metrics.rollingPercentile.numBuckets：该属性用来设置百分位统计滚动窗口中使用“桶”的数量。

注意：metrics.rollingPercentile.timeInMilliseconds参数的设置必须能被metrics.rollingPercentile.numBuckets参数整除，否则抛出异常，且1.4.12版本开始只有在应用初始化时生效，通过动态刷新不会产生效果，避免运行期监测数据丢失。
metrics.rollingPercentile.bucketSize：该属性用来设置在执行过程中每个“桶”中保留的最大执行次数。如果在滚动时间窗内发生超过该设定值的执行次数，就从最初的位置开始重写。例如，将该值设为100，滚动窗口10秒，若在10秒内一个“桶”中发生500次执行，那么该“桶”中只保留最后的100次执行统计。另外，增加该值的大小将会增加内存量的消耗，并增加排序百分位数所需的计算时间。

注意：该属性从1.4.12版本开始只有在应用初始化时生效，通过动态刷新不会产生效果，避免运行期监测数据丢失。
metrics.healthSnapshot.intervalInMilliseconds：该属性用来设置采集影响断路器状态的健康快照（请求成功、错误百分比）的间隔等待时间。

E. requestContext配置

下面的属性涉及HystrixCommand使用的HystrixRequestContext的设置。

requestCache.enabled：此属性用来设置是否开启请求缓存。
requestLog.enabled：该属性用来设置HystrixCommand的执行和事件是否打印日志到HystrixRequestLog中。

2.collapser属性
该属性除了在代码中用Setter()和配置文件配置外，也可使用注解进行配置。可使用@HystrixCollapser中的collapserProperties属性来设置，比如：

@HystrixCollapser(batchMethod="findByBatch",collapserProperties={
			@HystrixProperty(name="timerDelayInMilliseconds",value="100")
	})

下面这些属性用来控制命令合并相关的行为。

maxRequestsInBatch：该参数用来设置一次请求合并批处理中允许的最大请求数。
timerDelayInMilliseconds：该属性用来设置批处理过程中每个命令延迟的时间，单位为毫秒。
requestCache.enabled：该属性用来设置批处理过程中是否开启缓存。
10-07 14:35

小蜜蜂

Spring Cloud Hystrix 断路器