手写一个简易版断路器(hystrix)-蒲公英云

理论基础

根据熔断器模式(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 "执行自定义降级方法";
            });
        });
    }