Hystrix是什么？

在微服务架构中，微服务之间互相依赖较大，相互之间调用必不可免的会失败。但当下游服务A因为瞬时流量导致服务崩溃，其他依赖于A服务的B、C服务由于调用A服务超时耗费了大量的资源，长时间下去，B、C服务也会崩溃。Hystrix就是用来解决服务之间相互调用失败，避免产生蝴蝶效应的熔断器，以及提供降级选项。Hystrix通过隔离服务之间的访问点，阻止它们之间的级联故障以及提供默认选项来实现这一目标，以提高系统的整体健壮性。

用来解决什么问题？

用来避免由于服务之间依赖较重，出现个别服务宕机、停止服务导致大面积服务雪崩的情况。

小试牛刀

服务降级

目前有eureka、zuul、product、cart四个服务。

目的是通过zuul调用product接口，product接口通过feign调用cart服务的接口。product项目调用cart项目超时触发hystrix熔断及服务降级。

现将关键配置及代码陈列如下：

eureka：

server:
  port: 8761

eureka:
  client:
    service-url:
      defaultZone: http://${eureka.instance.hostname:localhost}:${server.port:8761}/eureka

spring:
  application:
    name: eureka
  profiles:
    active: ${boot.profile:dev}

zuul

服务配置：

spring:
  application:
    name: zuul #应用名称
server:
  port: 8080 #应用服务端口

eureka:
  client:
    service-url:
      defaultZone: http://localhost:8761/eureka  #配置eureka 默认分区的地址

ribbon:
  okhttp:
    enabled: true #开启ribbon 使用OKhttp发送http请求

  ReadTimeout: 5000
  ConnectTimeout: 5000

zuul:
  prefix: /api #定义全局路由前缀
  strip-prefix: true #路由到下游服务时开启去除前缀开关

路由配置：

@Configuration
public class ZuuPatternServiceRouteMapperConfiguration {

    /**
     * 获取没有版本号的路由匹配规则bean
     *
     * @return {@link PatternServiceRouteMapper}
     * @date 10:27 AM 2019/1/17
     **/
    @Bean
    public PatternServiceRouteMapper patternServiceRouteMapper() {

        return new PatternServiceRouteMapper("(?<version>v.*$)", "${name}");
    }
}

webMvc配置：

@Configuration
public class WebMvcConfig implements WebMvcConfigurer {

