Hystrix服务容错处理

Posted 2021-03-06 xc-xinxue

在微服务架构中存在多个可直接调用的服务，这些服务若在调用时出现故障会导致连锁效应，也就是可能会让整个系统变得不可用，这种情况我们称之为服务雪崩效应。

如何避免服务雪崩效应？通过Hystrix就能够解决。

1.Hystrix

Hystrix是Netflix针对微服务分布式系统采用的熔断保护中间件， 相当于电路中的保险 丝。在微服务架构下，很多服务都相互依赖，如果不能对依赖的服务进行隔离，那么服务 本身也有可能发生故障， Hystrix通过Hystrix Command对调用进行隔离， 这样可以阻止故 障的连锁反应，能够让接口调用快速失败并迅速恢复正常，或者回退并优雅降级。

1.1 Hystrix的简单实用

创建一个空的Maven项目，增加Hystrix的依赖：

<dependency>
    <groupId>com.netflix.hystrix</groupId>
    <artifactId>hystrix-core</artifactId>
    <version>1.5.18</version>
</dependency>

编写第一个HystrixCommand，如代码所示：

public class MyHystrixCommand extends HystrixCommand<String>{
    private final String name;
    public MyHystrixCommand(String name) {
        super(HystrixCommandGroupKey.Factory.asKey("MyGroup"));
        this.name=name;
    }
    @Override
    protected String run() throws Exception {
        // TODO Auto-generated method stub
        return this.name+":"+Thread.currentThread().getName();
    }

}

首先需要继承HystrixCommand，通过构造函数设置一个Groupkey。具体的逻辑在run方法中，我们返回了一个当前线程名称的值，写一个main方法来调用上面编写的MyHystrixCommand程序，如代码所示：

public static void main(String[] args) throws InterruptedException,ExecutionException{
        String result=new MyHystrixCommand("yinjihuan").execute();
        System.out.println(result);
    }

输出结果是：yinjihuan:hystrix-MyGroup-1.由此可以看出，构造函数中设置的组名变成了线程的名字。

上面是同步调用，如果需要异步调用可以使用代码如下：

public static void main(String[] args) throws InterruptedException,ExecutionException{

        Future<String> future=new MyHystrixCommand("yinjihuan").queue();
        System.out.println(future.get());
    }

1.2 回退支持

下面我们通过增加执行时间模拟调用失败的情况，首先改造MyHystrixCommand，增加getFallback方法返回回退内容，如代码所示：

public class MyHystrixCommand extends HystrixCommand<String>{
    private final String name;
    public MyHystrixCommand(String name) {
        super(HystrixCommandGroupKey.Factory.asKey("MyGroup"));
        this.name=name;
    }
    @Override
    protected String run() throws Exception {
        // TODO Auto-generated method stub
        try {
            Thread.sleep(1000*10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return this.name+":"+Thread.currentThread().getName();
    }
    protected String getFallback() {
        return "失败了";
    }
    
    
    public static void main(String[] args) throws InterruptedException,ExecutionException{
//        String result=new MyHystrixCommand("yinjihuan").execute();
//        System.out.println(result);
        Future<String> future=new MyHystrixCommand("yinjihuan").queue();
        System.out.println(future.get());
    }

}

执行代码，返回内容是”失败了”,证明触发了回退。

1.3 信号量策略配置

信号量策略配置方法如代码所示：

public MyHystrixCommand(String name) {
        super(HystrixCommand.Setter
                .withGroupKey(HystrixCommandGroupKey.Factory.asKey("MyGroup"))
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        .withExecutionIsolationStrategy(
                                HystrixCommandProperties
                                .ExecutionIsolationStrategy.SEMAPHORE)));
        this.name=name;
    }

之前在run方法中特意输出了线程名称，通过这个名称就可以确定当前是线程隔离还是信号量隔离。

1.4 线程隔离策略配置

系统默认采用线程隔离策略，我们可以通过andThreadPoolPropertiesDefaults配置线程池的一些参数，如代码所示：

public MyHystrixCommand(String name) {
        super(HystrixCommand.Setter
                .withGroupKey(HystrixCommandGroupKey.Factory.asKey("MyGroup"))
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        .withExecutionIsolationStrategy(
                                HystrixCommandProperties
                                .ExecutionIsolationStrategy.THREAD)
                        ).andThreadPoolPropertiesDefaults(
                                HystrixThreadPoolProperties.Setter()
                                .withCoreSize(10)
                                .withMaxQueueSize(100)
                                .withMaximumSize(100)
                                )
                );
        this.name=name;
    }

