搜文章
推荐 原创 视频 Java开发 iOS开发 前端开发 JavaScript开发 Android开发 PHP开发 数据库 开发工具 Python开发 Kotlin开发 Ruby开发 .NET开发 服务器运维 开放平台 架构师 大数据 云计算 人工智能 开发语言 其它开发
Lambda在线 > JayiceTalk > 微服务容错框架:Hystrix实现服务熔断、降级、限流

微服务容错框架:Hystrix实现服务熔断、降级、限流

JayiceTalk 2020-06-29

业务背景


在微服务架构体系下,服务间不可避免地会发生依赖关系,一般来说会通过REST Api来进行通信,这里先盗一个图来举例说明一个具体的业务场景(逃):


比如一个商城系统的微服务化结构,订单、商品、库存这三个服务是紧密依赖的,在理想情况下,什么问题都不发生当然是最好的。但服务运行期间难免会出现各种问题,如网络阻塞,延迟过高(比如因为内存泄露导致的Full GC次数飙高) ,甚至服务直接挂掉(比如流量激增把服务打挂了)等情况都是很有可能发生的。倘若库存服务挂掉了,那对于所有对库存服务有依赖关系的服务都会受到很大影响,最终甚至会扩散到整个微服务体系,这种就称之为雪崩效应。


因此,在某一个服务发生故障时,我们要及时对该服务的故障进行隔离,不能让其扩散到整个微服务体系中。因为,为了搭建一个稳定且可靠的微服务系统,我们就需要给系统加上自我保护,出现故障自动隔离的能力。而**Hystrix**就能做到这一点


什么是Hystrix


Hystrix是Netflix开源的一款分布式容错框架,Netflix旗下还有Eureka,Zuul等优秀的分布式开源项目,Spring Cloud也提供了对Netflix中部门项目的支持,成为了SpringCloud下的一些子项目 。


Hystrix的功能:


  • 阻止故障的连锁反应,实现熔断

  • 快速失败,实现优雅降级

  • 提供实时的监控和告警


Hystrix简单实现


  
    
    
  
public class QueryUserIdCommand extends HystrixCommand < Integer > {
private final static Logger logger = LoggerFactory . getLogger ( QueryUserIdCommand . class ) ;
private UserServiceProvider userServiceProvider ;

public QueryUserAgeCommand ( UserServiceProvider userServiceProvider ) {
super ( Setter . withGroupKey ( HystrixCommandGroupKey . Factory . asKey ( "userService" ) )
. andCommandKey ( HystrixCommandKey . Factory . asKey ( "queryByUserId" ) )
. andCommandPropertiesDefaults ( HystrixCommandProperties . Setter ( )
. withCircuitBreakerRequestVolumeThreshold ( 10 ) //至少有10个请求,熔断器才会开始进行错误率计算
. withCircuitBreakerSleepWindowInMilliseconds ( 5000 ) //熔断器中断请求,5秒后会进入一个半打开状态,放开部分请求去进行重试
. withCircuitBreakerErrorThresholdPercentage ( 50 ) //错误率达到50%就开启熔断保护
. withExecutionTimeoutEnabled ( true ) )
. andThreadPoolPropertiesDefaults ( HystrixThreadPoolProperties
. Setter ( ) . withCoreSize ( 10 ) ) ) ;
this . userServiceProvider = userServiceProvider ;
}

@Override
protected Integer run ( ) {
return userServiceProvider . queryByUserId ( ) ;
}

@Override
protected Integer getFallback ( ) {
return - 1 ;
} }


发起请求


  
    
    
  
Integer res = new QueryUserIdCommand ( userServiceProvider ) . execute ( ) ;
log
. info ( "result:{}" , res ) ;


访问接口,正常情况下,会返回正确的信息,当把UserServiceProvider所依赖的服务的接口改为直接抛出一个异常,就会发现总是返回-1了。这样就做到了对错误进行隔离。


Hystrix容错


那接下来从三个角度来聊一下Hystrix提供的容错功能,分别是资源隔离熔断降级


资源隔离


