Hystrix服务容错处理

Posted xc-xinxue

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Hystrix服务容错处理相关的知识,希望对你有一定的参考价值。

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

如何避免服务雪崩效应?通过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).

以上是关于Hystrix服务容错处理的主要内容,如果未能解决你的问题,请参考以下文章

使用Hystrix实现微服务的容错处理

SpringCloud之Hystrix容错处理

springCloud(12):使用Hystrix实现微服务的容错处理-Hystrix的监控

springcloud-hystrix断路器对微服务的容错处理

SpringCloud+Hystrix服务容错

将hystrix整合至spring cloud feign实现容错处理