Spring Cloud Hystrix 源码系列：隔离策略和请求缓存

比眉伴天荒 2022-03-16 12:21 112阅读 0赞

隔离是一种常见的风险控制(保护)手段，Hystrix 也使用了隔离策略，称之为 **bulkhead pattern**，翻译为：舱壁隔离模式（舱壁将船体内部空间划分成多个舱室，将舱与舱之间严密分开，在航行中，即使有舱破损进水，水也不会流到其他舱）。本文基于[hystrix-core][] 1.5.18（近年来几乎很少更新，建议升级）

**目录**

1. Hystrix 隔离

1.1 Thread Pool

1.2 Semaphore

1.3 源码

2. 请求缓存

2.1 ReplaySubject

2.2 缓存的生命周期

2.3 源码

--------------------

## 1. Hystrix 隔离 ##

Hystrix 的隔离指使用 bulkhead 模式来隔离各个依赖服务之间的调用，同时限制对各个依赖服务的并发访问，使得各依赖服务之间互不影响。Hystrix 提供了两种隔离方式。

### 1.1 Thread Pool ###

将各依赖服务的访问交由**独立的线程池**来处理，会为每个依赖服务创建一个线程池。虽然可以起到很好的隔离作用，但也增加了计算开销，每个命令的执行都涉及到queueing、scheduling、**context switching**。

### 1.2 Semaphore ###

通过为各依赖服务设置信号量(或计数器)来限制并发调用，相当于对各依赖服务做限流。信号量模式下任务由当前线程直接处理，**不涉及到线程切换**，自然也就没有超时控制。

### 1.3 源码 ###

Spring Cloud Hystrix 源码系列：[HystrixCommandAspect入口解析][HystrixCommandAspect]，介绍过 Hystrix 的原理，重点关注executeCommandWithSpecifiedIsolation()，隔离策略在这个方法中实现。

在Hystrix的工作机制中，在**判断熔断之后**，**准备执行任务前，会先判断信号量拒绝和线程池拒绝的情况**，如下：

private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
        if (circuitBreaker.attemptExecution()) {
            // 获取信号量
            final TryableSemaphore executionSemaphore = getExecutionSemaphore();
            ...
            // 尝试获取执行信号，能否执行当前命令
            if (executionSemaphore.tryAcquire()) {
                   ...
            } else {
                return handleSemaphoreRejectionViaFallback();
            }
        } else {
            return handleShortCircuitViaFallback();
        }
    }

**线程池隔离**

使用的是rxjava的subscribeOn，重点关注AbstractCommand.threadPool的构建及获取。

> 之前的[HystrixThreadPoolKey][HystrixCommandAspect]也讲过：
> 
> 具有相同的threadPoolKey的使用同一个线程池，如果注解未指定threadPoolKey，则使用groupKey作为HystrixThreadPoolKey。通过HystrixThreadPoolKey.Factory.asKey(groupKey.name())获取。

private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {
        if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
            return Observable.defer(new Func0<Observable<R>>() {
                @Override
                public Observable<R> call() {
                    ...
                    // 获取包裹实际Task的Observable
                    return getUserExecutionObservable(_cmd);
                    ...
                    }
                }
            }).doOnTerminate(...)
              .doOnUnsubscribe(...)
              // 线程池模式执行  
              .subscribeOn(threadPool.getScheduler(new Func0<Boolean>() {...}));
        } else {// 信号量模式在当前线程执行
            return Observable.defer(new Func0<Observable<R>>() {
                @Override
                public Observable<R> call() {
                    ...
                      return getUserExecutionObservable(_cmd); 
                    ...
                }
            });
        }
    }
    // 每个threadPoolKey会维护一个HystrixThreadPool
    public interface HystrixThreadPoolKey extends HystrixKey {
        class Factory {
            private Factory() {
            }
            // used to intern instances so we don't keep re-creating them millions of times for the same key
            private static final InternMap<String, HystrixThreadPoolKey> intern
                    = new InternMap<String, HystrixThreadPoolKey>(
                    new InternMap.ValueConstructor<String, HystrixThreadPoolKey>() {
                        @Override
                        public HystrixThreadPoolKey create(String key) {
                            return new HystrixThreadPoolKeyDefault(key);
                        }
                    });
            public static HystrixThreadPoolKey asKey(String name) {
               return intern.interned(name);
            }
    
            private static class HystrixThreadPoolKeyDefault extends HystrixKeyDefault implements HystrixThreadPoolKey {
                public HystrixThreadPoolKeyDefault(String name) {
                    super(name);
                }
            }
    
            static int getThreadPoolCount() {
                return intern.size();
            }
        }
    }

