2. 实现熔断器模式保护下游服务

任务说明

在微服务或分布式系统中，下游服务（如数据库、第三方API）可能因过载或故障而响应缓慢。若上游持续向故障下游发送请求，会导致自身线程池耗尽、级联故障。本案例实现一个最小可用的熔断器（Circuit Breaker），在连续失败达到阈值后快速失败（不调用下游），并在冷却期后允许试探性请求以恢复服务。

熔断器状态与核心逻辑

熔断器维护三种状态：CLOSED（正常调用）、OPEN（熔断打开，直接拒绝）、HALF_OPEN（半开，允许少量试探请求）。状态迁移规则：

• CLOSED 状态下，连续失败次数达到 failureThreshold 则转为 OPEN；
• OPEN 状态持续 timeoutMs 后自动转为 HALF_OPEN；
• HALF_OPEN 状态下，若试探请求成功则重置计数并转为 CLOSED；若失败则回到 OPEN。

代码实现

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

public class CircuitBreaker {
    public enum State { CLOSED, OPEN, HALF_OPEN }

    private final int failureThreshold;
    private final long timeoutMs;
    private final AtomicReference<State> state = new AtomicReference<>(State.CLOSED);
    private final AtomicInteger consecutiveFailures = new AtomicInteger(0);
    private volatile Instant openedAt;

    public CircuitBreaker(int failureThreshold, long timeoutMs) {
        if (failureThreshold <= 0 || timeoutMs <= 0) {
            throw new IllegalArgumentException("threshold and timeout must be positive");
        }
        this.failureThreshold = failureThreshold;
        this.timeoutMs = timeoutMs;
    }

    /**
     * 执行受保护的下游调用。
     * @param action 下游调用逻辑，可抛出异常表示失败
     * @return 下游返回值
     * @throws Exception 下游异常或熔断快速失败异常
     */
    public <T> T execute(Callable<T> action) throws Exception {
        if (!allowRequest()) {
            throw new IllegalStateException("Circuit breaker is OPEN, fast fail");
        }

        try {
            T result = action.call();
            onSuccess();
            return result;
        } catch (Exception e) {
            onFailure();
            throw e;
        }
    }

    private boolean allowRequest() {
        State current = state.get();
        if (current == State.CLOSED) {
            return true;
        }
        if (current == State.OPEN) {
            // 检查冷却时间是否已过
            if (Duration.between(openedAt, Instant.now()).toMillis() >= timeoutMs) {
                // 转为半开，允许一个试探请求
                if (state.compareAndSet(State.OPEN, State.HALF_OPEN)) {
                    return true;
                }
            }
            return false;
        }
        // HALF_OPEN：只允许一个试探请求
        return false;
    }

    private void onSuccess() {
        State current = state.get();
        if (current == State.HALF_OPEN) {
            // 试探成功，重置并关闭
            state.set(State.CLOSED);
            consecutiveFailures.set(0);
        } else if (current == State.CLOSED) {
            // 连续成功可重置失败计数（可选策略，这里保持简单：成功即清零）
            consecutiveFailures.set(0);
        }
    }

    private void onFailure() {
        State current = state.get();
        if (current == State.HALF_OPEN) {
            // 试探失败，回到打开
            state.set(State.OPEN);
            openedAt = Instant.now();
        } else if (current == State.CLOSED) {
            int failures = consecutiveFailures.incrementAndGet();
            if (failures >= failureThreshold) {
                state.set(State.OPEN);
                openedAt = Instant.now();
            }
        }
    }

    public State getState() {
        return state.get();
    }

    @FunctionalInterface
    public interface Callable<T> {
        T call() throws Exception;
    }
}

使用示例与验证

以下代码演示熔断器从关闭到打开再到半开恢复的完整路径：

public class Demo {
    public static void main(String[] args) throws Exception {
        // 阈值2次失败，超时500ms
        CircuitBreaker cb = new CircuitBreaker(2, 500);

        // 模拟下游：前3次抛异常，之后成功
        int[] callCount = {0};
        CircuitBreaker.Callable<String> downstream = () -> {
            callCount[0]++;
            if (callCount[0] <= 3) {
                throw new RuntimeException("downstream error");
            }
            return "ok";
        };

        // 第1次失败
        try { cb.execute(downstream); } catch (Exception e) { }
        System.out.println("After 1st failure: " + cb.getState()); // CLOSED

        // 第2次失败，达到阈值，熔断打开
        try { cb.execute(downstream); } catch (Exception e) { }
        System.out.println("After 2nd failure: " + cb.getState()); // OPEN

        // 熔断打开，快速失败
        try {
            cb.execute(downstream);
        } catch (IllegalStateException e) {
            System.out.println("Fast fail: " + e.getMessage());
        }

        // 等待冷却时间过去
        Thread.sleep(600);

        // 半开状态，允许试探请求（第3次调用，仍失败）
        try { cb.execute(downstream); } catch (Exception e) { }
        System.out.println("After half-open failure: " + cb.getState()); // OPEN

        // 再次等待冷却
        Thread.sleep(600);

        // 半开试探成功（第4次调用，成功）
        String result = cb.execute(downstream);
        System.out.println("Recovered result: " + result);
        System.out.println("Final state: " + cb.getState()); // CLOSED
    }
}

验证方法与判定标准

运行 Demo 类，观察输出顺序：

1. 前两次失败后状态从 CLOSED 变为 OPEN，说明连续失败计数与阈值判断正确；
2. OPEN 状态下调用立即抛出 IllegalStateException，不执行下游逻辑，说明快速失败生效；
3. 等待超过 timeoutMs 后，状态自动进入 HALF_OPEN 并允许试探请求；
4. 试探失败后回到 OPEN，试探成功后回到 CLOSED 且失败计数清零。

若输出与上述预期一致，则熔断器核心状态机工作正常。可调整 failureThreshold 与 timeoutMs 参数验证不同阈值下的行为。