1.5 结果缓存

缓存在开发中经常用到，我们常用Redis这种第三方的缓存数据库来对数据进行缓存处理，Hystrix中也为我们提供了方法级别的缓存。通过重写getCacheKey来判断是否返回缓存的数据，getCacheKey可以根据参数来生成，这样同样的参数就可以都用到缓存了。

改写之前的MyHystrixCommand，在其中增加getCacheKey的重写实现，如代码所示：

//返回缓存key
protected String getCacheKey() {
        return String.valueOf(this.name);
    }

上面代码中我们创建对象传进来的name参数作为缓存的key。

为了证明能够用到缓存，在run方法中加一行输出，在调用多次的情况下，如果控制台只输出了一次，那么可以知道后面的都是走的缓存逻辑，如代码所示：

protected String run()  {
        // TODO Auto-generated method stub
        System.err.println("gat data");
        return this.name+":"+Thread.currentThread().getName();
    }

执行main方法，发现报错： Request caching is not available. Maybe you need to initialize the HystrixRequestContext?

根据错误提示可以知道，缓存的处理请求取决于请求的上下文，我们必须初始化Hystrix-RequestContext.

改造main方法中的调用代码，初始化HystrixRequestContext，如代码所示：

public static void main(String[] args) throws InterruptedException,ExecutionException{
        HystrixRequestContext context=HystrixRequestContext.initializeContext();
        String result=new MyHystrixCommand("yinjihuan").execute();
        System.out.println(result);
        Future<String> future=new MyHystrixCommand("yinjihuan").queue();
        System.out.println(future.get());
        context.shutdown();
    }

改造完之后重新执行main方法，可以正常运行了，输出结构如下：

get data
yinjihuan:hystrix-MyGroup-1
yinjihuan:hystrix-MyGroup-1

可以看到只输出一次get data，缓存生效。

1.6 缓存清除

在上节学习了如何使用hystrix来实现数据缓存功能。有缓存必然就有清除缓存的动作，当数据发生变动时，必须将缓存中的数据也更新掉，不然就会产生脏数据的问题。同样，Hystrix也有清除缓存的功能。

增加一个支持清除缓存的类，如代码所示：

public class ClearCacheHystrixCommand extends HystrixCommand<String>{
    private final String name;
    private static final HystrixCommandKey GETTER_KEY=HystrixCommandKey.Factory.asKey("MyKey");
    public ClearCacheHystrixCommand(String name) {
        super(HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.
                Factory.asKey("MyGroup")).andCommandKey(GETTER_KEY));
        this.name=name;
    }
    
    public static void flushCache(String name) {
        HystrixRequestCache.getInstance(
                GETTER_KEY, HystrixConcurrencyStrategyDefault.getInstance()).clear(name);
    }
    
    
    protected String getCacheKey() {
        return String.valueOf(this.name);
    }
    @Override
    protected String run() throws Exception {
        System.err.println("get data");
        return this.name+":"+Thread.currentThread().getName();
    }
    
    protected String getFallback() {
        return "失败了";
    }

}

flushCache方法就是清除缓存的方法，通过HystrixRequestCache来执行清除操作，根据getCacheKey返回的key来清除。

修改调用代码来验证清除是否有效果，如代码所示：

HystrixRequestContext context=HystrixRequestContext.initializeContext();
        String result=new ClearCacheHystrixCommand("yinjihuan").execute();
        System.out.println(result);
        ClearCacheHystrixCommand.flushCache("yinjihuan");
        Future<String> future=new ClearCacheHystrixCommand("yinjihuan").queue();
        System.out.println(future.get());

执行两次相同的key，在第二次执行之前调用缓存清除的方法，也就是说第二次用不到缓存，输出结果如下：

get data
yinjihuan:hystrix-MyGroup-1
get data
yinjihuan:hystrix-MyGroup-2

1.7 合并请求

Hystrix支持将多个请求自动合并为一个请求，利用这个功能可以节省网络开销，比如每个请求都有通过网络访问远程资源。如果把多个请求合并为一个一起执行，将多次网络交互变成一次，则会极大节省开销。如代码所示：

public class MyHystrixCollapser extends HystrixCollapser<List<String>,String,String>{

    private final String name;
    public MyHystrixCollapser(String name) {
        this.name=name;
    }
    @Override
    public String getRequestArgument() {
        // TODO Auto-generated method stub
        return name;
    }

    @Override
    protected HystrixCommand<List<String>> createCommand(final Collection<CollapsedRequest<String, String>> requests) {
        // TODO Auto-generated method stub
        return new BatchCommand(requests);
    }