**信号量隔离**

隔离策略为SEMAPHORE时会创建TryableSemaphoreActual，否则返回一个什么也不做的TryableSemaphoreNoOp(tryAcquire()将永远返回true)。

//AbstractCommand.toObservable方法中定义applyHystrixSemantics变量
    final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
            @Override
            public Observable<R> call() {
                if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
                    return Observable.never();
                }
                return applyHystrixSemantics(_cmd);
            }
    };
    //在判断熔断之后，准备执行任务前，会先判断信号量拒绝和线程池拒绝的情况
    private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
        // mark that we're starting execution on the ExecutionHook
        // if this hook throws an exception, then a fast-fail occurs with no fallback.  No state is left inconsistent
        executionHook.onStart(_cmd);
    
        /* determine if we're allowed to execute */
        if (circuitBreaker.allowRequest()) {
            final TryableSemaphore executionSemaphore = getExecutionSemaphore();
            final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
            final Action0 singleSemaphoreRelease = new Action0() {
                @Override
                public void call() {
                    if (semaphoreHasBeenReleased.compareAndSet(false, true)) {
                        executionSemaphore.release();
                    }
                }
            };
    
            final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {
                @Override
                public void call(Throwable t) {
                    eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, commandKey);
                }
            };
            //执行控制
            if (executionSemaphore.tryAcquire()) {
                try {
                    /* used to track userThreadExecutionTime */
                    executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
                    return executeCommandAndObserve(_cmd)
                            .doOnError(markExceptionThrown)
                            .doOnTerminate(singleSemaphoreRelease)
                            .doOnUnsubscribe(singleSemaphoreRelease);
                } catch (RuntimeException e) {
                    return Observable.error(e);
                }
            } else {
                return handleSemaphoreRejectionViaFallback();
            }
        } else {
            return handleShortCircuitViaFallback();
        }
    }
    //获取信号量
    protected TryableSemaphore getExecutionSemaphore() {
        if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.SEMAPHORE) {
            TryableSemaphore _s = executionSemaphorePerCircuit.get(commandKey.name());
            return _s;
        } else {
            // return NoOp implementation since we're not using SEMAPHORE isolation
            return TryableSemaphoreNoOp.DEFAULT;//无任何实现直接通过
        }
    }
    //信号量，每个HystrixCommandKey默认信号量数量，默认10
    static class TryableSemaphoreActual implements TryableSemaphore {
        protected final HystrixProperty<Integer> numberOfPermits;
        private final AtomicInteger count = new AtomicInteger(0);
        public TryableSemaphoreActual(HystrixProperty<Integer> numberOfPermits) {
            this.numberOfPermits = numberOfPermits;
        }
        @Override
        public boolean tryAcquire() {
            int currentCount = count.incrementAndGet();
            // 如果信号量超过设定的信号量，则启动信号量拒绝
            if (currentCount > numberOfPermits.get()) {
                count.decrementAndGet();
                return false;
            } else {
                return true;
            }
        }
        ...
    }

## 2. 请求缓存 ##

微服务中，服务之间的依赖非常多，为了提升性能，如果每个方法都自行处理缓存的话，应用中可以想象有多少累赘的缓存代码。在请求生命周期内，无论是当前线程，还是其他线程，只要请求相同的数据，就自动做缓存，不侵入业务代码。

### 2.1 **ReplaySubject**  ###

Hystrix 的**请求缓存用的就是RxJava 中的 ReplaySubject 这个特性**。replay 译为重放，Subject 是个合体工具，既可以做数据发射器(被观察者、Observable)，也可以做数据消费者(观察者、Observer)。**如果请求相同数据，就把原先的结果发你一份**。

@Test
    public void replaySubject() {
        ReplaySubject<Integer> replaySubject = ReplaySubject.create();
        replaySubject.subscribe(v -> System.out.println("订阅者1:" + v));
        replaySubject.onNext(1);
        replaySubject.onNext(2);
        replaySubject.subscribe(v -> System.out.println("订阅者2:" + v));
        replaySubject.onNext(3);
        replaySubject.subscribe(v -> System.out.println("订阅者3:" + v));
    }
    /** 无论是 replaySubject 多久前发射的数据，新的订阅者都可以收到所有数据
    
    订阅者1:1
    订阅者1:2
    
    订阅者2:1
    订阅者2:2
    
    订阅者1:3
    订阅者2:3
    
    订阅者3:1
    订阅者3:2
    订阅者3:3
    
    **/

