Spring Cloud Netflix Hystrix介绍和使用

前面我们搭建了具有服务降级功能的Hystrix客户端，现在我们来详细了解下Hystrix的一些功能。



Hystrix的意思是豪猪，大家都知道，就是长满刺的猪。。。实际上，它表明了该框架的主要功能：自我保护功能。Hystrix具有服务降级，熔断，线程池隔离，信号量隔离，缓存等功能，基本上能覆盖到微服务中调用依赖服务会遇到的问题。下面我们介绍下，如何理解和使用这些功能。

1、最常用的的服务降级功能

　　当执行调用服务方法时，若调用方法出现问题，如：请求超时，抛出异常，线程池拒绝，熔断这些情况下，为该方法定义降级方法，以便在出现问题时执行，实现备用返回。之前我们已经实现了服务降级功能，主要就是通过@HystrixCommand(fallbackMethod = "defaultMethod")注释到需要在出现问题时降级的方法。fallbackMethod指定降级后执行的方法。方法定义在该类中，public，private，protected都可以。在注释的方法出问题后，如超时未返回(execution.isolation.thread.timeoutinMilliseconds来配置)，就会执行备用方法，返回备用方法的返回值。当然，降级的方法也可以定义再下一级的降级方法，实现和上面一样。

　　上面说到方法抛出异常也会触发服务降级，但是如果我们自定义了异常，并需要将异常抛出给上层做操作，不希望Hystrix捕捉到自定义异常执行服务降级时，可以使用@HystrixCommand(ignoreExceptions = {MyException.class})来定义忽略捕捉的异常。多个异常用逗号隔开。也可以将抛出的异常通过入参传到降级的方法，来实现不同类型异常的不同处理，需要将降级方法定义如下。

@HystrixCommand(fallbackMethod = "back")
public String getHello(String id)
{
return template.getForObject("http://helloclient/hello", String.class);
}

public String back(String id , Throwable e)
{
if (e instanceof NullPointerException)
{
return "client 2 has some error! NullPointerException";
}
else
{
return "client 2 has some error! Exception";
}
}

2、熔断器

　　熔断器，和物理概念中的断路器类似，断路器在高压高温的过载情况下，会自动断开，实现对电路的保护。熔断器也是一样，下面我们看下主要的接口类：HystrixCircuitBreaker.java，它定义了以下几个方法，并有两个内部实现类HystrixCircuitBreakerImpl，NoOpCircuitBreaker，断路器主要用到HystrixCircuitBreakerImpl。NoOpCircuitBreaker这个类表明不做任何操作，默认熔断器不打开，表明不起用熔断功能。以下的实现方法，都是指HystrixCircuitBreakerImpl的实现。熔断器有三个状态，OPEN，HALF_OPEN，CLOSED，如果要自定义参数配置，下面代码注释中可以找到。

/**
* Every {@link HystrixCommand} requests asks this if it is allowed to proceed or not.  It is idempotent and does
* not modify any internal state, and takes into account the half-open logic which allows some requests through
* after the circuit has been opened
*
* @return boolean whether a request should be permitted
*/
boolean allowRequest();

/**
* Whether the circuit is currently open (tripped).
*
* @return boolean state of circuit breaker
*/
boolean isOpen();

/**
* Invoked on successful executions from {@link HystrixCommand} as part of feedback mechanism when in a half-open state.
*/
void markSuccess();

/**
* Invoked on unsuccessful executions from {@link HystrixCommand} as part of feedback mechanism when in a half-open state.
*/
void markNonSuccess();

/**
* Invoked at start of command execution to attempt an execution.  This is non-idempotent - it may modify internal
* state.
*/
boolean attemptExecution();

(1) isOpen()方法用于判断熔断器是否打开。实现方法如下：