    @Override
    protected void mapResponseToRequests(List<String> batchResponse,
            Collection<CollapsedRequest<String, String>> requests) {
        int count=0;
        for(CollapsedRequest<String, String>request:requests) {
            request.setResponse(batchResponse.get(count++));
        }
    }
    
    private static final class BatchCommand extends HystrixCommand<List<String>>{
        private final Collection<CollapsedRequest<String,String>> requests;
        private BatchCommand(Collection<CollapsedRequest<String, String>>requests) {
            super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
                    .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueForKey")));
                this.requests=requests;
        }
        @Override
        protected List<String> run() throws Exception {
            System.out.println("真正执行请求......");
            ArrayList<String> response=new ArrayList<String>();
            for(CollapsedRequest<String, String>request:requests) {
                response.add("返回结果："+request.getArgument());
            }
            return response;
        }
    }

}

接下来编写测试代码，如代码所示：

HystrixRequestContext context=HystrixRequestContext.initializeContext();
        Future<String> f1=new MyHystrixCollapser("yinjihuan").queue();
        Future<String> f2=new MyHystrixCollapser("yinjihuan333").queue();
        System.out.println(f1.get()+"="+f2.get());
        context.shutdown();

通过

MyHystrixCollapser创建两个执行任务，按照正常的逻辑肯定是分别执行这两个任务，通过HystrixCollapser可以将多个任务合并到一起执行。从输出结果可以看出，任务的执行是在run方法中去做的，输出结果如下：

真正执行请求......
返回结果：yinjihuan=返回结果：yinjihuan333

2.在Spring Cloud中使用Hystrix

2.1 简单使用

在fsh-substitution服务中增加Hystrix的依赖：

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

在启动类上添加@EnableHystrix或者@EnableCircuitBreaker。注意，@EnableHystrix中包含了@EnableCircuitBreaker。

然后改造之前的callHello方法，在上面增加一个@HystrixCommand注解，用于指定依赖服务调用延迟或失败时调用的方法，如代码所示：

@GetMapping("/callHello")
    @HystrixCommand(fallbackMethod="defaultCallHello")
    public String callHello() {

String result=restTemplate.getForObject("http://fsh-house/house/hello", String.class);

        return result;
    }

当

调用失败触发熔断时会用defaultCallHello方法来回退具体的内容，定义defaultCallHello方法的代码如下所示：

public String defaultCallHello() {
        return "fail";
    }

只启动fsh-substitution服务而不启动fsh-house服务， 调用/callHello接口， 可以看到返回的内容是“fail”。

将启动类上的@EnableHystrix去掉， 重启服务， 再次调用/callHello接口可以看到返 回的是500错误信息，这个时候就没有用到回退功能了。

2.2 配置详解

HystrixCommand中除了fallbackMethod还有很多的配置，下面来看看这些配置：

官方的配置信息文档请参考：https://github.com/Netflix/Hystrix/wiki/Configuration

2.3 Feign整合Hystrix服务容错

首先需要执行2.1节中的集成步骤，然后在属性文件中开启Feign对Hystrix的支持：

feign.hystrix.enabled=true

1.Fallback方式

在Feign的客户端类上的@FeignClient注解中指定fallback进行回退，改造fsh-housed客户端类HouseRemoteClient，为其配置fallback。代码如下：

@FeignClient(value = "fsh-house", path = "/house", configuration = FeignConfiguration.class
        , fallback = HouseRemoteClientHystrix.class)
public interface HouseRemoteClient {
    
    @GetMapping("/{houseId}")
    HouseInfoDto hosueInfo@PathVariable("houseId")Long houseId);
}

HouseRemoteClientHystrix类需要实现HouseRemoteClient类中所有的方法，返回回退时的内容，如代码所示：

@Component
public class HouseRemoteClientHystrix implements HouseRemoteClient{
    
    public HouseInfoDto houseInfo(Long houseId){
        return new HouseInfoDto();
}

}

启动fsh-substitution服务，停掉所有fsh-house服务，然后访问http://localhost:8082/substitution/1接口，这时候fsh-house服务是不可用的，必然触发回退，返回的内容是正常的格式，只是house对象是空的，这证明回退生效了。在这种情况下，如果你的接口调用了多个服务的接口，那么只有fsh-house服务会没数据，不会影响别的服务，如果不用Hystrix回退处理，整个请求都将失败。

2.FallbackFactory方式

通过fallback已经可以实现服务不可用时回退的功能，如果你想知道触发回退的原因，可以使用FallbackFactory来实现回退功能，如代码所示：