我们之前也讨论过,微服务体系中,各个服务之间都通过REST Api来进行调用,从而建立依赖关系。倘若该服务调用和业务代码在同一个线程会中执行的话,如果 api在调用的时候出现了网络堵塞等情况,那么不仅会对业务代码进行阻塞,也会对后面的请求造成阻塞,因为线程池的线程数是额定的。所以,Hystrix也提供了资源隔离的机制,主要是线程隔离和信号量隔离


资源隔离-线程池


刚才我们简单地应用了一下Hystrix,很明显看到我们得先实现一个自己的HystrixCommand,然后把服务调用的操作封装在这个类里面。而实际上,线程级别的资源隔离就是在HystrixCommand中实现。Hystrix会给每一个Command分配一个单独的线程池,这样在进行单个服务调用的时候,就可以在独立的线程池里面进行,而不会对其他线程池造成影响。


Hystrix通过一个ConcurrentHashMap来维护这些线程池:


  
    
    
  
final static ConcurrentHashMap < String , HystrixThreadPool > threadPools = new ConcurrentHashMap < String , HystrixThreadPool > ( ) ; //其他代码 if ( ! threadPools . containsKey ( key ) ) {
threadPools
. put ( key , new HystrixThreadPoolDefault ( threadPoolKey , propertiesBuilder ) ) ; }


线程隔离的优点:


  • 保护当前应用免受来自其他服务故障的影响,最终提高整个微服务体系的稳定性


  • 可以对一个Command里面的线程调用参数进行单独设置,而不影响其他Command,如果使用Spring Cloud Hystrix的话,那就是:



  
    
    
  
@HystrixCommand ( groupKey = "UserGroup" , commandKey = "GetUserByIdCommand"
commandProperties
= {
@HystrixProperty ( name = "execution.isolation.thread.timeoutInMilliseconds" , value = "100" ) , //超时时间,单位毫秒。超时进fallback
@HystrixProperty ( name = "circuitBreaker.requestVolumeThreshold" , value = "10" ) , //判断熔断的最少请求数,默认是10;只有在一定时间内请求数量达到该值,才会进行成功率的计算
@HystrixProperty ( name = "circuitBreaker.errorThresholdPercentage" , value = "10" ) , //熔断的阈值默认值50,表示在一定时间内有50%的请求处理失败,会触发熔断
} ,
threadPoolProperties
= {
@HystrixProperty ( name = "coreSize" , value = "30" ) ,
@HystrixProperty ( name = "maxQueueSize" , value = "101" ) ,
@HystrixProperty ( name = "keepAliveTimeMinutes" , value = "2" ) ,
@HystrixProperty ( name = "queueSizeRejectionThreshold" , value = "15" ) ,
@HystrixProperty ( name = "metrics.rollingStats.numBuckets" , value = "12" ) ,
@HystrixProperty ( name = "metrics.rollingStats.timeInMilliseconds" , value = "1440" )
} )


  • 线程池如果处于饱和态,还往其中添加请求的话,会直接触发fallback,所以还有限流的作用


而为每一个Command配置一个线程池的缺点就是增加开销,线程一多起来,就增加了调度,上下文切换等额外的开销。但通常情况下,引入线程池的开销是很小的,相对于它带来的好处,大多数情况还是非常乐于进行这样的交换,但如果是一些耗时非常低的请求,比如直接从缓存中获取数据然后返回,引入线程池带来的影响可能会显得比较大,这种时候可以用另一种方法,就是**信号量隔离**


资源隔离-信号量隔离


信号量隔离本质上并没有做到资源隔离,只是起到了限流的作用,从而防止出现线程大面积阻塞,功能和JUC下的那个Semaphare类差不多,但区别就是,没有执行条件的线程在这里会直接调用fallback,而不是阻塞。启用方法也很简单,在实现Hystrix的时候声明一下即可:



  
    
    
  