@Override
public boolean isOpen() {
//判断熔断器是否被强制打开，如果强制打开，返回true，表示熔断器已打开。circuitBreaker.forceOpen这个配置指定
if (properties.circuitBreakerForceOpen().get()) {
return true;
}
//判断熔断器是否被强制关闭。circuitBreaker.forceClosed
if (properties.circuitBreakerForceClosed().get()) {
return false;
}
//判断上一次断路器打开的时间是否大于零，访问成功，该值为-1，访问失败，该值为访问失败时的系统时间。根据是否大于零，判断熔断器是否打开。
return circuitOpened.get() >= 0;
}

(2) attemptExecution()，该方法会在熔断器开启的时候，有访问时，熔断器第一个执行的方法。如果返回false，则直接执行fallback降级方法。

@Override
public boolean attemptExecution() {
//判断熔断器是否被强制打开，如果强制打开，返回false后，直接执行fallback
if (properties.circuitBreakerForceOpen().get()) {
return false;
}
//判断熔断器是否被强制关闭
if (properties.circuitBreakerForceClosed().get()) {
return true;
}
//如果circuitOpened为-1，返回true，正常执行
if (circuitOpened.get() == -1) {
return true;
} else {
//如果circuitOpened不为-1，则表示断路器打开了，此时，服务会从circuitOpened起，休眠5秒（circuitBreaker.sleepWindowInMilliseconds配置，
//默认5000），直接返回false，执行fallback。若休眠时间超过5秒，并且当前熔断状态为打开状态，则会将熔断状态置为半开状态。如它的注释，只有第一个
//请求满足第一次为打开，之后的请求都为半开状态，返回false。
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;
}
}
}

(3)markSuccess()，在执行完attemptExecution()返回true正常执行成功后（未fallback），才会执行该方法，标注成功，若之前断路器为关闭状态，则不做处理，若为半开状态，则重置熔断器。

@Override
public void markSuccess() {
//如果当前状态为半开，则将state设置成closed，关闭熔断器。如果之前由于断路器打开时，之后的请求，Hystrix会放开一个请求去尝试是否服务正常，并将断路器置为半开，
//如果正常，则将断路器关闭，并重置断路器。
if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
//This thread wins the race to close the circuit - it resets the stream to start it over from 0
metrics.resetStream();
Subscription previousSubscription = activeSubscription.get();
if (previousSubscription != null) {
previousSubscription.unsubscribe();
}
Subscription newSubscription = subscribeToStream();
activeSubscription.set(newSubscription);
circuitOpened.set(-1L);
}
}

(4) markNonSuccess()，用来在正常请求下，请求失败后调用。

@Override
public void markNonSuccess() {
//如果当前为半开状态，且请求失败，则重新打开断路器，将最近一次访问失败的时间置为当前时间。
if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) {
//This thread wins the race to re-open the circuit - it resets the start time for the sleep window
circuitOpened.set(System.currentTimeMillis());
}
}

(5) 熔断器的打开。上面的方法都不会去打开熔断器，熔断器打开是由另一个方法去判断的。这个观察者的方法应该是周期执行的。

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 onCompleted() {

}

@Override
public void onError(Throwable e) {

}

@Override
public void onNext(HealthCounts hc) {
// check if we are past the statisticalWindowVolumeThreshold
//检查时间窗内的请求总数小于配置文件中的数量（采用的是buckets，感兴趣的自己研究下）。默认时间窗为10S（metrics.rollingStats.timeInMilliseconds,metrics.rollingStats.numBuckets），默认请求总数为20(circuitBreaker.requestVolumeThreshold)。
if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
// we are not past the minimum volume threshold for the stat window,
// so no change to circuit status.
// if it was CLOSED, it stays CLOSED
// if it was half-open, we need to wait for a successful command execution
// if it was open, we need to wait for sleep window to elapse
} else {
//时间窗内，统计的错误（失败）请求比例是否小于配置比例，默认配置是50%，通过circuitBreaker.errorThresholdPercentage=50指定。
if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
//we are not past the minimum error threshold for the stat window,
// so no change to circuit status.
// if it was CLOSED, it stays CLOSED
// if it was half-open, we need to wait for a successful command execution
// if it was open, we need to wait for sleep window to elapse
} else {
// our failure rate is too high, we need to set the state to OPEN
//如果时间窗内请求数大于定义数，且失败比例大于定义比例，并且当前熔断器关闭的情况下，将熔断器置为打开，并将circuitOpened置为当前时间。
if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
circuitOpened.set(System.currentTimeMillis());
}
}
}
}
});
}

