发表更新SpringCloud12 分钟读完 (大约1829个字)0次访问

SpringCloud-熔断器 Hystrix

Hystrix 是 Netflix 的一个开源项目，它能够在服务失效的情况下，通过隔离系统依赖服务的方式，防止服务级联失败，造成服务雪崩。同时Hystrix 还提供了失败回滚机制，使得系统能够更快的从异常中恢复。Hystrix 为服务间调用提供了保护和控制。

Hystrix 具有的功能如下：

  • 当通过客户端调用服务出现高延迟或者调用失败时，能够为系统提供保护机制
  • 在复杂的分布式场景下，可以防止服务雪崩效应
  • 提供快速失败（Fail Fast） 同时能够快速恢复
  • 提供失败回滚和优雅的服务降级机制
  • 提供近实时的监控、报警和运维控制手段

Hystrix 在实际应用过程中的使用方式很丰富，可以通过注解，也可以通过集成 HystrixCommand 和HystrixObservableCommand 。本篇将通过案例简单说明下说用方式。

环境准备

类别 值
JDK 1.8.0_162
SOFABoot/SpringBoot 3.0.0/2.0.x.RELEASE
SpringCloud Finchley.RC1
IDE IDEA

工程背景

本节将会创建一个 sofa-hystrix-client 工程，通过 Spring Cloud 提供的负载均衡器 hystrix 实现服务的熔断降级。

新建 sofa-hystrix-client

本工程继续使用《SpringCloud-Eureka 服务注册》中的父工程来构建。

右击 sofa-eureka-parent 父工程 -> New -> Module，这里选择 Maven 工程；

  • artifactId：sofa-hystrix-client

修改pom文件

pom文件中加入 hysterix 的依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
<dependencies>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>

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

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

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

配置文件

1
2
3
4
5
6
7
8
9
eureka:
  client:
    service-url:
      defaultZone: http://localhost:8761/eureka/
spring:
  application:
    name: hystrix-client
server:
  port: 8787

没有什么特殊的配置，还是作为一个 eureka-client 存在。

启动类

启动类上增加开启断路器的注解@EnableCircuitBreaker

1
2
3
4
5
6
7
8
9
10
11
12
13
@SpringBootApplication
@EnableCircuitBreaker
public class SofaHystrixApplication {

    @Bean
    @LoadBalanced
    public RestTemplate restTemplate(){
        return new RestTemplate();
    }
    public static void main(String[] args) {
        SpringApplication.run(SofaHystrixApplication.class, args);
    }
}

资源类

  • NormalService

中通过@HystrixCommand标准一个受保护的资源方法 getByServiceId()。getByServiceId 中通过restTemplate 来调用远程服务。@HystrixCommand注解的 fallbackMethod 属性指定当服务不可用时需要执行的 fallback 方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Service
public class NormalService {
    @Autowired
    private RestTemplate restTemplate;

    @HystrixCommand(fallbackMethod = "fallBack")
    public String getByServiceId(){
        return restTemplate.getForObject("http://HELLOSOFABOOTSERVICE/hello",String.class);
    }

    private String fallBack(){
        return "Filed to get data";
    }
}
  • HystrixRibbonController：通过instanceService调用上面的NormalService资源类
1
2
3
4
5
6
7
8
9
@RestController
public class HystrixRibbonController {
    @Autowired
    public NormalService instanceService;
    @RequestMapping("/hystrix")
    public String test(){
        return instanceService.getByServiceId();
    }
}

启动&验证

先后启动sofa-eureka-server-center 、sofa-eureka-provider、sofa-hystrix-client 三个工程。浏览器中输入：

http://localhost:8787/hystrix ，结果如下：

1
Hello SOFA! Now Port is 8086 And hostname is HelloSOFABootService

关闭 sofa-eureka-provider ，刷新浏览器：

1
Filed to get data

执行了 NormalService 中的 fallback 方法了。

资源隔离

hystrix 中提供了两中隔离策略，一种是基于线程池的隔离、另外一种是基于信号量的隔离。本篇只演示案例，具体原理请参看 hystrix 原理分析 相关文章。

基于线程池的隔离实现

新建一个 SofaThreadPoolHystrixCommand 类，继承 HystrixCommand。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
public class SofaThreadPoolHystrixCommand extends HystrixCommand {

    private RestTemplate restTemplate;

    public SofaThreadPoolHystrixCommand(RestTemplate restTemplate) {
        super(initailize());
        this.restTemplate = restTemplate;
    }

