• 周三. 6月 29th, 2022

5G编程聚合网

5G时代下一个聚合的编程学习网

热门标签

Hystrix使用分析

admin

11月 28, 2021

Hystrix使用

使用Hystrix实现熔断

要实现熔断,首先需要在请求调用方pom文件中加入

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

在启动类上加入@EnableCircuitBreaker注解,并在调用到另一个微服务的方法上加入一些配置

 @GetMapping("/checkHystrix/{userId}")
    @HystrixCommand(threadPoolKey = "checkHystrix1",
    threadPoolProperties = {
            @HystrixProperty(name = "coreSize",value = "1"),
            @HystrixProperty(name = "maxQueueSize",value = "20")
    },
    commandProperties = {
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds",value="2000")
    })
    public Integer checkHystrix(@PathVariable Long userId) {
        String url = "http://zhao-service-resume/resume/openstate/"+userId;
        Integer forObject =
                restTemplate.getForObject(url, Integer.class);
        return forObject;
    }

将调用到的服务中加入线程休眠十秒。访问上述服务,在界面上即可发现Hystrix的超时错误
file

服务降级

在配置中再增加一个 fallbackMethod = “customeFallback”,
配置降级兜底方法的具体形式是

    public  Integer customeFallback(Long userId){
        return -1;
    }

当某个服务熔断之后,服务器将不再被调⽤,此刻客户端可以⾃⼰准备⼀个本地的fallback回调,返回⼀个缺省值,这样做,虽然服务⽔平下降,但好⽍可⽤,⽐直接挂掉要强。但是在配置服务降级策略时,降级(兜底)⽅法必须和被降级⽅法相同的⽅法签名(相同参数列表、相同返回值)
如果参数不同会出现com.netflix.hystrix.contrib.javanica.exception.FallbackDefinitionException: fallback method wasn’t found: customeFallback([class java.lang.Long])
如果返回值不同会出现。且包装类和基本类型不能共用
Hint: Fallback method ‘public int com.zp.controller.AutoDeliverController.customeFallback(java.lang.Long)’ must return: class java.lang.Integer or its subclass

仓壁模式(线程池隔离模式)

如果不进⾏任何线程池设置,所有熔断⽅法使⽤⼀个Hystrix线程池(默认为10个线程),那么这样的话会导致问题,这个问题并不是扇出链路微服务不可⽤导致的,⽽是我们的线程机制导致的,如果⽅法A的请求把10个线程都⽤了,⽅法2请求处理的时候压根都、没法去访问B,因为没有线程可⽤,并不是B服务不可⽤。因此在配置Hystrix线程时,多个方法应该写多个线程池。这样能够让线程之间互不影响

  @GetMapping("/checkHystrix/{userId}")
    @HystrixCommand(threadPoolKey = "checkHystrix1",
    threadPoolProperties = {
            @HystrixProperty(name = "coreSize",value = "1"),
            @HystrixProperty(name = "maxQueueSize",value = "20")
    },
    commandProperties = {
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds",value="2000")
    })
    public Integer checkHystrix(@PathVariable Long userId) {
        String url = "http://zhao-service-resume/resume/openstate/"+userId;
        Integer forObject =
                restTemplate.getForObject(url, Integer.class);
        return forObject;
    }

    @GetMapping("/checkHystrixFallback/{userId}")
    @HystrixCommand(threadPoolKey = "checkHystrix2",
            threadPoolProperties = {
                    @HystrixProperty(name = "coreSize",value = "2"),
                    @HystrixProperty(name = "maxQueueSize",value = "20")
            },
            commandProperties = {
                    @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds",value="2000")
            },
            fallbackMethod = "customeFallback"


    )

通过postman发送批量请求,并通过jstack命令可以看到两个方法的线程池进行了隔离
file