(6) 过程：先文字敲吧，没画图工具。

　　正常情况：请求——>subscribeToStream未打开熔断器——>attemptExecution——>markSuccess

　　异常情况：请求——>subscribeToStream打开熔断器——>attemptExecution最后一个return返回false——>markNonSuccess，这个时候断路器打开状态，且在休眠时间窗内。

　　　　　　　请求——>subscribeToStream未处理——>attemptExecution在超过休眠时间窗后，放开一个请求，并把熔断器设置成半开——>请求成功，执行markSuccess，将熔断器从半开置为关闭，并重置熔断器；请求失败，则将半开状态置为打开状态，失败时间起点重置成当前时间，再次循环。

3、缓存

　　之前我以为是每次相同请求，会使用缓存直接返回。其实理解错了，Hystrix的缓存是在当次请求的缓存，当次请求中，多次使用同一方法时，会使用缓存。其他请求不能用到。而且还需初始化HystrixRequestContext，不然直接使用会报错，我们采用定义filter来初始化。不多说了，贴代码大家看下，代码中注释很清楚，启动注册中心和服务实例后（环境搭建见之前章节），就可以测试。

(1)pom.xml，application.yml配置，大家参见之前的章节。

(2)启动类，注意注解上@ServletComponentScan。

package com.example.demo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.servlet.ServletComponentScan;
import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker;
import org.springframework.cloud.client.loadbalancer.LoadBalanced;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
import org.springframework.context.annotation.Bean;
import org.springframework.web.client.RestTemplate;

@EnableCircuitBreaker
@SpringBootApplication
@EnableEurekaClient
@ServletComponentScan
public class ConsumerApplication {

@Bean
@LoadBalanced
RestTemplate template()
{
return new RestTemplate();
}

public static void main(String[] args) {
SpringApplication.run(ConsumerApplication.class, args);
}
}

(3)Filter类，用于初始化HystrixRequestContext。

package com.example.demo;

import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext;

import javax.servlet.*;
import javax.servlet.annotation.WebFilter;
import java.io.IOException;

@WebFilter(filterName = "HystrixRequestContextServletFilter",urlPatterns = "/*",asyncSupported = true)
public class HystrixRequestContextServletFilter implements Filter {

@Override
public void init(FilterConfig filterConfig) throws ServletException {

}

@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
HystrixRequestContext context = HystrixRequestContext.initializeContext();

try
{
chain.doFilter(request,response);
}
finally {
context.shutdown();
}
}

@Override
public void destroy() {

}
}

(4)controller类。

package com.example.demo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class ConsumerContorller {

@Autowired
HystrixServer server;

//注意，在这个controller中调用具有缓存功能的方法才会具备缓存效果。
@RequestMapping("/hello")
public String sayHello()
{
System.out.println("请求了一次hello2");
server.getHello2("1","ibethfy");
System.out.println("请求了二次hello2，不会打印hello2 initinized");
server.getHello2("1","ibethfy");
System.out.println("请求了三次hello2，清空缓存，会打印hello2 initinized");
server.updateHello2("1","ibethfy");
server.getHello2("1","ibethfy");
System.out.println("请求了四次hello2，入参不同，会打印hello2 initinized");
server.getHello2("1","ibethfy1");
return server.getHello2("1","ibethfy1");
}
}

(5)server类。

package com.example.demo;