    public static HystrixCommand.Setter initailize(){
        // 线程池配置
        HystrixThreadPoolProperties.Setter hystrixThreadPoolProperties = 
            HystrixThreadPoolProperties.Setter()
                .withCoreSize(5)
                .withKeepAliveTimeMinutes(5)
                // 线程等待队列最大长度,默认值:-1 表示不等待直接拒绝,测试表明线程池使用直接决绝策略+ 合适大小的非回缩线程池效率最高.所以不建议修改此值。
                .withMaxQueueSize(10)
                .withQueueSizeRejectionThreshold(100);

        // 命令属性配置,这里指定隔离策略是 THREAD
        HystrixCommandProperties.Setter hystrixCommand = 
            HystrixCommandProperties.Setter() .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.THREAD)
                //意味着线程最多允许执行fallback的并发数为10,超过10 报fallback execution rejected
                .withFallbackIsolationSemaphoreMaxConcurrentRequests(10);

        HystrixCommand.Setter setter = HystrixCommand.Setter
            .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SofaThreadPoolHystrixCommand"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("sofaBootService"))
                .andCommandPropertiesDefaults(hystrixCommand)
                .andThreadPoolPropertiesDefaults(hystrixThreadPoolProperties)
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("sofa-hystrix-thread"));

        return setter;
    }

    /**
     * 受保护的资源
     * @return
     * @throws Exception
     */
    @Override
    protected Object run() throws Exception {
        return restTemplate.getForObject("http://HELLOSOFABOOTSERVICE/hello",String.class);
    }

    /**
     * 失败执行的保护方法
     * @return
     */
    @Override
    protected Object getFallback() {
        return "this is fail back policy";
    }
}

相关参数说明：

  • HystrixCommandGroupKey：配置全局唯一标识服务分组的名称，比如账户系统就是一个服务分组，监控时，相同分组的服务会聚合在一起，必填选项。
  • HystrixCommandKey：配置全局唯一标识服务的名称，比如账户系统有一个获取账号名的服务，那么就可以为这个服务起一个名字来唯一识别该服务，如果不配置，则默认是简单类名。
  • HystrixThreadPoolKey：配置全局唯一标识线程池的名称，相同线程池名称的线程池是同一个，如果不配置，则默认是分组名，此名字也是线程池中线程名字的前缀。
  • HystrixThreadPoolProperties：配置线程池参数
  • HystrixCommandProperties：配置该命令的一些参数，如 executionIsolationStrategy 配置执行隔离策略，默认是使用线程隔离。配置为 THREAD，线程池隔离；配置为 SEMAPHORE ，信号量隔离

这里为了模拟并发，使用 CountDownLatch 类来控制，在 HystrixRibbonController 中添加 testThread 资源方法：

1
2
3
4
5
6
7
8
9
@RequestMapping("/testThread")
public String testThread(){
    CountDownLatch countDownLatch = new CountDownLatch(1);
    for (int i = 0; i < THREAD_NUM; i ++) {
        new Thread(new ConsumerThread(countDownLatch)).start();
    }
    countDownLatch.countDown();
    return "data";
}