public class QueryUserIdCommand extends HystrixCommand < Integer > {
private final static Logger logger = LoggerFactory . getLogger ( QueryUserIdCommand . class ) ;
private UserServiceProvider userServiceProvider ;
public QueryUserAgeCommand ( UserServiceProvider userServiceProvider ) {
super ( Setter . withGroupKey ( HystrixCommandGroupKey . Factory . asKey ( "userService" ) )
. andCommandKey ( HystrixCommandKey . Factory . asKey ( "queryByUserId" ) )
. andCommandPropertiesDefaults ( HystrixCommandProperties . Setter ( )
. withCircuitBreakerRequestVolumeThreshold ( 10 ) //至少有10个请求,熔断器才会开始进行错误率计算
. withCircuitBreakerSleepWindowInMilliseconds ( 5000 ) //熔断器中断请求,5秒后会进入一个半打开状态,放开部分请求去进行重试
. withCircuitBreakerErrorThresholdPercentage ( 50 ) //错误率达到50%就开启熔断保护
. withExecutionIsolationStrategy ( HystrixCommandProperties . ExecutionIsolationStrategy . SEMAPHORE ) //here!!!!!
. withExecutionTimeoutEnabled ( true ) )
. andThreadPoolPropertiesDefaults ( HystrixThreadPoolProperties
. Setter ( ) . withCoreSize ( 10 ) ) ) ;
this . userServiceProvider = userServiceProvider ;
}

@Override
protected Integer run ( ) {
return userServiceProvider . queryByUserId ( ) ;
}

@Override
protected Integer getFallback ( ) {
return - 1 ;
} }


总结一下,资源隔离就大概是这两种了,大多数情况用的还是线程隔离居多,毕竟它才是真正意义上的资源隔离,而信号量隔离只是起到一个限流的作用而已


熔断


一个服务出现故障之后,防止这个故障蔓延到所有依赖它的服务,这就是熔断。一开始的时候我们也已经写过代码来简单实现熔断了,所以直接介绍几个重要的熔断相关的参数:


  
    
    
  
- circuitBreaker . enabled:是否启用熔断器,默认 true - circuitBreaker . forceOpen:强制打开熔断器,默认 false
- circuitBreaker . forceClosed:强制关闭熔断器,默认 false

- circuitBreaker . errorThresholdPercentage:错误率,默认 50 % 。在一段时间内,服务调用超时或者失败率超过 50 % ,则打开熔断器

- circuitBreaker . requestVolumeThreshold:默认 20 。意思为在一段时间内要有 20 个及以上的请求才会去计算错误率。比如只来了 19 个请求,就算全失败了,那也不算错误率 100 % - circuitBreaker . sleepWindowInMilliseconds:半开状态试探睡眠时间,默认为 5000 ms。也就是熔断器打开 5 s后,开始半打开状态,放出一点请求去调用服务,试探一下能否成功




来写代码测试一下其中几个参数:


  
    
    
  
@HystrixCommand ( groupKey = "productStockOpLog" , commandKey = "addProductStockOpLog" , fallbackMethod = "addProductStockOpLogFallback" ,
commandProperties
= {
@HystrixProperty ( name = "execution.isolation.thread.timeoutInMilliseconds" , value = "400" ) , //指定多久超时,单位毫秒。超时进fallback
@HystrixProperty ( name = "circuitBreaker.requestVolumeThreshold" , value = "10" ) , //判断熔断的最少请求数,默认是10;只有在一个统计窗口内处理的请求数量达到这个阈值,才会进行熔断与否的判断
@HystrixProperty ( name = "circuitBreaker.errorThresholdPercentage" , value = "10" ) , //判断熔断的阈值,默认值50,表示在一个统计窗口内有50%的请求处理失败,会触发熔断
} )


我先设置了超时时间为400ms。然后在服务提供接口方法改成这样:


  
    
    
  
@Service public class ServiceProviderImpl implements ServiceProvider {
private int c = 0 ;

@Override
public Integer service ( ) {

if ( c < 10 ) {
try {
Thread . sleep ( 500 ) ;
} catch ( InterruptedException e ) {
}
}
return c ;
} }


实现也很简单,前十次调用都睡500ms,也就是说,前十次调用肯定是超时的。


而在客户端,实现如下:


  
    
    
  