import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand;
import com.netflix.hystrix.contrib.javanica.annotation.HystrixProperty;
import com.netflix.hystrix.contrib.javanica.cache.annotation.CacheKey;
import com.netflix.hystrix.contrib.javanica.cache.annotation.CacheRemove;
import com.netflix.hystrix.contrib.javanica.cache.annotation.CacheResult;
import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

@Service
public class HystrixServer {

@Autowired
RestTemplate template;

//通过指定生成缓存key的方法生成key，commandKey指定一个HystrixCommand的key，表示注解@HystrixCommand的方法的key。groupKey表示一个类型分组的key。threadPoolKey指定线程池的key。
//fallbackMethod指定降级方法，commandProperties指定该HystrixCommand方法的参数，是个数组类型，里面的值是@HystrixProperty，多个用逗号隔开。
@CacheResult(cacheKeyMethod = "generateCacheKey")
@HystrixCommand(commandKey = "getHello1",groupKey = "getHello",threadPoolKey = "getHelloThreadPool",fallbackMethod = "back",commandProperties = {
@HystrixProperty(name="execution.isolation.thread.timeoutinMilliseconds", value = "5000")
})
public String getHello1()
{
System.out.println("hello1 initinized");
return template.getForObject("http://helloclient/hello", String.class);
}

private String generateCacheKey()
{
return "myHelloKey";
}

//若不指定cache的key，默认使用方法的所有参数作为key
@CacheResult
@HystrixCommand(commandKey = "getHello2",groupKey = "getHello",threadPoolKey = "getHelloThreadPool")
public String getHello2(String id,String name)
{
System.out.println("hello2 initinized");
return template.getForObject("http://helloclient/hello", String.class);
}

//使用@CacheRemove在数据更新时，移除对应key的缓存，需要指定commandKey，@HystrixCommand里面的参数可以指定亦可以不用
@CacheRemove(commandKey = "getHello2")
@HystrixCommand(commandKey = "getHello2",groupKey = "getHello",threadPoolKey = "getHelloThreadPool")
public void updateHello2(String id,String name)
{
System.out.println("hello2 id = "+ id + ", name = "+ name + " removed");
}

//使用@CacheKey指定参数作为key
@CacheResult
@HystrixCommand(commandKey = "getHello3",groupKey = "getHello",threadPoolKey = "getHelloThreadPool")
public String getHello3(@CacheKey("id") String id, String name)
{
System.out.println("请求了一次hello3");
return "hello3 " + id + name;
}

public String back(Throwable e)
{
if (e instanceof NullPointerException)
{
return "client 2 has some error! NullPointerException";
}
else
{
return "client 2 has some error! Exception";
}
}

}

4、线程隔离和信号量隔离。

　　Hystrix为了避免多个不同服务间的调用影响，使用了线程隔离模式，它为每个依赖服务单独创建自己的线程池，就算某个服务延迟或问题阻塞，其余的服务也能正常执行。总之，使得我们的服务更加健壮。当然，创建这么多线程池，可能会对性能造成影响，但Hystrix测试后，独立线程池带来的好处，远大于性能损耗的坏处。所以，大家可以放心使用。

　　ExecutionIsolationStrategy枚举中定义了两个THREAD, SEMAPHORE，一个是线程池，一个是信号量，Hystix默认使用线程池。通过execution.isolation.strategy可以切换。

　　Hystrix默认情况下，会让配置了同组名的groupKey的command使用同一线程池，但也支持使用threadPoolKey配置线程池key。

　　对于那些本来延迟就比较小的请求（例如访问本地缓存成功率很高的请求）来说，线程池带来的开销是非常高的，这时，可以考虑采用非阻塞信号量（不支持超时），来实现依赖服务的隔离，使用信号量的开销很小。但绝大多数情况下，Netflix 更偏向于使用线程池来隔离依赖服务，因为其带来的额外开销可以接受，并且能支持包括超时在内的所有功能。

好了，Hystrix的主要功能基本介绍完了，码字不容易呀，，，，