Resilience4j CircuitBreaker 源码分析

//抽象的主体角色
interface AbstractSubject{
  void registerObserver();
  void notifyObserver();
}
//具体的主体角色
class ConcreteSubject implements AbstractSubject{
  List<AbstractObserver> observers = new ArrayList<>();
  
  public void registerObserver(Observer o) {
        this.observers.add(o);
    }

  public void notifyObservers(String tweet) {
        observers.forEach(o -> o.notify(tweet));
    }
}
//抽象的观察者角色
interface AbstractObserver{
  void notify(String tweet);
}
//具体的观察的角色
class ConcreteNYTimes implements Observer {

    @Override
    public void notify(String tweet) {
        if (tweet != null && tweet.contains("money")) {
            System.out.println("Breaking news in NY! " + tweet);
        }
    }
}
//具体的观察的角色
class ConcreteGuardian implements Observer {

    @Override
    public void notify(String tweet) {
        if (tweet != null && tweet.contains("queen")) {
            System.out.println("Yet another news in London... " + tweet);
        }
    }
}

//测试
public static void main(String[] args){
  ConcreteSubject cs = new ConcreteSubject();
  cs.registerObserver(new ConcreteNYTimes());
  cs.registerObserver(new ConcreteGuardian());
  cs.notifyObservers("Hi");
}

public interface CircuitBreakerRegistry extends Registry<CircuitBreaker, CircuitBreakerConfig> {
    
    /**
     * <1> 根据自定义的配置，创建CircuitBreakerRegistry实例  线程安全的单例
     */
    static CircuitBreakerRegistry of(CircuitBreakerConfig circuitBreakerConfig) {
        return new InMemoryCircuitBreakerRegistry(circuitBreakerConfig);
    }

    //...省略 都差不多 就是创建创建CircuitBreakerRegistry实例 
   
    //默认配置
    static CircuitBreakerRegistry ofDefaults() {
        return new InMemoryCircuitBreakerRegistry();

    //获取所有的熔断器  
    Seq<CircuitBreaker> getAllCircuitBreakers();

}

//io.github.resilience4j.circuitbreaker.internal.InMemoryCircuitBreakerRegistry
public InMemoryCircuitBreakerRegistry(CircuitBreakerConfig defaultConfig) {
        super(defaultConfig);
    }

//io.github.resilience4j.core.registry.AbstractRegistry
public AbstractRegistry(C defaultConfig) {
        this(defaultConfig, HashMap.empty());
    }

public AbstractRegistry(C defaultConfig, List<RegistryEventConsumer<E>> registryEventConsumers,
        Map<String, String> tags) {
        this.configurations = new ConcurrentHashMap<>();
        //可以看到是用ConcurrentHashMap 来作为管理熔断器的容器，保证线程安全
        this.entryMap = new ConcurrentHashMap<>();
        this.eventProcessor = new RegistryEventProcessor(
            Objects.requireNonNull(registryEventConsumers, CONSUMER_MUST_NOT_BE_NULL));
        this.registryTags = Objects.requireNonNull(tags, TAGS_MUST_NOT_BE_NULL);
        this.configurations
            .put(DEFAULT_CONFIG, Objects.requireNonNull(defaultConfig, CONFIG_MUST_NOT_BE_NULL));
    }

public class CircuitBreakerConfig {

    // 请求调用失败的阈值，百分比。默认是50%
    public static final int DEFAULT_FAILURE_RATE_THRESHOLD = 50; // Percentage
    // 请求调用慢请求的阀值， 百分比。 默认50%
    public static final int DEFAULT_SLOW_CALL_RATE_THRESHOLD = 100; // Percentage
    // 在open 状态等待的时间， 时间到了 就进去 half-open状态  默认60s
    public static final int DEFAULT_WAIT_DURATION_IN_OPEN_STATE = 60; // Seconds
    // 在half-open 状态下 允许通过请求数量  默认10个
    public static final int DEFAULT_PERMITTED_CALLS_IN_HALF_OPEN_STATE = 10;
    // 最少需要多少个请求调用 进行计算错误率  默认100
    public static final int DEFAULT_MINIMUM_NUMBER_OF_CALLS = 100;
    // 在滑动窗口的模式下， 需要多少个请求调用进行计算，  默认100
    public static final int DEFAULT_SLIDING_WINDOW_SIZE = 100;
    public static final int DEFAULT_SLOW_CALL_DURATION_THRESHOLD = 60; // Seconds
    // 统计的模式。  默认 计数， 还有一个是 滑动窗口
    public static final SlidingWindowType DEFAULT_SLIDING_WINDOW_TYPE = SlidingWindowType.COUNT_BASED;
    public static final boolean DEFAULT_WRITABLE_STACK_TRACE_ENABLED = true;
    // 是否记录请求调用失败的断言，默认所有异常都记录。
    private static final Predicate<Throwable> DEFAULT_RECORD_EXCEPTION_PREDICATE = throwable -> true;
    // 是否忽视请求调用失败的断言，  默认false
    private static final Predicate<Throwable> DEFAULT_IGNORE_EXCEPTION_PREDICATE = throwable -> false;
    // The default exception predicate counts all exceptions as failures.
    private Predicate<Throwable> recordExceptionPredicate = DEFAULT_RECORD_EXCEPTION_PREDICATE;
    // The default exception predicate ignores no exceptions.
    private Predicate<Throwable> ignoreExceptionPredicate = DEFAULT_IGNORE_EXCEPTION_PREDICATE;