内部定义一个内部类，模拟调用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class ConsumerThread implements Runnable {
        private final CountDownLatch startLatch;
        public ConsumerThread(CountDownLatch startLatch) {
            this.startLatch = startLatch;
        }
        @Override
        public void run() {
            try {
                // 线程等待
                startLatch.await();
                // 执行操作
                SofaThreadPoolHystrixCommand sofaThreadPoolHystrixCommand = new SofaThreadPoolHystrixCommand(restTemplate);
                System.out.println(sofaThreadPoolHystrixCommand.execute().toString());

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

重启当前工程，浏览器执行 http://localhost:8787/testThread

1
2
3
4
5
6
7
8
this is fail back policy
this is fail back policy
this is fail back policy
this is fail back policy
this is fail back policy
// ... 省略
Hello SOFA! Now Port is 8086 And hostname is HelloSOFABootService
// ... 省略

基于信号量隔离

新建一个 SofaSemaphoreHystrixCommand 类，继承 HystrixCommand。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
public class SofaSemaphoreHystrixCommand extends HystrixCommand {
    private RestTemplate restTemplate;
    public SofaSemaphoreHystrixCommand(RestTemplate restTemplate) {
        super(initailize());
        this.restTemplate = restTemplate;
    }
    public static HystrixCommand.Setter initailize(){
        // 命令属性配置,这里指定隔离策略是 THREAD
        HystrixCommandProperties.Setter hystrixCommand = HystrixCommandProperties.Setter()
                .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)
                 //至少有10个请求，熔断器才进行错误率的计算
                .withCircuitBreakerRequestVolumeThreshold(0)
                //熔断器中断请求5秒后会进入半打开状态,放部分流量过去重试
                .withCircuitBreakerSleepWindowInMilliseconds(5000)
                //错误率达到50开启熔断保护
                .withCircuitBreakerErrorThresholdPercentage(50)
                //最大并发请求量
                .withExecutionIsolationSemaphoreMaxConcurrentRequests(10)
                //意味着信号量最多允许执行fallback的并发数为10,超过10 报fallback execution rejected
                .withFallbackIsolationSemaphoreMaxConcurrentRequests(10);

        HystrixCommand.Setter setter = HystrixCommand.Setter.
                withGroupKey(HystrixCommandGroupKey.Factory.asKey("SofaSemaphoreHystrixCommand"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("sofaBootService"))
                .andCommandPropertiesDefaults(hystrixCommand)
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("sofa-hystrix-thread"));

        return setter;
    }

    /**
     * 受保护的资源
     * @return
     * @throws Exception
     */
    @Override
    protected Object run() throws Exception {
        return restTemplate.getForObject("http://HELLOSOFABOOTSERVICE/hello",String.class);
    }

    /**
     * 失败执行的保护方法
     * @return
     */
    @Override
    protected Object getFallback() {
        return "this is fail back policy";
    }
}

同样使用 CountDownLatch 来模拟并发。在 HystrixRibbonController 中添加 testSemaphore 资源方法：

1
2
3
4
5
6
7
8
9
@RequestMapping("/testSemaphore")
    public String testSemaphore(){
        CountDownLatch countDownLatch = new CountDownLatch(1);
        for (int i = 0; i < THREAD_NUM; i ++) {
            new Thread(new ConsumerSemaphore(countDownLatch)).start();
        }
        countDownLatch.countDown();
        return "data";
    }

内部定义一个内部类 ConsumerSemaphore ，模拟调用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class ConsumerSemaphore implements Runnable {
        private final CountDownLatch startLatch;
        public ConsumerSemaphore(CountDownLatch startLatch) {
            this.startLatch = startLatch;
        }
        @Override
        public void run() {
            try {
                // 线程等待
                startLatch.await();
                // 执行操作
                SofaSemaphoreHystrixCommand sofaThreadPoolHystrixCommand = new SofaSemaphoreHystrixCommand(restTemplate);
                System.out.println(sofaThreadPoolHystrixCommand.execute().toString());

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

结果和线程隔离的差不多。不贴结果了。

SpringCloud-熔断器 Hystrix

http://www.glmapper.com/2018/12/31/springcloud/spring-cloud-hysitrx-project/

作者

卫恒

发布于

2018-12-31

更新于

2022-04-23

许可协议

#

喜欢这篇文章？打赏一下作者吧

评论

关注我

链接

标签

BeanDefinition1
BeanPostProcessor1
BeanWrapper1
CAS1
ClassLoader4
CopyOnWriteArraySet1
Disruptor1
Embedded Tomcat1
Exception1
Executor1
FX Application1
FatJar1
Ioc3
JAR1
JDK1
JPA1
JUC3
Kafka1
MANIFEST.MF1
Mockito1
Mysql1
OOM2
OpenTracing1
PropertySource1
Readiness Check1
ResourceLoader1
RestTemplate1
RocketMQ3
SOFA2
SOFAArk4
SOFAStack1
SpringBoot17
TestDouble1
ThreadPoolExecutor1
Tracer4
Zipkin1
ab test1
actuator1
aop3
api1
apollo1
arping1
bean 扩展机制1
bean 生命周期2
bootstrap1
cache2
classloader2
config6
cookie1
curator1
docker2
eureka5
event1
feign2
gateway2
gc2
git3
github1
guava1
http3
hystrix1
java5
junit2
jvm5
kafka1
kubernetes1
linux9
lock1
log5
logback1
maven4
maven debug1
maven plugin3
merge1
microservices1
mock1
mvcc1
mybatis1
mysql4
nacos2
namespace1
netflix1
nginx2
ngrok1
nohup1
oop-klass1
openjdk1
reactor1
rebase1
redis5
rpc1
samba1
session6
set1
shell1
slf4j1
spring11
spring cloud1
spring mvc1
spring 扩展机制2
sql1
ssh1
starter 机制1
test5
testing1
thread2
tomcat1
tool2
top1
typecheck1
yapi1
yum1
zookeeper3
中台1
事件机制1
依赖注入3
分布式6
分布式链路跟踪4
反向代理2
反射1
合肥1
并发编程5
微服务2
杭州1
架构4
框架1
泛型2
注册中心1
消息2
类加载2
线程池1
自动配置2
负载均衡3
运维2
迭代器1
配置2
集群选主1
×