手写一个简易版断路器(hystrix)

Posted 2021-07-02 小奏技术

理论基础

根据熔断器模式(Circuit Breaker Pattern)

共有上面几种状态

• Closed:默认状态，Circuit Breaker 内部维护着最近的失败次数(failure count)。每操作失败一次，失败次数就会加一。当失败次数达到设定的阈值，就会进入Open状态
• Open: 操作不会执行，会立即失败。Circuit Breaker  内部维护一个计时器。当时间达到设定的阈值，就会从Open状态进入到Half-Open状态
• Half-Open: 只要操作失败一次，立即进入Open状态。否则增加操作成功的次数，当成功次数达到设定阈值，进入Closed状态

简化为自己的理解模型如下：

基于上面的理论，我们来实现我们的一个断路器

实现

定义状态

• State.java 首先我们定义一个枚举，定义上面的三种状态

public enum State {

    CLOSED,
    HALF_OPEN,
    OPEN
}

定义阈值配置

然后我们要有一个配置类，定义失败多少次进入Open状态，Half-Open状态的超时时间等。具体看代码

• Config.java

@Data
public class Config {

    /**
     * Closed 状态进入 Open 状态的错误个数阀值
     */
    private int failureCount = 5;

    /**
     * failureCount 统计时间窗口（超过该时间重置失败次数）
     */
    private long failureTimeInterval = 2 * 1000;

    /**
     * Open 状态进入 Half-Open 状态的超时时间
     */
    private int halfOpenTimeout = 5 * 1000;

    /**
     * Half-Open 状态进入 Open 状态的成功个数阀值
     */
    private int halfOpenSuccessCount = 2;


}

计数器

配置类我们基本都解决了，我们还需要有一个计数器，来记录上次操作调用失败的时间戳，失败次数等。用于和配置类比对来作状态转换。

• Counter.java

public class Counter {

    // Closed 状态进入 Open 状态的错误个数阀值
    private final int failureCount;

    // failureCount 统计时间窗口
    private final long failureTimeInterval;

    // 当前错误次数
    private final AtomicInteger currentCount;

    // 上一次调用失败的时间戳
    private long lastTime;

    // Half-Open 状态下成功次数
    private final AtomicInteger halfOpenSuccessCount;

    public Counter(int failureCount, long failureTimeInterval) {
        this.failureCount = failureCount;
        this.failureTimeInterval = failureTimeInterval;
        this.currentCount = new AtomicInteger(0);
        this.halfOpenSuccessCount = new AtomicInteger(0);
        this.lastTime = System.currentTimeMillis();
    }

    /**
     * 失败次数 + 1
     * synchronized 保证线程安全
     * @return
     */
    public synchronized int incrFailureCount() {
        long current = System.currentTimeMillis();
        // 超过时间窗口，当前失败次数重置为 0
        if (current - lastTime > failureTimeInterval) { 
            lastTime = current;
            currentCount.set(0);
        }
        return currentCount.getAndIncrement();
    }

    /**
     * half Open 状态下 成功数 + 1
     * @return
     */
    public int incrSuccessHalfOpenCount() {
        return this.halfOpenSuccessCount.incrementAndGet();
    }

    /**
     * 失败总次数是否超过阈值
     * @return
     */
    public boolean failureThresholdReached() {
        return getCurCount() >= failureCount;
    }

    public int getCurCount() {
        return currentCount.get();
    }

    /**
     * 重置
     */
    public synchronized void reset() {
        halfOpenSuccessCount.set(0);
        currentCount.set(0);
    }

}

核心实现类

上面的一切都准备好了，我们就来编写我们的核心实现类（CircuitBreaker）。CircuitBreaker主要是负责状态间的状态转换，并提供方法给外部调用达到限流降级作用。

• CircuitBreaker.java

public class CircuitBreaker {

    private State state;

    private final Config config;

    private final Counter counter;

    private long lastOpenedTime;

    public CircuitBreaker(Config config) {
        this.counter = new Counter(config.getFailureCount(), config.getFailureTimeInterval());
        this.state = CLOSED;
        this.config = config;
    }


