SpringCloud.Hoxton 版本, 全新熔断器CircuitBreaker

约定不等于承诺〃 2023-06-26 03:26 4阅读 0赞

### SpringCloud.Hoxton 版本, 全新熔断器CircuitBreaker ###

*  SpringCloud.Hoxton 版本, 全新熔断器CircuitBreaker
 *   *  前置说明
     *  如何使用
     *  自动装配
     *  总结

# SpringCloud.Hoxton 版本, 全新熔断器CircuitBreaker #

## 前置说明 ##

上文讲到了Honxton 版本的新的负载均衡器, 这次准备讲一下新的熔断器. 其实也不是新的熔断器, 是一个新的统一接口. 用过hystrix 的应该知道, 其实它整合到springcloud的时候, 其用法包括注解方式还是编程方式都是hystrix自带的. 这样的用法有个不好的地方就是, 如果我要切换新的熔断器(hystrix官方不在更新,而是推荐我们使用Resilience4J, 虽然本身hystrix已经比较成熟了,不替换也不会有太多问题). 这就有点蛋疼了, 虽然网关那里只要改下依赖和配置就行了, 但是许多微服务中无论使用了注解的方式还是编程的方式,我都要一一替换.  
所以在这个版本, springcloud提供了熔断的统一接口(暂时好像没有注解的方式, 但是我们可以自己用aop实现) `CircuitBreakerFactory` , `ReactiveCircuitBreakerFactory`. 这两个接口分别用于非响应式和响应式编程.  
主要的依赖, 也可以直接下载我之前的 sample-springcloud 项目, 里面有 h版本的分支demo, 链接放到评论区

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

## 如何使用 ##

首先介绍一下非响应式的统一熔断接口`CircuitBreakerFactory`

// 基本用法
    @Autowired
    private CircuitBreakerFactory cbFactory;
    @Autowired
    private RestTemplate rest;
    
    public String slow() { 
            return cbFactory.create("slow").run(() -> rest.getForObject("/slow", String.class), throwable -> "fallback");
        }

首先很明显最开始是一个工厂方法, 所以会调用create方法, 这个接口的参数是一个个`id`, 这个id是用来获取对应的配置的.具体看下面,分别是`hystrix`和`resilience4j`的`create`方法

// org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerFactory#create
    public HystrixCircuitBreaker create(String id) { 
    		Assert.hasText(id, "A CircuitBreaker must have an id.");
    		HystrixCommand.Setter setter = getConfigurations().computeIfAbsent(id,
    				defaultConfiguration);
    		return new HystrixCircuitBreaker(setter);
    	}
    
    // org.springframework.cloud.circuitbreaker.resilience4j.Resilience4JCircuitBreakerFactory#create
    	public Resilience4JCircuitBreaker create(String id) { 
    		Assert.hasText(id, "A CircuitBreaker must have an id.");
    		Resilience4JConfigBuilder.Resilience4JCircuitBreakerConfiguration config = getConfigurations()
    				.computeIfAbsent(id, defaultConfiguration);
    		return new Resilience4JCircuitBreaker(id, config.getCircuitBreakerConfig(),
    				config.getTimeLimiterConfig(), circuitBreakerRegistry, executorService,
    				Optional.ofNullable(circuitBreakerCustomizers.get(id)));
    	}

可以看到基本都是获取相关的配置, 没有就取默认的配置.

接下来run方法, 第一个参数就是原逻辑执行, 第二个参数是降级逻辑.

// org.springframework.cloud.netflix.hystrix.HystrixCircuitBreaker#run
    public <T> T run(Supplier<T> toRun, Function<Throwable, T> fallback) { 
    		// 这里的用法就是编程式的用法,没啥区别
    		HystrixCommand<T> command = new HystrixCommand<T>(setter) { 
    			@Override
    			protected T run() throws Exception { 
    				return toRun.get();
    			}
    
    			@Override
    			protected T getFallback() { 
    				return fallback.apply(getExecutionException());
    			}
    		};
    		return command.execute(); //这里就是hystrix的真正执行了, 底层是rxjava实现的
    	}
    
    // org.springframework.cloud.circuitbreaker.resilience4j.Resilience4JCircuitBreaker#run
    public <T> T run(Supplier<T> toRun, Function<Throwable, T> fallback) { 
    		TimeLimiter timeLimiter = TimeLimiter.of(timeLimiterConfig);
    		Supplier<Future<T>> futureSupplier = () -> executorService.submit(toRun::get); // 将原逻辑丢到线程池, 获得一个future
    		Callable restrictedCall = TimeLimiter.decorateFutureSupplier(timeLimiter,
    				futureSupplier); // 装饰上 超时逻辑, 其实很简单哈 ,就是通过Future的超时获取就可以了
    
    		io.github.resilience4j.circuitbreaker.CircuitBreaker defaultCircuitBreaker = registry
    				.circuitBreaker(id, circuitBreakerConfig);
    		circuitBreakerCustomizer
    				.ifPresent(customizer -> customizer.customize(defaultCircuitBreaker));
    		Callable<T> callable = io.github.resilience4j.circuitbreaker.CircuitBreaker
    				.decorateCallable(defaultCircuitBreaker, restrictedCall);
    		return Try.of(callable::call).recover(fallback).get(); // 执行逻辑和降级处理
    	}