    @Override
    public void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/*")
                .allowedOrigins("*");
    }
}

product

server:
  port: 8082

eureka:
  client:
    service-url:
      defaultZone: http://localhost:8761/eureka
spring:
  application:
    name: product
  profiles:
    active: ${boot.profile:dev}

feign:
  client:
    config:
      default:
        connectTimeout: 1000 #feign调用连接超时时间
        readTimeout: 1000 #feign调用读取超时时间
        loggerLevel: basic #feign调用的log 等级

启动类代码：

@EnableFeignClients
@EnableDiscoveryClient
@SpringBootApplication
@EnableCircuitBreaker
public class ProductApplication {

    public static void main(String[] args) {
        SpringApplication.run(ProductApplication.class, args);
    }
}

controller代码：

@RequestMapping("/product")
@RestController
public class ProductController {

    @Autowired
    private CartFeignClient cartFeignClient; //调用cart服务的feign client

    @HystrixCommand(fallbackMethod = "getDefaultValue") //指定出现异常的默认回退方法
    @PostMapping("/toCart/{productId}")
    public ResponseEntity addCart(@PathVariable("productId") Long productId) throws InterruptedException {
        Thread.sleep(5000); //特意让线程休眠一段时间以触发熔断及服务降级
        Long aLong = cartFeignClient.addCart(productId);
        return ResponseEntity.ok(productId);
    }
    //默认降级方法
    private ResponseEntity getDefaultValue(Long productId) {
        return ResponseEntity.ok(0);
    }
}

cart服务的feign client代码：

@FeignClient(value = "CART") //指定服务名称
public interface CartFeignClient {

    @PostMapping("/cart/{productId}") //指定接口路径
    Long addCart(@PathVariable("productId")Long productId); //参数
}

cart服务没什么特殊配置及代码，只要注册到eureka以及提供接口就可以了。

服务配置：

server:
  port: 8081

eureka:
  client:
    service-url:
      defaultZone: http://localhost:8761/eureka
  instance:
    status-page-url-path: /info
    health-check-url-path: /health

spring:
  application:
    name: cart
  profiles:
    active: ${boot.profile:dev}

启动类代码：

@EnableDiscoveryClient
@SpringBootApplication
public class CartApplication {

    public static void main(String[] args) {
        SpringApplication.run(CartApplication.class, args);
    }

}

接口：

@RestController
@RequestMapping("/cart")
@Api(value = "购物车", tags = {"购物车"}) //swagger
public class CartControllrr {

    @ApiOperation("添加商品到购物车") //swagger 
    @ApiImplicitParam(name = "productId",value = "productId",required = true,paramType ="path",dataType = "String")
    @PostMapping("/{productId}")
    public ResponseEntity addCart(@PathVariable("productId") Long productId) {
        System.out.println(productId);
        return ResponseEntity.ok(productId);
    }
}

现在将服务全部启动，并通过zuul调用接口即可。

原先直接调用product接口地址为：http://localhost:8082/product/toCart/1

经过zuul后的地址为：http://localhost:8080/api/product/product/toCart/1

正常调用结果返回为1，

由于product处理代码中加入了sleep，导致product服务调用超时，触发hystrix熔断及服务降级，调用了指定的回退方法，getDefaultValue，返回了0.

[图片上传失败...(image-422c2f-1548757736276)]

这里需要注意的有几点：

1. 通过网关zuul调用时，zuul的超时时间要大于等于hystrix的超时时间配置，否则在zuul层转发时就已经触发了zuul的超时，返回 GATEWAY TIME OUT

2. 无重试机制时，通过feign加ribbon进行服务之间调用时，hystrix配置超时时间要小于ribbon超时时间，否则在ribbon调用其他服务时就已经超时了，hystrix无法进行熔断及降级

3. 如果有重试时，如有组件跟Hystrix配合使用，一般来讲，建议Hystrix的超时 > 其他组件的超时，否则将可能导致重试特性失效。例如，如果ribbon超时时间为1秒，重试3次，hystrix超时时间应略大于3秒。

4. 定义一个fallback方法需要注意以下几点：

   • fallback方法必须和指定fallback方法的主方法在一个类中。

   • fallback方法的参数必须要和主方法的参数一致，否则不生效。

   • 使用fallback方法需要根据依赖服务设置合理的超时时间，即execution.isolation.thread.timeoutInMilliseconds的设置，可以在@HystrixCommand注解上通过HystrixProperty指定。

5. 如果要在fallback方法中获取异常信息，只需要在fallback方法中，加上一个参数Throwable throwable就可以了。

GitHub上Netflix给的例子是通过继承HystrixCommand或者HystrixObservableCommand，然后实现execute()或者queue()方法来运行，但是这种方式代码侵入性较大。使用注解@HystrixCommand方式的侵入性小一点。

刚才的hystrix的服务降级示例是同步模式的，也是通常情况下我们所使用的模式。

• 同步command，同步fallback

@HystrixCommand(fallbackMethod = "getDefaultValue")
@PostMapping("/toCart/{productId}")
public ResponseEntity addCart(@PathVariable("productId") Long productId) throws InterruptedException {
    Long aLong = cartFeignClient.addCart(productId);
    System.out.println(aLong);
    return ResponseEntity.ok(productId);
}

private ResponseEntity getDefaultValue(Long productId) {
    return ResponseEntity.ok(0);
}

• 异步command，同步fallback

@HystrixCommand(fallbackMethod = "getDefaultAsyncAddCart")
public Future<ResponseEntity<Long>> asyncAddCart(@PathVariable("productId") Long productId) throws InterruptedException {
    log.info("异步command：run。。。");
    Thread.sleep(5000);//触发降级逻辑
    return new AsyncResult<ResponseEntity<Long>>() {
        @Override
        public ResponseEntity<Long> invoke() {
            return ResponseEntity.ok(cartFeignClient.addCart(productId));
        }
    };
}

private ResponseEntity<Long> getDefaultAsyncAddCart(Long productId, Throwable throwable) {
    log.info("异步command，同步fallback：run。。。");
    return ResponseEntity.ok(0L);
}

• 异步command，异步fallback

@HystrixCommand(fallbackMethod = "getDefaultAsyncAddCart2")
public Future<ResponseEntity<Long>> asyncAddCart2(@PathVariable("productId") Long productId) throws InterruptedException {
    log.info("异步command：run。。。");
    Thread.sleep(5000);
    return new AsyncResult<ResponseEntity<Long>>() {
        @Override
        public ResponseEntity<Long> invoke() {
            return ResponseEntity.ok(cartFeignClient.addCart(productId));
        }
    };
}

@HystrixCommand //注意，异步fallback 这里必须加@HystixCommand注解，否则运行时报错
private Future<ResponseEntity<Long>> getDefaultAsyncAddCart2(Long productId, Throwable throwable) {
    log.info("异步command，同步fallback：run。。。");
    log.warn("", throwable);
    return new AsyncResult<ResponseEntity<Long>>() {
        @Override
        public ResponseEntity<Long> invoke() {
            return ResponseEntity.ok(0L);
        }
    };
}

注意，hystrix不支持同步command，异步fallback。

hystrix配置

我们先看下@HystrixCommand注解中有哪些配置。

其他配置：详情见https://github.com/Netflix/Hystrix/wiki/Configuration

@HystixCommand属性配置官方示例：https://github.com/Netflix/Hystrix/tree/master/hystrix-contrib/hystrix-javanica

命令属性配置：

执行时：

回退时：

断路器配置：

Metrics：

Request Context：

Collapser Properties：

Thread Properties：

配置hystrix线程池的属性

image

服务容错保护（Hystrix依赖隔离）

hystrix为每一个命令创建一个独立的线程池，这样就算某个hystrix命令由于依赖其他服务导致出现延迟过高的情况，也只是对该依赖服务的调用产生影响，而不会拖累其他服务。

通过对依赖服务的线程池隔离实现，可以带来如下优势：

• 应用自身得到完全的保护，不会受不可控的依赖服务影响。即便给依赖服务分配的线程池被填满，也不会影响应用自身的额其余部分。
• 可以有效的降低接入新服务的风险。如果新服务接入后运行不稳定或存在问题，完全不会影响到应用其他的请求。
• 当依赖的服务从失效恢复正常后，它的线程池会被清理并且能够马上恢复健康的服务，相比之下容器级别的清理恢复速度要慢得多。
• 当依赖的服务出现配置错误的时候，线程池会快速的反应出此问题（通过失败次数、延迟、超时、拒绝等指标的增加情况）。同时，我们可以在不影响应用功能的情况下通过实时的动态属性刷新（后续会通过Spring Cloud Config与Spring Cloud Bus的联合使用来介绍）来处理它。
• 当依赖的服务因实现机制调整等原因造成其性能出现很大变化的时候，此时线程池的监控指标信息会反映出这样的变化。同时，我们也可以通过实时动态刷新自身应用对依赖服务的阈值进行调整以适应依赖方的改变。
• 除了上面通过线程池隔离服务发挥的优点之外，每个专有线程池都提供了内置的并发实现，可以利用它为同步的依赖服务构建异步的访问。

总之，通过对依赖服务实现线程池隔离，让我们的应用更加健壮，不会因为个别依赖服务出现问题而引起非相关服务的异常。同时，也使得我们的应用变得更加灵活，可以在不停止服务的情况下，配合动态配置刷新实现性能配置上的调整。

原理及设计

[图片上传失败...(image-f885c3-1548757736276)]

通过翻看hystrix-javanica的ReadMe以及查看注解@HystrixCommand的引用可以发现，HystrixCommandAspect类是一个很关键的类。

接下来我们从HystrixCommandAspect来入手。

@Aspect
public class HystrixCommandAspect {

    private static final Map<HystrixPointcutType, MetaHolderFactory> META_HOLDER_FACTORY_MAP;

    static {
        META_HOLDER_FACTORY_MAP = ImmutableMap.<HystrixPointcutType, MetaHolderFactory>builder()
                .put(HystrixPointcutType.COMMAND, new CommandMetaHolderFactory())
                .put(HystrixPointcutType.COLLAPSER, new CollapserMetaHolderFactory())
                .build();
    }

//指定切点为@HystrixCommand注解 @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)")

    public void hystrixCommandAnnotationPointcut() {
    }

    @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
    public void hystrixCollapserAnnotationPointcut() {
    }
    //切面
    @Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
    public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
        //先获取到执行的方法数据
        Method method = getMethodFromTarget(joinPoint);
        Validate.notNull(method, "failed to get method from joinPoint: %s", joinPoint);
        //...
        //根据注解类型获取不同methodHolder的构造工厂，methodHolder用来保存方法的一些数据，如注解以及注解的属性值等
        MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
        //根据method创建一个methodHolder
        MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
        //构造hystrixCommand对象
        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() != null ? e.getCause() : e;
        } catch (HystrixRuntimeException e) {
            throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
        }
        return result;
    }
}    

非流式命令执行代码：

CommandExecutor.execute()。

public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException {
    //......
    switch (executionType) {
        case SYNCHRONOUS: {
            //同步模式执行command
            return castToExecutable(invokable, executionType).execute();
        }
        case ASYNCHRONOUS: {
            //异步模式执行command
            HystrixExecutable executable = castToExecutable(invokable, executionType);
            if (metaHolder.hasFallbackMethodCommand()
                    && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) {
                return new FutureDecorator(executable.queue());
            }
            return executable.queue();
        }
        case OBSERVABLE: {
            //流式模式执行command
            HystrixObservable observable = castToObservable(invokable);
            return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
        }
        default:
            throw new RuntimeException("unsupported execution type: " + executionType);
    }
}

private static HystrixExecutable castToExecutable(HystrixInvokable invokable, ExecutionType executionType) {
        //转换
        if (invokable instanceof HystrixExecutable) {
            return (HystrixExecutable) invokable;
        }
        throw new RuntimeException("Command should implement " + HystrixExecutable.class.getCanonicalName() + " interface to execute in: " + executionType + " mode");
    }

    private static HystrixObservable castToObservable(HystrixInvokable invokable) {
        //转换
        if (invokable instanceof HystrixObservable) {
            return (HystrixObservable) invokable;
        }
        throw new RuntimeException("Command should implement " + HystrixObservable.class.getCanonicalName() + " interface to execute in observable mode");
    }

同步执行的execute()方法有两个实现类，分别是HystrixCommand以及HystrixCollapser。

通常是HystrixCommand，只有在指定合并多个请求时才会是HystrixCollapser。

我们先看下HystrixCommand的execute()方法。

public R execute() {
    try {
        //直接调用queue()方法获得结果Future后再调用get()方法。
        return queue().get();
    } catch (Exception e) {
        //重新抛出异常
        throw Exceptions.sneakyThrow(decomposeException(e));
    }
}

public Future<R> queue() {
        /*
         * The Future returned by Observable.toBlocking().toFuture() does not implement the
         * interruption of the execution thread when the "mayInterrupt" flag of Future.cancel(boolean) is set to true;
         * thus, to comply with the contract of Future, we must wrap around it.
         */
        final Future<R> delegate = toObservable().toBlocking().toFuture();
        
        final Future<R> f = new Future<R>() {

            //...
            
        };

        /* special handling of error states that throw immediately */
        if (f.isDone()) {
            try {
                f.get();
                return f;
            } catch (Exception e) {
                //...处理异常
            }
        }

        return f;
    }

Hystrix还有哪些待改进的地方？

待完善。

有没有更好的解决方式

待完善。