@Test public void test ( ) throws InterruptedException {
for ( int i = 1 ; i < 15 ; i ++ ) {
HystrixCommand < Integer > command = new QueryByUserIdCommand ( userServiceProvider ) ;
Integer r = command . execute ( ) ;
String res = r == - 1 ? "fallback" : "success" ;
System . out . println ( "第" i + "次调用,结果为" + method ) ;
}
//等待5.5s,让熔断器进入半打开状态
Thread . sleep ( 5500 ) ;
for ( int i = 15 ; i < 20 ; i ++ ) {
HystrixCommand < Integer > command = new UserByOrderIdCommand ( userServiceProvider ) ;
Integer r = command . execute ( ) ;
String method = r == - 1 ? "fallback" : "success" ;
System . out . println ( "第" i + "次调用,结果为" + method ) ;
} }


查看执行结果:


  
    
    
  
1 次调用,结果为fallback
2 次调用,结果为fallback
3 次调用,结果为fallback
4 次调用,结果为fallback
5 次调用,结果为fallback
6 次调用,结果为fallback
7 次调用,结果为fallback
8 次调用,结果为fallback
9 次调用,结果为fallback
10 次调用,结果为fallback
11 次调用,结果为fallback
12 次调用,结果为fallback
13 次调用,结果为fallback
14 次调用,结果为fallback
15 次调用,结果为success
16 次调用,结果为success
17 次调用,结果为success
18 次调用,结果为success
19 次调用,结果为success
20 次调用,结果为success


分析可得,1-10次触发fallback的原因是超时

而11-14次的fallback,是因为满足了“请求次数到达10次”和“失败率超过了50%”而触发了熔断器,所以直接快速失败。

待沉睡了5.5秒后,熔断器进入了半打开状态,此时熔断器放开第15个请求去调用服务,发现成功了,此时熔断器关闭,后续都成功了。


降级


熔断和降级其实它们的原理都相似,都是服务调用失败后的进行一些快速失败措施。但它们的出发点不一样,熔断是为了防止异常不扩散,保证系统的稳定性


而降级则是人为操作。在一些流量顶峰期,为了保证某些热门接口的正常运作,有时候会牺牲一些非核心接口,把资源全都让给热点接口,这就是服务降级。每年12306抢票的时候,大家都集中抢购那几个热门车次的车票,而如果此时有其他用户去查询几天后的非热门车票,有可能会查不出来。这就是降级的表现,在秒杀期间,其他不参与秒杀的接口停止服务,把资源都让给参与秒杀的接口。


所以降级的操作其实也很简单,和前面的熔断一样,编写好调用失败的补救逻辑,然后对其他的服务直接停止运行,这样这些接口就无法正常调用,但又不至于直接报错,只是服务水平下降了。




  
    
    
  
@FeignClient ( value = "microservicecloud-test" , fallbackFactory = UserServiceFallbackFactory . class ) public interface UserService {
public boolean add ( User user ) ;

public User queryUserByUserId ( Long id ) ;
}


这里有一个Service接口,可以看到在注解的地方给它配置了一个fallbackFactory


  
    
    
  
@Component public class UserServiceFallbackFactory implements FallbackFactory < UserService > {
@Override
public UserService create ( Throwable throwable )
{
return new UserService ( ) {
@Override
public boolean add ( User user )
{
return false ;
}

@Override
public User queryUserByUserId ( Long id )
{
User user = new User ( ) ;
return user ;
}
} ;
} }


这样就大概完成了一个降级配置了。(其实也可以当熔断来用,是吧?没有任何问题)




Hystrix执行调用的几个方法


Hystrix一共有四种调用方法:


execute


还记得一开始的时候,写了一个简单使用Hystrix的demo,其中调用的代码是:


  
    
    
  
Integer res = new QueryUserIdCommand ( userServiceProvider ) . execute ( ) ;
log
. info ( "result:{}" , res ) ;


调用的是execute方法,它将以同步的方式调用run方法。


那问题来了,如果我系统它以异步的方式执行怎么办?我还得自己封装Callable吗?哈,其实Hystrix已经考虑到这一点了,那就是使用queue方法


queue


这个调用也挺简单的,我直接贴代码


  
    
    
  
Future < Integer > future = new QueryUserIdCommand ( userServiceProvider ) . queue ( ) ; return future . get ( ) ;


虽然Future.get()方法依然是阻塞的,但是服务调用总是不阻塞了嘛~


observe