以上就是执行的过程, 可以看出`resilience4j`是要轻量很多, 而`hystrix`的源码看下去能看上一整天 /(ㄒoㄒ)/~~

接下来是响应式接口的使用

@Autowired
        private WebClient webClient; //对应restTemplate响应式版本
        @Autowired
        private ReactiveCircuitBreakerFactory rcbFactory; //响应式熔断工厂
    
    public Mono<String> slow1() { 
            return webClient.get().uri("/slow").retrieve() // 拿到DefaultResponseSpec 
            .bodyToMono(String.class) // 转换Mono<ClientResponse> -> Mono<String>
            .transform(  // 转换 Mono<String> 增加 熔断逻辑
                    it -> rcbFactory.create("slow").run(it, throwable -> { 
                        return Mono.just("fallback");
                    }));
        }

具体还是看下 run 的实现

//org.springframework.cloud.netflix.hystrix.ReactiveHystrixCircuitBreaker#run(reactor.core.publisher.Mono<T>, java.util.function.Function<java.lang.Throwable,reactor.core.publisher.Mono<T>>)
    public <T> Mono<T> run(Mono<T> toRun, Function<Throwable, Mono<T>> fallback) { 
    		HystrixObservableCommand<T> command = createCommand(toRun, fallback);
    		
    		return Mono.create(s -> { 
    		// 这里也能看到 hystrix 底层就是响应式的, 只不过可以通过阻塞方法适配成非响应式的调用, 不过因为spring用reactor 而 hystrix 用的rxjava, 所以需要通过标准接口(reactive stream规范)进行转换一下
    			Subscription sub = command.toObservable().subscribe(s::success, s::error,
    					s::success);
    			s.onCancel(sub::unsubscribe);
    		});
    	}
    
    // org.springframework.cloud.circuitbreaker.resilience4j.ReactiveResilience4JCircuitBreaker#run(reactor.core.publisher.Mono<T>, java.util.function.Function<java.lang.Throwable,reactor.core.publisher.Mono<T>>)
    public <T> Mono<T> run(Mono<T> toRun, Function<Throwable, Mono<T>> fallback) { 
    		io.github.resilience4j.circuitbreaker.CircuitBreaker defaultCircuitBreaker = registry
    				.circuitBreaker(id, config.getCircuitBreakerConfig());
    		circuitBreakerCustomizer
    				.ifPresent(customizer -> customizer.customize(defaultCircuitBreaker));
    		Mono<T> toReturn = toRun
    				.transform(CircuitBreakerOperator.of(defaultCircuitBreaker))
    				.timeout(config.getTimeLimiterConfig().getTimeoutDuration()) // reactor 提供的超时设置
    				.doOnError(TimeoutException.class,
    						t -> defaultCircuitBreaker.onError(config.getTimeLimiterConfig()
    								.getTimeoutDuration().toMillis(), TimeUnit.MILLISECONDS,
    								t));
    		if (fallback != null) { 
    			toReturn = toReturn.onErrorResume(fallback);
    		}
    		return toReturn;
    	}

可以看出来resilience4j的响应式还是通过spring去适配的, 本身没有提供响应式的api.

## 自动装配 ##

先看下 hystrix 的自动装配