@Component
public class HouseRemoteClientFallbackFactory implements FallbackFactory<HouseRemoteClient> {
    @Override
    public HouseRemoteClient create(final Throwable cause) {
        return new HouseRemoteClient() {

            @Override
            public HouseInfoDto hosueInfo(Long houseId) {
                HouseInfoDto info = new HouseInfoDto();
                info.setData(new HouseInfo(1L, "", "", cause.getMessage()));
                return info;
            }
};
}
}

FallbackFactory的使用就是在@FeignClient中用fallbackFactory指定回退处理类，如代码所示：

@FeignClient(value = "fsh-house", path = "/house", configuration = FeignConfiguration.class
        , fallbackFactory = HouseRemoteClientFallbackFactory.class)

在这个回退处理的时候，将异常信息设置到HouseInfo中的name属性中了。我们重启fsh-substitution,调用接口，可以看到异常信息返回在结果里面了，FallbackFactory和Fallback唯一的区别就在这里。

2.4 Feign中禁用Hystrix

禁用Hystrix还是比较简单的，目前有两种方式可以禁用，其一是在属性文件中进行全部禁用：

feign.hystrix.enabled=false

另一种是通过代码的方式禁用某个客户端，在Feign的配置类中增加代码如下所示：

@Configuration
public class FeignConfiguration{

    @Bean
    @Scope("prototype")
    public Feign.Builder feignBuilder(){
        return Feign.builder();
    }
}

3.Hystrix监控

在微服务架构中，Hystrix除了实现容错外，还提供了实时监控功能。在服务调用时，Hystrix会实时累积关于HystrixCommand的执行信息，比如每秒的请求数、成功数等。更多的指标信息请查看官方文档地址：

https://github.com/Netflix/Hystrix/wiki/Metrics-and-Monitoring.

Hystrix监控需要两个必备条件：

(1)必须有Actuator的依赖，如代码所示：

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

(2)必须有Hystrix的依赖，Spring Cloud中必须在启动类中添加@EnableHystrix开启Hystrix，如代码所示：

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

我们访问fsh-substitution中的地址（http://localhost:8082/hystrix.sream）中可以看到一直在输出“ping:”,这种情况是因为还没有数据，等到HystrixCommand执行了之后就可以看到具体数据了。

4.整合Dashboard查看监控数据

上面讲了通过hystrix.stream端点来获取监控数据，但是这些数据是以字符串的形式展现的，实际使用中不方便查看，我们可以借助Hystrix-dashboard对监控进行图形化展示。

源码参考：https://github.com/yinjihuan/spring-cloud/tree/master/fangjia-hystrix-dashboard

创建一个maven项目，在pom.xml中添加dashboard的依赖，如代码：

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

创建启动类，在启动类上添加@EnableHystrixDashboard注解，如代码所示：

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

在属性配置文件中只需要配置服务名称和服务端口：

spring.application.name=fangjia-hystrix-dashboard
server.port=9011

5.Turbine聚合集群数据

5.1 Turbine使用

Turbine是聚合服务器发送事件流数据的一个工具。Hystrix只能监控单个节点， 然后通 过dashboard进行展示。实际生产中都为集群， 这个时候我们可以通过Turbine来监控集群 下Hystrix的metrics情况， 通过Eureka来发现Hystrix服务。

首先增加turbine的依赖：

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

在启动上增加@EnableTurbine和@EnableDiscoveryClient。在属性文件中配置如下：

eureka.client.serviceUrl.defaultZone=http://yinjihuan:123456@master:8761/eureka/

#配置需要聚合的服务名称
turbine.appConfig=fsh-substitution,fsh-house

#Turbine需要聚合的集群名称
turbine.aggregator.clusterConfig=default

#集群名表达式
turbine.clusterNameExpression=new String("default")

重启服务，就可以使用http://localhost:9011/turbine.stream来访问集群的监控数据了。

5.2 context-path导致监控失败

如果被监控的服务中设置了context-path，则会导致Turbine无法获取监控数据。这时需要在Turbine中指定turbine.instanceUrlSuffix来解决这个问题：turbine.instanceUrlSuffix=/sub/hystrix.stream

sub用于监控服务的context-path。这种方式是全局配置，如果每个服务的context-path都不一样，这个时候有一些就会出问题，那么就需要对每个服务做一个集群，然后配置集群对应的context-path:

turbine.instanceUrlSuffix.集群名称=/sub/hystrix.stream

更多内容可参见官方文档：https://github.com/Netflix/Turbine/wiki/Configuration-(1.x).