### **2.2 缓存的生命周期** ###

缓存是request scope，在同一个请求范围内（Thread），可以使用相同cacheKey已缓存的结果。

public class HystrixCommandCacheTest extends HystrixCommand<String> {
        private final String value;
        public HystrixCommandCacheTest(String value) {
            super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
            this.value = value;
        }
        // 重要：要重写getCacheKey()方法
        @Override
        protected String getCacheKey() {
            return value;
        }
        @Override
        protected String run() throws Exception {
            return "hello," + value;
        }
        public static void main(String[] args) {
            // 第一个请求环境
            HystrixRequestContext context1 = HystrixRequestContext.initializeContext();
            HystrixCommandCacheTest cmd1 = new HystrixCommandCacheTest("kitty");
            System.out.println("cmd1结果：" + cmd1.execute() + ";使用缓存：" + cmd1.isResponseFromCache());
            // 模拟10个相同请求参数的命令执行
            for (int i = 0; i < 10; i++) {
                HystrixCommandCacheTest temp2 = new HystrixCommandCacheTest("kitty");
                System.out.println("第" + i + " 次执行:" + temp2.execute() + ";使用缓存：" + temp2.isResponseFromCache());
            }
            try {
                new Thread(() -> {
                    HystrixCommandCacheTest tempCmd = new HystrixCommandCacheTest("kitty");
                    System.out.println("线程2执行:" + tempCmd.execute() + ";使用缓存：" + tempCmd.isResponseFromCache());
                }).start();
            } catch (Exception ex) {//报错：Request caching is not available. Maybe you need to initialize the HystrixRequestContext?
                ex.printStackTrace();
            }
            context1.shutdown();
            // 第二个请求环境
            HystrixRequestContext context2 = HystrixRequestContext.initializeContext();
            HystrixCommandCacheTest cmd2 = new HystrixCommandCacheTest("kitty");
            System.out.println("cmd2结果:" + cmd2.execute() + ";使用缓存：" + cmd2.isResponseFromCache());
        }
    }
    /**
    cmd1结果：hello,kitty;使用缓存：false
    第0 次执行:hello,kitty;使用缓存：true
    第1 次执行:hello,kitty;使用缓存：true
    第2 次执行:hello,kitty;使用缓存：true
    第3 次执行:hello,kitty;使用缓存：true
    第4 次执行:hello,kitty;使用缓存：true
    第5 次执行:hello,kitty;使用缓存：true
    第6 次执行:hello,kitty;使用缓存：true
    第7 次执行:hello,kitty;使用缓存：true
    第8 次执行:hello,kitty;使用缓存：true
    第9 次执行:hello,kitty;使用缓存：true
    cmd2结果:hello,kitty;使用缓存：false
    **/

### 2.3 源码 ###

AbstractCommand.toObservable()中关于请求缓存的源码。请求缓存有2个条件：

1.  启用了请求缓存
2.  有相同cacheKey

整个逻辑还是非常简单的，在启用缓存的前提后，有缓存则读缓存，没缓存则缓存结果供下次使用。

HystrixRequestCache实例的存储是由自身的静态变量搞定，未提供public的构造器，通过 **getInstance()** 的静态方法来获取与cacheKey对应的实例。

requestVariableForCache缓存这请求的结果，它的value存储HystrixCachedObservable（数据结构是ConcurrentHashMap，cacheKey作为key，HystrixCachedObservable为value）

