hystrix 学习

简介

hystrix 作用有多个，主要用来解决雪崩效应，即rpc 链条 A-B-C,C 节点服务崩溃、变慢直接拖累上游的B A 服务变慢、崩溃。

从使用上感受 hystrix

使用命令模式将所有对外部服务（或依赖关系）的调用包装在HystrixCommand或HystrixObservableCommand对象中，并将该对象放在单独的线程中执行；

public class QueryOrderIdCommand extends HystrixCommand<Integer> {
    private final static Logger logger = LoggerFactory.getLogger(QueryOrderIdCommand.class);
    private OrderServiceProvider orderServiceProvider;

    public QueryOrderIdCommand(OrderServiceProvider orderServiceProvider) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("orderService"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("queryByOrderId"))
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        .withCircuitBreakerRequestVolumeThreshold(10)//至少有10个请求，熔断器才进行错误率的计算
                        .withCircuitBreakerSleepWindowInMilliseconds(5000)//熔断器中断请求5秒后会进入半打开状态,放部分流量过去重试
                        .withCircuitBreakerErrorThresholdPercentage(50)//错误率达到50开启熔断保护
                        .withExecutionTimeoutEnabled(true))
                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties
                        .Setter().withCoreSize(10)));
        this.orderServiceProvider = orderServiceProvider;
    }

    @Override
    protected Integer run() {
        return orderServiceProvider.queryByOrderId();
    }

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

调用逻辑

public void testQueryByOrderIdCommand() {
    Integer r = new QueryOrderIdCommand(orderServiceProvider).execute();
    logger.info("result:{}", r);
}

从中可以看到以下几点：

实际的服务提供方 OrderServiceProvider 被封装
除服务提供方外，构造HystrixCommand 时，对其进行了熔断相关的配置
run 方法容纳业务逻辑，getFallback 方法容纳降级后的逻辑

命令模式

A request is wrapped under an object as command and passed to invoker object. Invoker object looks for the appropriate object which can handle this command and passes the command to the corresponding object which executes the command.

命令模式：在软件系统中，“行为请求者”与“行为实现者”通常呈现一种“紧耦合”。但在某些场合，比如要对行为进行“记录、撤销/重做、事务”等处理，这种无法抵御变化的紧耦合是不合适的（即无法通过新增来应对修改）。在这种情况下，如何将“行为请求者”与“行为实现者”解耦？将一组行为抽象为对象（行为 ==> 对象），实现二者之间的松耦合。这就是命令模式（Command Pattern）

总结一下就是，原本只要调用方和被调用方，现在是调用方 ==> invoker ==> 被调用方。按照程序=控制 + 逻辑的理念，调用方和被调用方仍然专注于自己的业务（负责逻辑），控制的实现（何时执行、同步还是异步、日志记录等）交给invoker。否则，这些代码会出现在调用方的代码中，与调用方业务混在一起。

说句题外话，程序=控制 + 逻辑，对于排序算法来说，决定谁大谁小是逻辑，采用何种算法、是否分治是控制。对于业务代码来说，订单服务调用库存服务扣掉库存是逻辑，同步/异步、熔断等是控制。

因为多了一个invoker，所以请求要被封装成（命令）对象以便传递。命令对象除了封装请求，还可以封装一些参数，以便调用方指定执行请求的方式（比如同步还是异步等）。对于前文的例子就是调用方只管command.execute()( new QueryOrderIdCommand(orderServiceProvider).execute()) 线程池个数、熔断等通过显式或默认参数控制，而不是将这些逻辑写在调用代码里。

源码分析

不要对hystrix 有畏难情绪，其几个子项目有用的就是hystrix-core 和 hystrix-serialization ，熟悉原理只看一个hystrix-core即可

1.0.0 时还未使用 rxjava，代码整体比较清晰

整体

官方wiki How it Works

HystrixCommand - Used to wrap code that will execute potentially risky functionality (typically meaning a service call over the network) with fault and latency tolerance, statistics and performance metrics capture, circuit breaker(熔断器) and bulkhead （隔离）functionality.This command is essentially a blocking command but provides an Observable facade if used with observe()

一个HystrixCommand 有一个成员

HystrixCommandGroupKey
HystrixCommandKey
HystrixThreadPoolKey
HystrixCommandProperties
HystrixThreadPoolProperties

几个key的作用

command key，每个command 对应一个key，默认是当前类的名字
command group key，监控、报警用的
thread pool key，维护了一个<thread pool key,thread pool>
cache key , 相当于建立参数到 R 的映射

隔离如何实现

假设存在一个

class HelloService{
	run(){
		service1
		service2
	}
}

请求线程与执行线程的隔离

其实跟 netty 通信中，将线程池分为 acceptor 线程和 worker 线程是一样一样的。
```
 Future future1 = threadPool.submit(new Runnable(){
     service1
 })
 Future future2 = threadPool.submit(new Runnable(){
     service2
 })
 future1.get()
 future2.get(1000)		
```
此时的好处在于，future get 可以设置timeout，calling thread 只是提交代码和处理结果。
一个业务一个thread pool
```
 class HelloService{
     Map<String,ThreadPool> cache
     run(){
         cache.get(key).execute()
         cache.get(key).execute()
     }
 }
```
可以看到，我们将获取线程池、future.get(1000)、超时后的处理逻辑这些单独拎出来，hystrix 抽象成了HystrixCommand。

从本质上讲，hystrix 通过将业务逻辑从调用线程中剥离，达到保护调用线程的效果。这里的调用线程，比如tomcat 线程池的中线程，通常非常宝贵，用于处理耗时任务时会严重影响项目的吞吐量。保护的手段就是强行限定一个方法的执行时间（待斟酌）。

熔断如何实现

HystrixCircuitBreaker 本身更偏向业务一点，该接口定义了三个方法