其他属性的含义

            commandProperties = {
                    @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds",value="2000"),
                    @HystrixProperty(name = "metrics.rollingStats.timeInMilliseconds",value = "8000"),
                    @HystrixProperty(name = "circuitBreaker.requestVolumeThreshold",value = "2"),
                    @HystrixProperty(name = "circuitBreaker.errorThresholdPercentage",value="50"),
                    @HystrixProperty(name = "circuitBreaker.sleepWindowInMilliseconds",value="3000")
            }

metrics.rollingStats.timeInMilliseconds该属性是下面的熔断的统计时间窗口定义
circuitBreaker.requestVolumeThreshold该属性统计时间窗⼝内的失败的次数,达到此次数之后开启熔断操作
circuitBreaker.errorThresholdPercentage 窗口内失败的次数的百分比,达到这个百分比之后开启熔断操作
circuitBreaker.sleepWindowInMilliseconds 熔断多久间隔多久以后开始尝试是否恢复

源码简要分析

首先我们根据注解类@EnableCircuitBreaker可以找到SpringFactoryImportSelector类,该类通过泛型在spring.factories文件中找到注解了该泛型的配置类

@Override
	public String[] selectImports(AnnotationMetadata metadata) {
		if (!isEnabled()) {
			return new String[0];
		}
		AnnotationAttributes attributes = AnnotationAttributes.fromMap(
				metadata.getAnnotationAttributes(this.annotationClass.getName(), true));

		Assert.notNull(attributes, "No " + getSimpleName() + " attributes found. Is "
				+ metadata.getClassName() + " annotated with @" + getSimpleName() + "?");

		// Find all possible auto configuration classes, filtering duplicates
		List<String> factories = new ArrayList<>(new LinkedHashSet<>(SpringFactoriesLoader
				.loadFactoryNames(this.annotationClass, this.beanClassLoader)));

		if (factories.isEmpty() && !hasDefaultFactory()) {
			throw new IllegalStateException("Annotation @" + getSimpleName()
					+ " found, but there are no implementations. Did you forget to include a starter?");
		}

		if (factories.size() > 1) {
			// there should only ever be one DiscoveryClient, but there might be more than
			// one factory
			log.warn("More than one implementation " + "of @" + getSimpleName()
					+ " (now relying on @Conditionals to pick one): " + factories);
		}

		return factories.toArray(new String[factories.size()]);
	}

随后在spring.factories中找到了加载的配置类

org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker=
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerConfiguration

配置类中配置了一个切面HystrixCommandAspect,Hystrix实现的主要功能都在这个切面中进行了执行

    public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
        Method method = getMethodFromTarget(joinPoint);
        Validate.notNull(method, "failed to get method from joinPoint: %s", joinPoint);
        if (method.isAnnotationPresent(HystrixCommand.class) && method.isAnnotationPresent(HystrixCollapser.class)) {
            throw new IllegalStateException("method cannot be annotated with HystrixCommand and HystrixCollapser " +
                    "annotations at the same time");
        }
        MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
        MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
        HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
        ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
                metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();

        Object result;
        try {
            if (!metaHolder.isObservable()) {
                result = CommandExecutor.execute(invokable, executionType, metaHolder);
            } else {
                result = executeObservable(invokable, executionType, metaHolder);
            }
        } catch (HystrixBadRequestException e) {
            throw e.getCause();
        } catch (HystrixRuntimeException e) {
            throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
        }
        return result;
    }

使用同步调用会走executeObservable最终进入HystrixCommand进入最终的处理逻辑

    public R execute() {
        try {
            return queue().get();
        } catch (Exception e) {
            throw Exceptions.sneakyThrow(decomposeException(e));
        }
    }

进入执行队列并异步获取
最后执行 final Future delegate = toObservable().toBlocking().toFuture();具体Hystrix中的熔断降级操作即在toObservable()完成

欢迎搜索关注本人与朋友共同开发的微信面经小程序【大厂面试助手】和公众号【微瞰技术】,以及总结的分类面试题https://github.com/zhendiao/JavaInterview

file
file

发表评论

您的电子邮箱地址不会被公开。