public Observable<R> toObservable() {
        final AbstractCommand<R> _cmd = this;
        ...
        return Observable.defer(new Func0<Observable<R>>() {
            @Override
            public Observable<R> call() {
                ...
                final boolean requestCacheEnabled = isRequestCachingEnabled();
                final String cacheKey = getCacheKey();
                // 启用了requestCache, 则尝试从缓存中获取
                if (requestCacheEnabled) {
                    HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);
                    if (fromCache != null) {
                        isResponseFromCache = true;
                        // 从缓存中获取数据
                        return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
                    }
                }
                Observable<R> hystrixObservable =
                        Observable.defer(applyHystrixSemantics)
                                .map(wrapWithAllOnNextHooks);
                Observable<R> afterCache;
                // 启用缓存而且有cacheKey
                if (requestCacheEnabled && cacheKey != null) {
                    // 使用HystrixCachedObservable来包装Obervable,并且添加到requestCache中
                    HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
                    HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.putIfAbsent(cacheKey, toCache);
                    ...
                    afterCache = toCache.toObservable();
                }
                ...
            }
        });
    // 缓存的工具
    /**
    Cache that is scoped to the current request as managed by HystrixRequestVariableDefault.
    This is used for short-lived caching of HystrixCommand instances to allow de-duping of command executions within a request.
    缓存仅在请求范围内使用，主要用途是减少HystrixCommand实例的执行次数(缓存结果后执行次数自然少了)
    **/
    public class HystrixRequestCache {
        private final static ConcurrentHashMap<RequestCacheKey, HystrixRequestCache> caches = new ConcurrentHashMap<RequestCacheKey, HystrixRequestCache>();
        /**
         * A ConcurrentHashMap per 'prefix' and per request scope that is used to to dedupe requests in the same request.
         * <p>
         * Key => CommandPrefix + CacheKey : Future<?> from queue()
         */
        private static final HystrixRequestVariableHolder<ConcurrentHashMap<ValueCacheKey, HystrixCachedObservable<?>>> requestVariableForCache = new HystrixRequestVariableHolder<ConcurrentHashMap<ValueCacheKey, HystrixCachedObservable<?>>>(new HystrixRequestVariableLifecycle<ConcurrentHashMap<ValueCacheKey, HystrixCachedObservable<?>>>() {
    
            @Override
            public ConcurrentHashMap<ValueCacheKey, HystrixCachedObservable<?>> initialValue() {
                return new ConcurrentHashMap<ValueCacheKey, HystrixCachedObservable<?>>();
            }
    
            @Override
            public void shutdown(ConcurrentHashMap<ValueCacheKey, HystrixCachedObservable<?>> value) {
                // nothing to shutdown
            }
    
        });
        public static HystrixRequestCache getInstance(HystrixCommandKey key, HystrixConcurrencyStrategy concurrencyStrategy) {
            return getInstance(new RequestCacheKey(key, concurrencyStrategy), concurrencyStrategy);
        }
    }

再看看缓存的结果HystrixCachedObservable，这个就用到了上面提过的ReplaySubject。将一个普通的Observable包装成了HystrixCachedObservable。因此，command执行一次拿到结果来自于ReplaySubject。后续无论有多少次订阅(即执行command)，都把已有的结果推送一次，从而达到缓存请求结果的效果。

public class HystrixCachedObservable<R> {
        protected final Subscription originalSubscription;
        protected final Observable<R> cachedObservable;
        private volatile int outstandingSubscriptions = 0;
        protected HystrixCachedObservable(final Observable<R> originalObservable) {
            ReplaySubject<R> replaySubject = ReplaySubject.create();
            // 订阅普通的Observable, 拿到其中的数据
            this.originalSubscription = originalObservable
                    .subscribe(replaySubject);
            this.cachedObservable = replaySubject...
        }
        ...
        // 将cachedObservable作为数据源提供出去, 完成普通Observable向ReplaySubject的转换
        public Observable<R> toObservable() {
            return cachedObservable;
        }
    }

**如何使用缓存的结果**

由于toObservable()拿到的是一个ReplaySubject，下次命令再次执行时，订阅ReplaySubject后，可以直接拿到之前已有的结果。

// 以HystrixCommand的 queue() 方法为例
    public Future<R> queue() {
        // 调用 toObservable 拿到数据源
        final Future<R> delegate = toObservable().toBlocking().toFuture();
        ...
    }
    //在toFuture()中会订阅这个数据源：
    public static <T> Future<T> toFuture(Observable<? extends T> that) {
        final CountDownLatch finished = new CountDownLatch(1);
        final AtomicReference<T> value = new AtomicReference<T>();
        final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
        // 首先,通过single()确保从Observable中拿到单个结果. 然后订阅数据源
        @SuppressWarnings("unchecked")
        final Subscription s = ((Observable<T>)that).single().subscribe(new Subscriber<T>() {
            @Override
            public void onNext(T v) {
                // 拿到执行的结果后放到AtomicReference中
                value.set(v);
            }
        });
        return new Future<T>() {
            private volatile boolean cancelled;
            // 返回执行结果
            @Override
            public T get() throws InterruptedException, ExecutionException {
                finished.await();
                return getValue();
            }
        };
    }

利用缓存可以极大的提升性能，但是Hystrix的缓存比较鸡肋。![89844066][]

[hystrix-core]: https://mvnrepository.com/artifact/com.netflix.hystrix/hystrix-core
[HystrixCommandAspect]: https://blog.csdn.net/alex_xfboy/article/details/88720949
[89844066]: /images/20220316/94e8deff85c54f1482fac8e934b89d00.png