    @SuppressWarnings("unchecked")
    // 请求调用失败，存储异常记录的集合
    private Class<? extends Throwable>[] recordExceptions = new Class[0];

    @SuppressWarnings("unchecked")
    // 请求调用失败，忽略异常记录的集合
    private Class<? extends Throwable>[] ignoreExceptions = new Class[0];
}

    //CircuitBreaker 的状态
    private interface CircuitBreakerState {

        //判断是否允许调用后端接口， 当CircuitBreaker 状态为open的时候这里返回false 反之true
        boolean tryAcquirePermission();
        //获取请求后端接口的许可
        void acquirePermission();
        //释放
        void releasePermission();

        /**
         * 请求调用失败，记录指标
         * 当达到设定的度量指标值后，调用状态机实例触发状态转换
         * @param duration  持续时间
         * @param durationUnit
         * @param throwable
         */
        void onError(long duration, TimeUnit durationUnit, Throwable throwable);

        /**
         * 调用成功，记录指标
         * 当达到设定的度量指标后，调用状态机实例触发状态转换
         * @param duration  持续时间
         * @param durationUnit
         */
        void onSuccess(long duration, TimeUnit durationUnit);

        int attempts();

        /**
         * 返回当前状态在CircuitBreaker 接口中对应的枚举值
         * @return
         */
        CircuitBreaker.State getState();

        /**
         * 返回当前状态封装的度量指标的类实例
         * @return
         */
        CircuitBreakerMetrics getMetrics();
    }

//CircuitBreakerStateMachine 的内部类
private class HalfOpenState implements CircuitBreakerState {
        //允许通过的数量
        private final AtomicInteger permittedNumberOfCalls;
        //是否是半开状态
        private final AtomicBoolean isHalfOpen;
 
        //是否允许通过，就是查看permittedNumberOfCalls 
        //减去这次请求是否>0是的话就是true，反之false
        @Override
        public boolean tryAcquirePermission() {
            if (permittedNumberOfCalls.getAndUpdate(current -> current == 0 ? current : --current)
                > 0) {
                return true;
            }
            circuitBreakerMetrics.onCallNotPermitted();
            return false;
        }
        //就是给permittedNumberOfCalls + 1
        @Override
        public void releasePermission() {
            permittedNumberOfCalls.incrementAndGet();
        }

        //onError 和 onSuccess 就是来进行状态转换的，调用了checkIfThresholdsExceeded 方法
        @Override
        public void onError(long duration, TimeUnit durationUnit, Throwable throwable) {
            // CircuitBreakerMetrics is thread-safe
            checkIfThresholdsExceeded(circuitBreakerMetrics.onError(duration, durationUnit));
        }
        @Override
        public void onSuccess(long duration, TimeUnit durationUnit) {
            // CircuitBreakerMetrics is thread-safe
            checkIfThresholdsExceeded(circuitBreakerMetrics.onSuccess(duration, durationUnit));
        }
    }

       private void checkIfThresholdsExceeded(Result result) {
            //<1> 如果已经达到了阀值， 那么先把Half-Open 状态关闭，然后调用transitionToOpenState 升级到open 状态
            if (Result.hasExceededThresholds(result)) {
                if (isHalfOpen.compareAndSet(true, false)) {
                    transitionToOpenState();
                }
            }
            //<2> 如果在阀值的下面， 那么就先把Half-Open 状态关闭，然后调用transitionToClosedState 降到close 状态
            if (result == BELOW_THRESHOLDS) {
                if (isHalfOpen.compareAndSet(true, false)) {
                    transitionToClosedState();
                }
            }
        }

public interface CircuitBreaker {
  static <T> Supplier<T> decorateSupplier(CircuitBreaker circuitBreaker, Supplier<T> supplier) {
        return () -> {
            //<1> 如果熔断器是打开状态，则抛出异常，不执行下面的代码
            circuitBreaker.acquirePermission();
            //计时 start
            long start = System.nanoTime();
            try {
                //<2> 执行被装饰的方法
                T returnValue = supplier.get();
                //计时 end
                long durationInNanos = System.nanoTime() - start;
                //<3> 调用成功 在ring buffer 中记录成功状态，计算失败率 失败达到阀值，则触发状态转换
                circuitBreaker.onSuccess(durationInNanos, TimeUnit.NANOSECONDS);
                return returnValue;
            } catch (Exception exception) {
                // Do not handle java.lang.Error
                long durationInNanos = System.nanoTime() - start;
                //<4> 调用失败  在ring buffer 中记录失败状态，计算失败率 失败达到阀值，则触发状态转换
                circuitBreaker.onError(durationInNanos, TimeUnit.NANOSECONDS, exception);
                throw exception;
            }
        };
    }
}