    /**
     * 供外部调用达到限流降级作用
     * @param toRun
     * @param fallback 异常回滚方法
     * @param <T>
     * @return
     */
    public <T> T run(Supplier<T> toRun, Function<Throwable, T> fallback) {
        try {
            if (state == OPEN) {
                // 判断 是否超时可以进入 half-open状态
                if (halfOpenTimeout()) {
                    return halfOpenHandle(toRun, fallback);
                }
                // 直接执行失败回调方法
                return fallback.apply(new DegradeException("degrade by circuit breaker"));
                // 如果降级关闭则正常执行
            } else if (state == CLOSED) {
                T result = toRun.get();
                // 注意重置 错误数
                closed();
                return result;
            } else {
                // if (state == half-Open)
                return halfOpenHandle(toRun, fallback);
            }
        } catch (Exception e) {
            // 执行失败 错误次数+ 1
            counter.incrFailureCount();
            // 错误次数达到阀值，进入 open 状态
            if (counter.failureThresholdReached()) {
                open();
            }
            return fallback.apply(e);
        }
    }

    /**
     *  转换为 half-Open状态
     * @param toRun
     * @param fallback
     * @param <T>
     * @return
     */
    private <T> T halfOpenHandle(Supplier<T> toRun, Function<Throwable, T> fallback) {
        try {
            // closed 状态超时进入 half-open 状态
            halfOpen();
            T result = toRun.get();
            int halfOpenSuccCount = counter.incrSuccessHalfOpenCount();
            // half-open 状态成功次数到达阀值，进入 closed 状态
            if (halfOpenSuccCount >= this.config.getHalfOpenSuccessCount()) {
                closed();
            }
            return result;
        } catch (Exception e) {
            // half-open 状态发生一次错误进入 open 状态
            open();
            return fallback.apply(new DegradeException("degrade by circuit breaker"));
        }
    }

    /**
     * 判断是否 Open是否超时， 如果是则进入 halfOpen状态
     * @return
     */
    private boolean halfOpenTimeout() {
        return System.currentTimeMillis() - lastOpenedTime > config.getHalfOpenTimeout();
    }

    private void closed() {
        counter.reset();
        state = CLOSED;
    }

    private void open() {
        state = OPEN;
        lastOpenedTime = System.currentTimeMillis();
    }

    private void halfOpen() {
        state = HALF_OPEN;
    }

}

测试

首先我们测试Open状态

@Test
    public void testScene1() {
        CircuitBreaker cb = new CircuitBreaker(new Config());
        IntStream.range(0, 10).forEach(s -> {
            String result = cb.run(() -> {
                // 假装执行失败
                System.out.println("执行方法次数:" + s);
                int i = 1 / 0;
                return "Success";
            }, t -> {
                System.err.println(t);
                return "执行自定义降级方法";
            });
        });
    }

执行结果：

可以看到当达到5次后就触发降级操作，不再处理，直接降级

如果我们在每个方法执行后休眠2s，会看到不会触发降级操作

@Test
    public void testScene1() {
        CircuitBreaker cb = new CircuitBreaker(new Config());
        IntStream.range(0, 10).forEach(s -> {
            String result = cb.run(() -> {
                // 假装执行失败
                System.out.println("执行方法次数:" + s);
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                int i = 1 / 0;
                return "Success";
            }, t -> {
                System.err.println(t);
                return "执行自定义降级方法";
            });
        });
    }

执行结果：

测试 转换为half_open 然后转close

@Test
    public void testScene1() {
        CircuitBreaker cb = new CircuitBreaker(new Config());
        IntStream.range(0, 10).forEach(s -> {
            if (s == 5) {
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            String result = cb.run(() -> {
                // 假装执行失败
                System.out.println("执行方法次数:" + s);
                if (s < 5) {
                    int i = 1 / 0;
                }
                System.out.println("方法执行成功");
                return "Success";
            }, t -> {
                System.err.println(t);
                return "执行自定义降级方法";
            });
        });
    }

为了让测试查看方便，我在上面的状态转换里面加了一些输出，然后执行结果如下

可以看到最终都达到了我们想要的效果