接下里又来了一个新的需求了,我需要发多个请求,并且每请求得到一个结果,就要做一下相关的处理 。这个无论用queue还是execute都不太方便。那此时我们可以选择使用observe。如果使用observe的话,那么就不再是使用HystrixCommand了,而是HystrixObservableCommand


  
    
    
  
public class UserServiceObserveCommand extends HystrixObservableCommand < String > {

private RestTemplate restTemplate ;

protected HelloServiceObserveCommand ( String commandGroupKey , RestTemplate restTemplate ) {
super ( HystrixCommandGroupKey . Factory . asKey ( commandGroupKey ) ) ;
this . restTemplate = restTemplate ;
}

@Override
protected Observable < String > construct ( ) {       
return Observable . create ( new Observable . OnSubscribe < String > ( ) {
@Override
public void call ( Subscriber < ? super String > subscriber ) {
try {
if ( ! subscriber . isUnsubscribed ( ) ) {
System . out . println ( "服务调用开始!" ) ;

String result = restTemplate . getForEntity ( "http://your-url1" , String . class ) . getBody ( ) ;  
//触发监听
subscriber
. onNext ( result ) ;

String result1 = restTemplate . getForEntity ( "http://your-url2" , String . class ) . getBody ( ) ;
//触发监听
subscriber
. onNext ( result1 ) ;
subscriber
. onCompleted ( ) ;
}
} catch ( Exception e ) {
subscriber
. onError ( e ) ;
}
}
} ) ;
}
  
//降级Fallback
@Override
protected Observable < String > resumeWithFallback ( ) {
return Observable . create ( new Observable . OnSubscribe < String > ( ) {
@Override
public void call ( Subscriber < ? super String > subscriber ) {
try {
if ( ! subscriber . isUnsubscribed ( ) ) {
subscriber
. onNext ( "fallback" ) ;
subscriber
. onCompleted ( ) ;
}
} catch ( Exception e ) {
subscriber
. onError ( e ) ;
}
}
} ) ;
} }


可以看到,每调用完一个方法,就调用一次subscriber.onNext来触发监听。那具体的监听实现怎么写呢?



  
    
    
  
@Test
public void test ( ) throws ExecutionException , InterruptedException {
UserServiceObserveCommand command = new HelloServiceObserveCommand ( "user" , restTemplate ) ;
//observe调用
Observable < String > observable = command . observe ( ) ;
    
//注册监听
observable
. subscribe ( new Observer < String > ( ) {  
//onCompleted监听
@Override
public void onCompleted ( ) {
System . out . println ( "调用完毕!" ) ;
}
      
@Override
public void onError ( Throwable t ) {
t
. printStackTrace ( ) ;
}
      
      
//onNext监听
@Override
public void onNext ( String s ) {
System . out . println ( "调用完了一个服务!" ) ;
}
} ) ;
}


通过以上代码应该大概能知道observe的作用了


toObservable


observe叫做热执行,而toObservable叫做冷执行。有啥区别呢?


observe热执行,无需等客户端注册监听,就可以直接执行call方法里面的内容。相当于:你监听


不监听是你的事情,反正我该调用的都调用了,要是你不想监听我也没办法(逃)


toObservable冷执行,需要等客户端注册监听,才可以执行call方法里面的内容。相当于:你要是不监听的话,那我就不会执行call方法,直到你注册监听,我才开始执行调用




Hystrix作为熔断降级限流的主要功能和用法就是这些。其实Hystrix还能起到监控的功能:Hystrix-DashBorad,这个以后有机会再聊~


版权声明:本站内容全部来自于腾讯微信公众号,属第三方自助推荐收录。《微服务容错框架:Hystrix实现服务熔断、降级、限流》的版权归原作者「JayiceTalk」所有,文章言论观点不代表Lambda在线的观点, Lambda在线不承担任何法律责任。如需删除可联系QQ:516101458

文章来源: 阅读原文

相关阅读

关注JayiceTalk微信公众号

JayiceTalk微信公众号:gh_a34451f4196c

JayiceTalk

手机扫描上方二维码即可关注JayiceTalk微信公众号

JayiceTalk最新文章

精品公众号随机推荐