微服务集群中服务端和客户端都支持集群部署。而庞大的集群中部分服务出现问题不可避免如何减少故障的影响以及确保集群高可用是一个需要解决的问题。Hystrix就是Spring Cloud提供的一个集群保护框架。Hystrix 是一个延迟和容错库旨在隔离远程系统、服务和第三方库的访问点停止级联故障并在故障不可避免的复杂分布式系统中实现弹性。Hystrix通过添加延迟阈值和容错逻辑帮助控制分布式系统组建的交互。一、实现的功能1、当所依赖的网络服务发生延迟或者失败时对访问的客户端程序进行保护。如下图基础服务无法访问则立即执行回退逻辑。2、在分布式系统中停止级联故障。3、网络恢复后可以快速恢复客户端的访问能力。4、调用失败时执行服务回退。5、可以支持实时监控、报警以及其他操作。二、运行流程第一步在命令开始执行时会准备一些工作比如命令会创建相应的线程池。第二步判断是否打开了缓存打开缓存的情况会直接查找缓存并返回结果。第三步判断断路器是否打开打开的话说明链路不可用直接执行回退响应用户请求。第四步判断线程池、信号量计数器等条件如线程池超负荷则直接执行回退方法否则就去执行命令的内容。第五步执行命令计算是否要对断路器进行处理。执行完毕后如果满足一定的条件则需求开启断路器。如果执行成功则返回结果否则执行回退。整个Hystrix运行流程中断路器开启状态是最核心的。三、使用要点1、命令属性配置超时配置命令组名称GroupKey,必须配置。默认情况下全局维护的线程池Map作为该值的key,该Map的value为执行命令的线程池。命令名称CommanKey可选配置。线程池名称ThreadPoolKey,指定线程池的key后全局维护的线程池Map将作为该值的Key。2、回退触发回退的三种情况a、断路器被打开b、线程池、信号量满载c、实际执行失败满足三种条件之一即可进行回退。执行父类HystrixCommand的getFallback方法即可实现回退。回退模式回退机制灵活,特别注意的是如果A和B两个命令时在一个事务内的在控制类中调用回退方法A和B不能有回退方法不然就不能执行控制类的回退方法了。断路器开启如果断路器开启了将不再执行命令也不更新链路的健康状况在一段时间内一直在回退。触发断路器开启的需要同时满足两个条件第一个整个链路达到一定阈值。默认10秒内产生20个请求则符合该条件。第二个达到第一个条件的同时如果请求失败的百分比大于阈值则开启断路器。失败百分比默认为50%。断路器关闭断路器打开后在一段时间内默认为5秒命令不在执行而是在不断触发回退这段时间被称为休眠期。休眠期结束后此时断路器处于半开状态非开启也非关闭hystrix会尝试执行一次命令如果命令执行成功那么将关闭断路器并清空链路的健康信息如果执行失败则断路器依旧保持打开的状态。隔离机制除了断路器要处于关闭状态外命令真正执行还需要考虑命令的线程池或者信号量是否满载。如果满载则不会执行命令而是触发回退从机制上控制了命令的执行实现了错误的隔离。隔离策略THRED:默认策略有线程池来决定是否执行命令如果线程池满载则不会执行命令。默认线程池大小为10。SEMAPHORE:由信号量来决定命令的执行。当请求并发数高于阈值时不再执行命令。默认最大并发数为10.信号量策略开销小但是不支持超时和异步一般不建议使用。合并请求在线程隔离策略下可以将一段时间内的同类请求URL一致参数不同收集到一个命令中执行。这样做的好处节省线程开销、减少网络连接。合并请求的三个条件a、需要一个执行请求的命令将全部的参数进行整理然后调用外部服务。b、需求一个合并处理器用于收集请求以及处理结果c、外部接口提供支持。提供批量查询等接口import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Random; import java.util.UUID; import java.util.concurrent.Future; import com.netflix.config.ConfigurationManager; import com.netflix.hystrix.HystrixCollapser; import com.netflix.hystrix.HystrixCollapser.CollapsedRequest; import com.netflix.hystrix.HystrixCommand; import com.netflix.hystrix.HystrixCommandGroupKey; import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext; public class CollapseTest { public static void main(String[] args) throws Exception { // 收集 1 秒内发生的请求合并为一个命令执行 ConfigurationManager.getConfigInstance().setProperty( hystrix.collapser.default.timerDelayInMilliseconds, 1000); // 请求上下文 HystrixRequestContext context HystrixRequestContext .initializeContext(); // 创建请求合并处理器 MyHystrixCollapser c1 new MyHystrixCollapser(Angus); MyHystrixCollapser c2 new MyHystrixCollapser(Crazyit); MyHystrixCollapser c3 new MyHystrixCollapser(Sune); MyHystrixCollapser c4 new MyHystrixCollapser(Paris); // 异步执行 FuturePerson f1 c1.queue(); FuturePerson f2 c2.queue(); FuturePerson f3 c3.queue(); FuturePerson f4 c4.queue(); System.out.println(f1.get()); System.out.println(f2.get()); System.out.println(f3.get()); System.out.println(f4.get()); context.shutdown(); } /** * 合并执行的命令类 */ static class CollapserCommand extends HystrixCommandMapString, Person { // 请求集合第一个类型是单个请求返回的数据类型第二是请求参数的类型 CollectionCollapsedRequestPerson, String requests; private CollapserCommand( CollectionCollapsedRequestPerson, String requests) { super(Setter.withGroupKey(HystrixCommandGroupKey.Factory .asKey(ExampleGroup))); this.requests requests; } Override protected MapString, Person run() throws Exception { System.out.println(收集参数后执行命令参数数量 requests.size()); // 处理参数 ListString personNames new ArrayListString(); for(CollapsedRequestPerson, String request : requests) { personNames.add(request.getArgument()); } // 调用服务此处模拟调用根据名称获取Person的Map MapString, Person result callService(personNames); return result; } // 模拟服务返回 private MapString, Person callService(ListString personNames) { MapString, Person result new HashMapString, Person(); for(String personName : personNames) { Person p new Person(); p.id UUID.randomUUID().toString(); p.name personName; p.age new Random().nextInt(30); result.put(personName, p); } return result; } } static class Person { String id; String name; Integer age; public String toString() { // TODO Auto-generated method stub return id: id , name: name , age: age; } } /** * 合并处理器 * 第一个类型为批处理返回的结果类型 * 第二个为单请求返回的结果类型 * 第三个是请求参数类型 */ static class MyHystrixCollapser extends HystrixCollapserMapString, Person, Person, String { String personName; public MyHystrixCollapser(String personName) { this.personName personName; } Override public String getRequestArgument() { return personName; } Override protected HystrixCommandMapString, Person createCommand( CollectionCollapsedRequestPerson, String requests) { return new CollapserCommand(requests); } Override protected void mapResponseToRequests(MapString, Person batchResponse, CollectionCollapsedRequestPerson, String requests) { // 让结果与请求进行关联 for (CollapsedRequestPerson, String request : requests) { // 获取单个响应返回的结果 Person singleResult batchResponse.get(request.getArgument()); // 关联到请求中 request.setResponse(singleResult); } } } }其中CollapserCommand是命令内部类MyHystrixCollapser是合并处理器。请求缓存针对同一次请求中多个地方调用同一个接口的情况可以考虑使用缓存。缓存被打开后下次命令不会执行而是直接从缓存中获取响应并返回。开启在命令中重写父类的getCacheKey方法即可。合并请求、请求缓存需要在一次请求过程中才能实现因此需要先初始化请求上下文。四、单独使用Hystrix1、服务提供方import javax.servlet.http.HttpServletRequest; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.RestController; RestController public class MyController { GetMapping(/normalHello) ResponseBody public String normalHello(HttpServletRequest request) { return Hello World; } GetMapping(/errorHello) ResponseBody public String errorHello(HttpServletRequest request) throws Exception { // 模拟需要处理10秒 Thread.sleep(10000); return Error Hello World; } }2、服务调用方第一步添加hystrix依赖dependency groupIdcom.netflix.hystrix/groupId artifactIdhystrix-core/artifactId version1.5.12/version /dependency dependency groupIdorg.slf4j/groupId version1.7.25/version artifactIdslf4j-log4j12/artifactId /dependency dependency groupIdcommons-logging/groupId artifactIdcommons-logging/artifactId version1.2/version /dependency dependency groupIdorg.apache.httpcomponents/groupId artifactIdhttpclient/artifactId version4.5.2/version /dependency第二步编写命令类需要继承HystrixCommandimport org.apache.http.HttpResponse; import org.apache.http.client.methods.HttpGet; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; import org.apache.http.util.EntityUtils; import com.netflix.hystrix.HystrixCommand; import com.netflix.hystrix.HystrixCommandGroupKey; /** * 调用一个正常的服务 * */ public class HelloCommand extends HystrixCommandString { private String url; CloseableHttpClient httpclient; public HelloCommand(String url) { // 调用父类的构造器设置命令组的key默认用来作为线程池的key super(HystrixCommandGroupKey.Factory.asKey(ExampleGroup)); // 创建HttpClient客户端 this.httpclient HttpClients.createDefault(); this.url url; } protected String run() throws Exception { try { // 调用 GET 方法请求服务 HttpGet httpget new HttpGet(url); // 得到服务响应 HttpResponse response httpclient.execute(httpget); // 解析并返回命令执行结果 return EntityUtils.toString(response.getEntity()); } catch (Exception e) { e.printStackTrace(); } return ; } protected String getFallback() { System.out.println(执行 HelloCommand 的回退方法); return error; } }第三步编写调用类public class HelloMain { public static void main(String[] args) { // 请求正常的服务 String url http://localhost:8080/normalHello; HelloCommand command new HelloCommand(url ); String result command.execute(); System.out.println(请求正常的服务结果 result); // 请求异常的服务 String url http://localhost:8080/errorHello; command new HelloCommand(url ); result command.execute(); System.out.println(请求异常的服务结果 result); } }五、在Spring Cloud中使用Hystrix1、整合Hystrix流程(服务调用方)第一步添加Hystrix依赖添加eureka、ribbon以及hystrixdependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-config/artifactId /dependency dependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-eureka/artifactId /dependency dependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-ribbon/artifactId /dependency dependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-hystrix/artifactId /dependency第二步在启动类中加入启用断路器的注解import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.web.servlet.ServletComponentScan; import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker; import org.springframework.cloud.client.discovery.EnableDiscoveryClient; import org.springframework.cloud.client.loadbalancer.LoadBalanced; import org.springframework.cloud.netflix.feign.EnableFeignClients; import org.springframework.context.annotation.Bean; import org.springframework.web.client.RestTemplate; SpringBootApplication EnableDiscoveryClient EnableCircuitBreaker public class InvokerApplication { LoadBalanced Bean public RestTemplate getRestTemplate() { return new RestTemplate(); } public static void main(String[] args) { SpringApplication.run(InvokerApplication.class, args); } }第三步编写服务类import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.springframework.web.client.RestTemplate; import com.netflix.hystrix.contrib.javanica.annotation.DefaultProperties; import com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand; import com.netflix.hystrix.contrib.javanica.annotation.HystrixProperty; Component public class PersonService { Autowired private RestTemplate restTemplate; HystrixCommand(fallbackMethod getPersonFallback) public Person getPerson(Integer id) { // 使用RestTemplate调用Eureka服务 Person p restTemplate.getForObject( http://spring-hystrix-provider/person/{personId}, Person.class, id); return p; } /** * 回退方法返回一个默认的Person */ public Person getPersonFallback(Integer id) { Person p new Person(); p.setId(0); p.setName(Crazyit); p.setAge(-1); p.setMessage(request error); return p; } /** * 测试配置对3个key进行命名 * 设置命令执行超时时间为1000毫秒 * 设置命令执行的线程池大小为1 */ HystrixCommand( fallbackMethodtestConfigFallback, groupKeyMyGroup, commandKeyMyCommandKey, threadPoolKeyMyCommandPool, commandProperties{ HystrixProperty(name execution.isolation.thread.timeoutInMilliseconds, value 1000) }, threadPoolProperties{ HystrixProperty(name coreSize, value 1) }) public String testConfig() { try { Thread.sleep(500); } catch (Exception e) { e.printStackTrace(); } return ok; } public String testConfigFallback() { return error; } /** * 声明了忽略MyException如果方法抛出MyException则不会触发回退 */ HystrixCommand(ignoreExceptions {MyException.class}, fallbackMethodtestExceptionFallBack) public String testException() { throw new MyException(); } public String testExceptionFallBack() { return error; } }第四步编写控制器package org.crazyit.cloud; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Future; import org.crazyit.cloud.cache.CacheService; import org.crazyit.cloud.collapse.CollapseService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RestController; RestController Configuration public class InvokerController { Autowired private PersonService personService; Autowired private CacheService cacheService; Autowired private CollapseService collapseService; RequestMapping(value /router/{personId}, method RequestMethod.GET, produces MediaType.APPLICATION_JSON_VALUE) public Person router(PathVariable Integer personId) { Person p personService.getPerson(personId); return p; } RequestMapping(value /testConfig, method RequestMethod.GET) public String testConfig() { String result personService.testConfig(); return result; } RequestMapping(value /testException, method RequestMethod.GET) public String testException() { String result personService.testException(); return result; } RequestMapping(value /cache1/{personId}, method RequestMethod.GET, produces MediaType.APPLICATION_JSON_VALUE) public Person testCacheResult(PathVariable Integer personId) { // 调用多次服务 for(int i 0; i 3; i) { Person p cacheService.getPerson(personId); System.out.println(控制器调用服务 i); } return new Person(); } RequestMapping(value /cache2, method RequestMethod.GET, produces MediaType.APPLICATION_JSON_VALUE) public String testCacheRemove() { for(int i 0; i 3; i) { cacheService.cacheMethod(a); System.out.println(控制器调用服务 i); } // 清空缓存 cacheService.updateMethod(a); System.out.println( 清空了缓存); // 再执行多次 for(int i 0; i 3; i) { cacheService.cacheMethod(a); System.out.println(控制器调用服务 i); } return ; } RequestMapping(value /collapse, method RequestMethod.GET, produces MediaType.APPLICATION_JSON_VALUE) public String testCollapse() throws Exception { // 连续执行3次请求 FuturePerson f1 collapseService.getSinglePerson(1); FuturePerson f2 collapseService.getSinglePerson(2); FuturePerson f3 collapseService.getSinglePerson(3); Person p1 f1.get(); Person p2 f2.get(); Person p3 f3.get(); System.out.println(p1.getId() --- p1.getName()); System.out.println(p2.getId() --- p2.getName()); System.out.println(p3.getId() --- p3.getName()); return ; } }2、常用注解缓存注解在使用CacheResult和HystrixCommand共同注释服务方法如下CacheResult HystrixCommand public Person getPerson(Integer id) { System.out.println(执行 getPerson 方法); Person p new Person(); p.setId(id); p.setName(angus); return p; }合并请求注解使用HystrixCollapser修饰方法如下图在一秒内collapserProperties属性指定收集到的请求交由指定了方法getPersonsbatchMethod属性指定进行批处理。Component public class CollapseService { // 配置收集1秒内的请求 HystrixCollapser(batchMethod getPersons, collapserProperties { HystrixProperty(name timerDelayInMilliseconds, value 1000) } ) public FuturePerson getSinglePerson(Integer id) { System.out.println(执行单个获取的方法); return null; } HystrixCommand public ListPerson getPersons(ListInteger ids) { System.out.println(收集请求参数数量 ids.size()); ListPerson ps new ArrayListPerson(); for (Integer id : ids) { Person p new Person(); p.setId(id); p.setName(crazyit); ps.add(p); } return ps; } }