boolean allowRequest();
boolean isOpen();
void markSuccess();

就是预设几个参数，根据当前的qps、失败率等做出决策就可以。笔者的兴趣在hystrix 如何高效的统计qps 等

浅析HystrixRollingNumber(用于qps计数的数据结构)

qps表示每秒的请求数目，能想到的最简单的方法就是统计一定时间内的请求总数然后除以总统计时间。这样的方法虽然简单但是对有一定的问题，比如说统计出的qps跳跃性会比较大，不够平滑等。所以采用滑动窗口的方式，比如窗口大小是10s，interval 是1s，那么某日第n秒的qps 是”n-10到n秒的访问量之和/10”。毕竟熔断器我们想让它发挥作用，但也不想让它太过敏感，一惊一乍的。假如某一秒的qps 突然增大，常规方案会导致立即熔断，而滑动窗口则会在qps 连续10s（最坏情况）处于高位时才熔断，这便是HystrixRollingNumber 类rolling 的含义所在，简单说就是控制熔断器的敏感度。
假设存在一个QPS类，线程使用qps.add() 增加计数，可以想见线程竞争强度会非常大。

[从LongAdder看更高效的无锁实现

酷壳- CoolShell](https://coolshell.cn/articles/11454.html) 一个比AtomicLong 原子类更高效的计数实现。里面用了一系列技术，比如填充缓存行（削减volatile 带来的同步耗费）、分段技术以使空间换时间。一个变量用作计数线程竞争大，则One or more variables that together maintain an initially zero sum。 add 时，对线程做hashcode 找到对应的分段变量，对该分段变量加1。若发现竞争还是大，则对分段扩容。

rxjava 在其中的作用

从各种材料看，rxjava 在服务端用处并不多，为什么在hystrix 被广泛使用呢？

根据hystrix changelog ，hystrix 于 1.3.0 引入RxJava ，This version integrations Hystrix with RxJava to enable non-blocking reactive execution and functional composition.

所以，关键就是两个词non-blocking reactive execution 和 functional composition。

从hystrix 哪些部分被rxjava 重写了，或许可以看到一些猫腻。

public abstract class HystrixCommand<R> implements HystrixExecutable<R> {
	public Future<R> queue() {
		final ObservableCommand<R> o = toObservable(Schedulers.immediate(), false);
        final Future<R> f = o.toBlockingObservable().toFuture();
        ...
	}
}

private ObservableCommand<R> toObservable(Scheduler observeOn, boolean performAsyncTimeout) {
	...
	/* try from cache first */
	Observable<R> o = Observable.create(new Func1<Observer<R>, Subscription>() {
		 public Subscription call(Observer<R> observer) {
		 	if(熔断了){
		 		try {
                    observer.onNext(getFallbackOrThrowException(HystrixEventType.SHORT_CIRCUITED, FailureType.SHORTCIRCUIT, "short-circuited"));
                    observer.onCompleted();
                } catch (Exception e) {
                    observer.onError(e);
                }
		 	}else{
		 		if (使用线程隔离) {
                    return subscribeWithThreadIsolation(observer);
                } else {
                    return subscribeWithSemaphoreIsolation(observer);
                }
		 	}
		 }
	});
	// error handling
	o = o.onErrorResumeNext(...)
	// we want to hand off work to a different scheduler so we don't tie up the Hystrix thread
	o = o.observeOn(observeOn);
	o = o.finallyDo(...)
	// put in cache
	...
}

private Subscription subscribeWithThreadIsolation(final Observer<R> observer) {
 	...
 	final Future<R> f = threadPool.getExecutor().submit(new Callable(){
 		public R call() throws Exception {
 			try{	
 				...
 				R r = executeCommand();
 				observer.onNext(r);
 				observer.onCompleted();
 				return r;
 			}catch (Exception e) {
 				observer.onError(e);
 				...
 			}
 		}

 	});
 	return new Subscription() {
        @Override
        public void unsubscribe() {
            f.cancel(properties.executionIsolationThreadInterruptOnTimeout().get());
        }
    };
 }

2018.8.5补充：从现在看，使用rxjava的重要原因不是用rxjava 去重构原有HystrixCommand 的实现（确实可以简化一些逻辑），而是试图使用Observable<String> s = new CommandHelloWorld("World").observe(); 去替换 Future<String> f = new CommandHelloWorld("World").queue();

其实内在驱动逻辑是：

没有hystrix java，则一个web 服务的处理从request 到 db 再回来，都是tomcat 线程池中的线程打理的。

有了hystrix，即有了线程隔离、熔断等机制， tomcat 线程 ==> 网络通信 ==> rpc server 变成了 tomcat 线程 ==> command thread pool 线程 ==> 网络通信 ==> rpc server，驱动线程和执行线程不一致隐含的一个约束就是，command.execute 实质必须是异步的。在具体的业务中，通常不是一个 rpc 调用就结束了，如果是两个相互依赖的rpc，

Future<String> f = new CommandHelloWorld("World").queue();
R r = f.get();
R1 r1 = map(r);
Future<String> f1 = new CommandHelloWorld(r1).queue();
f1.get();

使用rxjava，则是为了更加简化这类逻辑的实现。

技术

生活

架构

产品

标签

Container 23

Concurrency 14

Life 41

Tool 8

Algorithm 8

JVM 10

Go 21

Kubernetes 65

Other 5

Network 15

Python 7

Java 20

Spring 17

Netty 10

Storage 24

Distribute 9

MQ 8

WEB 5

Linux 11

Scala 1

Code 9

MachineLearning 74

Practice 16

RPC 6

Compute 11

Architecture 20

DDD 6

Reactive 5

Basic 13

Product 3

Monitor 7

CPP 2

Mesh 12

简介