@Configuration(proxyBeanMethods = false) // sb新特性, 减少运行时的类生成
    @ConditionalOnClass({  Hystrix.class }) //必须引入hystrix依赖
    @ConditionalOnProperty(name = "spring.cloud.circuitbreaker.hystrix.enabled",
    		matchIfMissing = true) // 配置开启, 默认就是开启的
    public class HystrixCircuitBreakerAutoConfiguration { 
    	// 对应的工厂类
    	@Bean 
    	@ConditionalOnMissingBean(CircuitBreakerFactory.class)
    	public CircuitBreakerFactory hystrixCircuitBreakerFactory() { 
    		return new HystrixCircuitBreakerFactory();
    	}
    	// 响应式的工厂类
    	@Bean
    	@ConditionalOnMissingBean(ReactiveCircuitBreakerFactory.class)
    	@ConditionalOnClass(
    			name = {  "reactor.core.publisher.Mono", "reactor.core.publisher.Flux" })
    	public ReactiveHystrixCircuitBreakerFactory reactiveHystrixCircuitBreakerFactory() { 
    		return new ReactiveHystrixCircuitBreakerFactory();
    	}
    	// 自定义定制
    	@Configuration(proxyBeanMethods = false)
    	protected static class HystrixCircuitBreakerCustomizerConfiguration { 
    
    		@Autowired(required = false)
    		private List<Customizer<HystrixCircuitBreakerFactory>> customizers = new ArrayList<>();
    
    		@Autowired(required = false)
    		private HystrixCircuitBreakerFactory factory;
    
    		@PostConstruct
    		public void init() { 
    			customizers.forEach(customizer -> customizer.customize(factory));
    		}
    
    	}
    	// 自定义定制
    	@Configuration(proxyBeanMethods = false)
    	@ConditionalOnClass(
    			name = {  "reactor.core.publisher.Mono", "reactor.core.publisher.Flux" })
    	protected static class ReactiveHystrixCircuitBreakerCustomizerConfiguration { 
    
    		@Autowired(required = false)
    		private List<Customizer<ReactiveHystrixCircuitBreakerFactory>> customizers = new ArrayList<>();
    
    		@Autowired(required = false)
    		private ReactiveHystrixCircuitBreakerFactory factory;
    
    		@PostConstruct
    		public void init() { 
    			customizers.forEach(customizer -> customizer.customize(factory));
    		}
    
    	}
    
    }

resilience4j的自动配置也是类似

@Configuration(proxyBeanMethods = false)
    @ConditionalOnProperty(name = "spring.cloud.circuitbreaker.resilience4j.enabled",
    		matchIfMissing = true) //默认开启
    public class Resilience4JAutoConfiguration { 
    	// 工厂类
    	@Bean
    	@ConditionalOnMissingBean(CircuitBreakerFactory.class)
    	public Resilience4JCircuitBreakerFactory resilience4jCircuitBreakerFactory() { 
    		return new Resilience4JCircuitBreakerFactory();
    	}
    	// 自定义配置,定制的
    	@Configuration(proxyBeanMethods = false)
    	@ConditionalOnMissingClass({ 
    			"io.github.resilience4j.micrometer.tagged.TaggedCircuitBreakerMetrics",
    			"io.micrometer.core.instrument.MeterRegistry" })
    	public static class Resilience4JCustomizerConfiguration { 
    
    		@Autowired(required = false)
    		private List<Customizer<Resilience4JCircuitBreakerFactory>> customizers = new ArrayList<>();
    
    		@Autowired(required = false)
    		private Resilience4JCircuitBreakerFactory factory;
    
    		@PostConstruct
    		public void init() { 
    			customizers.forEach(customizer -> customizer.customize(factory));
    		}
    
    	}
    	// 用于监控的
    	@Configuration(proxyBeanMethods = false)
    	@ConditionalOnBean({  MeterRegistry.class }) //springboot actuator这个依赖要有
    	@ConditionalOnClass(name = { 
    			"io.github.resilience4j.micrometer.tagged.TaggedCircuitBreakerMetrics" })
    	public static class MicrometerResilience4JCustomizerConfiguration { 
    
    		@Autowired(required = false)
    		private List<Customizer<Resilience4JCircuitBreakerFactory>> customizers = new ArrayList<>();
    
    		@Autowired(required = false)
    		private Resilience4JCircuitBreakerFactory factory;
    
    		@Autowired
    		private MeterRegistry meterRegistry;
    
    		@PostConstruct
    		public void init() { 
    			customizers.forEach(customizer -> customizer.customize(factory));
    			if (factory != null) { 
    				TaggedCircuitBreakerMetrics
    						.ofCircuitBreakerRegistry(factory.getCircuitBreakerRegistry())
    						.bindTo(meterRegistry);
    			}
    		}
    
    	}
    
    }

基本大同小异

## 总结 ##

在H 版本以后, 熔断提供了新的统一接口, 这样就可以快速的替换底层的实现.

关于响应式: 因为spring开始推广响应式编程(reactor), 所以在各个地方包块webflux, webclient等都能见到响应式的使用. 响应式可以大大降低并发编程的困难, 但是它繁多的api和复杂的底层实现也会让学习的曲线很陡. 当然响应式能不能提升性能这个就仁者见仁了.

关于hystrix vs resilience4j:　两者都是很优秀的熔断框架，简单来说功能上两者都有熔断的功能, 超时的功能, 降级的功能; 架构上resilience4j会稍微轻量很多; 性能上, resilience4j因为它的轻量级的实现, 所以性能会稍好(即使hystrix 底层用的rxjava, 但是我的理解是响应式并不能提升性能); 隔离性上我觉得hystrix基于线程池的隔离会更加好点.

分析到此结束了, 欢迎大家能互相交流