为什么需要熔断限流？从一次生产事故说起

java.net.SocketTimeoutException: Read timed out
    at UserServiceClient.getUser(UserServiceClient.java:45)
    at OrderService.createOrder(OrderService.java:87)
    ...

// 订单服务配置
FlowRule flowRule = new FlowRule("createOrder");
flowRule.setCount(2000);  // 系统容量评估为2000 QPS
flowRule.setGrade(RuleConstant.FLOW_GRADE_QPS);

结果:
  - 流量8000 QPS → 通过2000 QPS + 拒绝6000 QPS
  - 订单服务响应时间稳定在50ms
  - 线程池占用稳定在50/200

// 用户服务调用熔断配置
DegradeRule degradeRule = new DegradeRule("getUserInfo");
degradeRule.setGrade(RuleConstant.DEGRADE_GRADE_RT);
degradeRule.setCount(1000);  // RT超过1秒
degradeRule.setSlowRatioThreshold(0.5);  // 慢调用比例50%
degradeRule.setTimeWindow(10);  // 熔断10秒

结果:
  - 用户服务响应慢(5秒)
  - 触发熔断,订单服务停止调用用户服务
  - 快速失败,返回默认用户信息
  - 订单创建降级为"游客下单"模式

@Configuration
public class ThreadPoolConfig {

    // 核心接口线程池(优先级最高)
    @Bean("coreThreadPool")
    public ThreadPoolExecutor coreThreadPool() {
        return new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(100);    // 核心线程100
        executor.setMaxPoolSize(150);     // 最大线程150
        executor.setQueueCapacity(1000);  // 队列1000
        executor.setThreadNamePrefix("core-");
        executor.initialize();
        return executor;
    }

    // 常用接口线程池(优先级中)
    @Bean("commonThreadPool")
    public ThreadPoolExecutor commonThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(30);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(500);
        executor.setThreadNamePrefix("common-");
        executor.initialize();
        return executor;
    }

    // 非核心接口线程池(优先级低)
    @Bean("nonCoreThreadPool")
    public ThreadPoolExecutor nonCoreThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("non-core-");
        executor.initialize();
        return executor;
    }
}

// 接口绑定线程池
@RestController
public class OrderController {

    @Autowired
    @Qualifier("coreThreadPool")
    private Executor coreThreadPool;

    @PostMapping("/order/create")
    public CompletableFuture<Result> createOrder(@RequestBody OrderRequest request) {
        return CompletableFuture.supplyAsync(() -> {
            // 创建订单业务逻辑
            return Result.success(orderService.createOrder(request));
        }, coreThreadPool);
    }
}

@RestController
public class ReportController {

    @Autowired
    @Qualifier("nonCoreThreadPool")
    private Executor nonCoreThreadPool;

    @GetMapping("/report/export")
    public CompletableFuture<Result> exportReport() {
        return CompletableFuture.supplyAsync(() -> {
            // 导出报表业务逻辑(耗时5秒)
            return Result.success(reportService.export());
        }, nonCoreThreadPool);
    }
}

@RestController
public class OrderController {

    @Autowired
    private OrderService orderService;

    @PostMapping("/order/create")
    public Result createOrder(@RequestBody OrderRequest request) {
        // 无任何限流保护
        return Result.success(orderService.createOrder(request));
    }
}

/**
 * 固定窗口计数器
 * 算法: 每秒最多允许N个请求
 */
public class FixedWindowRateLimiter {

    private final int maxRequests;  // 窗口内最大请求数
    private final AtomicInteger counter = new AtomicInteger(0);
    private volatile long windowStart = System.currentTimeMillis();

    public FixedWindowRateLimiter(int maxRequests) {
        this.maxRequests = maxRequests;
    }

    public boolean tryAcquire() {
        long now = System.currentTimeMillis();

        // 新窗口开始,重置计数器
        if (now - windowStart >= 1000) {
            synchronized (this) {
                if (now - windowStart >= 1000) {
                    counter.set(0);
                    windowStart = now;
                }
            }
        }

        // 当前窗口未超限
        if (counter.incrementAndGet() <= maxRequests) {
            return true;
        }

        // 超限,拒绝
        counter.decrementAndGet();
        return false;
    }
}

// 使用示例
@RestController
public class OrderController {

    private FixedWindowRateLimiter rateLimiter = new FixedWindowRateLimiter(2000);

    @PostMapping("/order/create")
    public Result createOrder(@RequestBody OrderRequest request) {
        if (!rateLimiter.tryAcquire()) {
            return Result.fail("系统繁忙,请稍后重试");
        }

        return Result.success(orderService.createOrder(request));
    }
}

/**
 * 滑动窗口限流器
 * 算法: 统计过去N毫秒内的请求数
 */
public class SlidingWindowRateLimiter {

    private final int maxRequests;
    private final long windowSize;  // 窗口大小(毫秒)
    private final Queue<Long> timestamps = new LinkedList<>();

    public SlidingWindowRateLimiter(int maxRequests, long windowSizeMs) {
        this.maxRequests = maxRequests;
        this.windowSize = windowSizeMs;
    }

    public synchronized boolean tryAcquire() {
        long now = System.currentTimeMillis();

        // 移除窗口外的时间戳
        while (!timestamps.isEmpty() && now - timestamps.peek() >= windowSize) {
            timestamps.poll();
        }

        // 当前窗口未超限
        if (timestamps.size() < maxRequests) {
            timestamps.offer(now);
            return true;
        }

        // 超限,拒绝
        return false;
    }
}

/**
 * 漏桶算法
 * 算法: 请求进入桶,按固定速率流出
 */
public class LeakyBucketRateLimiter {

    private final int capacity;  // 桶容量
    private final int leakRate;  // 流出速率(请求/秒)
    private final AtomicInteger water = new AtomicInteger(0);
    private volatile long lastLeakTime = System.currentTimeMillis();

    public LeakyBucketRateLimiter(int capacity, int leakRate) {
        this.capacity = capacity;
        this.leakRate = leakRate;
    }

    public synchronized boolean tryAcquire() {
        // 先漏水
        leak();

        // 尝试加水
        if (water.get() < capacity) {
            water.incrementAndGet();
            return true;
        }

        // 桶满,拒绝
        return false;
    }

    private void leak() {
        long now = System.currentTimeMillis();
        long elapsedTime = now - lastLeakTime;

        // 计算漏出的水量
        int leaked = (int) (elapsedTime * leakRate / 1000);

        if (leaked > 0) {
            int currentWater = water.get();
            int newWater = Math.max(0, currentWater - leaked);
            water.set(newWater);
            lastLeakTime = now;
        }
    }
}

/**
 * 令牌桶算法
 * 算法: 匀速生成令牌,请求消费令牌
 */
public class TokenBucketRateLimiter {

    private final int capacity;  // 桶容量(最大突发量)
    private final int rate;      // 生成速率(令牌/秒)
    private final AtomicInteger tokens = new AtomicInteger(0);
    private volatile long lastRefillTime = System.currentTimeMillis();

    public TokenBucketRateLimiter(int capacity, int rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.tokens.set(capacity);  // 初始化令牌桶为满
    }

    public synchronized boolean tryAcquire() {
        // 先补充令牌
        refill();

        // 消费令牌
        if (tokens.get() > 0) {
            tokens.decrementAndGet();
            return true;
        }

        // 无令牌,拒绝
        return false;
    }

    private void refill() {
        long now = System.currentTimeMillis();
        long elapsedTime = now - lastRefillTime;

        // 计算新生成的令牌数
        int newTokens = (int) (elapsedTime * rate / 1000);

        if (newTokens > 0) {
            int currentTokens = tokens.get();
            int updatedTokens = Math.min(capacity, currentTokens + newTokens);
            tokens.set(updatedTokens);
            lastRefillTime = now;
        }
    }
}

// 使用Guava RateLimiter(推荐)
@RestController
public class OrderController {

    // 每秒生成2000个令牌
    private RateLimiter rateLimiter = RateLimiter.create(2000);

    @PostMapping("/order/create")
    public Result createOrder(@RequestBody OrderRequest request) {
        // 非阻塞式获取令牌
        if (!rateLimiter.tryAcquire()) {
            return Result.fail("系统繁忙,请稍后重试");
        }

        // 阻塞式获取令牌(可设置超时)
        // boolean acquired = rateLimiter.tryAcquire(100, TimeUnit.MILLISECONDS);

        return Result.success(orderService.createOrder(request));
    }
}

// 危险代码!
@RestController
public class OrderController {

    @Autowired
    private RestTemplate restTemplate;

    @PostMapping("/order/create")
    public Result createOrder(@RequestBody OrderRequest request) {
        // 调用用户服务(无超时控制)
        String url = "http://user-service/users/" + request.getUserId();
        User user = restTemplate.getForObject(url, User.class);

        // 如果用户服务响应5秒,这个线程会阻塞5秒
        // 40个这样的请求,就会占满200个线程

        // ...后续业务逻辑
    }
}

@Configuration
public class RestTemplateConfig {

    @Bean
    public RestTemplate restTemplate() {
        // 配置HTTP客户端
        HttpComponentsClientHttpRequestFactory factory =
            new HttpComponentsClientHttpRequestFactory();

        // 连接超时: 1秒
        factory.setConnectTimeout(1000);

        // 读超时: 1秒
        factory.setReadTimeout(1000);

        // 连接池配置
        PoolingHttpClientConnectionManager connectionManager =
            new PoolingHttpClientConnectionManager();
        connectionManager.setMaxTotal(200);        // 最大连接数
        connectionManager.setDefaultMaxPerRoute(20); // 每个路由最大连接数

        CloseableHttpClient httpClient = HttpClientBuilder.create()
            .setConnectionManager(connectionManager)
            .build();

        factory.setHttpClient(httpClient);

        return new RestTemplate(factory);
    }
}

// 使用
@RestController
public class OrderController {

    @Autowired
    private RestTemplate restTemplate;

    @PostMapping("/order/create")
    public Result createOrder(@RequestBody OrderRequest request) {
        try {
            // 超时后会抛出 ResourceAccessException
            User user = restTemplate.getForObject(
                "http://user-service/users/" + request.getUserId(),
                User.class
            );
            // ...业务逻辑
        } catch (ResourceAccessException e) {
            // 超时处理
            log.warn("用户服务调用超时: {}", request.getUserId(), e);
            return Result.fail("系统繁忙,请稍后重试");
        }
    }
}

@Configuration
public class SentinelConfig {

    @PostConstruct
    public void initDegradeRules() {
        List<DegradeRule> rules = new ArrayList<>();

        // 规则1: 慢调用比例熔断
        DegradeRule slowCallRule = new DegradeRule("getUserInfo");
        slowCallRule.setGrade(RuleConstant.DEGRADE_GRADE_RT);
        slowCallRule.setCount(1000);  // RT超过1秒算慢调用
        slowCallRule.setSlowRatioThreshold(0.5);  // 慢调用比例50%
        slowCallRule.setMinRequestAmount(5);  // 最小请求数5个
        slowCallRule.setStatIntervalMs(1000);  // 统计时长1秒
        slowCallRule.setTimeWindow(10);  // 熔断时长10秒
        rules.add(slowCallRule);

        // 规则2: 异常比例熔断
        DegradeRule exceptionRule = new DegradeRule("getUserInfo");
        exceptionRule.setGrade(RuleConstant.DEGRADE_GRADE_EXCEPTION_RATIO);
        exceptionRule.setCount(0.5);  // 异常比例50%
        exceptionRule.setMinRequestAmount(5);
        exceptionRule.setStatIntervalMs(1000);
        exceptionRule.setTimeWindow(10);
        rules.add(exceptionRule);

        // 规则3: 异常数熔断
        DegradeRule exceptionCountRule = new DegradeRule("getUserInfo");
        exceptionCountRule.setGrade(RuleConstant.DEGRADE_GRADE_EXCEPTION_COUNT);
        exceptionCountRule.setCount(10);  // 异常数10个
        exceptionCountRule.setStatIntervalMs(1000);
        exceptionCountRule.setTimeWindow(10);
        rules.add(exceptionCountRule);

        DegradeRuleManager.loadRules(rules);
    }
}

// 业务代码
@Service
public class OrderService {

    @Autowired
    private UserServiceClient userServiceClient;

    @SentinelResource(
        value = "getUserInfo",
        fallback = "getUserFallback",       // 降级方法
        blockHandler = "getUserBlockHandler" // 限流/熔断处理
    )
    public User getUser(Long userId) {
        return userServiceClient.getUser(userId);
    }

    // 降级方法: 返回默认用户
    public User getUserFallback(Long userId, Throwable ex) {
        log.warn("用户服务调用失败,返回默认用户: {}", userId, ex);
        User defaultUser = new User();
        defaultUser.setId(userId);
        defaultUser.setName("默认用户");
        defaultUser.setLevel(UserLevel.GUEST);
        return defaultUser;
    }

    // 限流/熔断处理
    public User getUserBlockHandler(Long userId, BlockException ex) {
        log.warn("用户服务被限流或熔断: {}", userId);
        return getUserFallback(userId, ex);
    }
}

// Hystrix命令
public class GetUserCommand extends HystrixCommand<User> {

    private Long userId;
    private UserServiceClient userServiceClient;

    public GetUserCommand(Long userId, UserServiceClient client) {
        super(HystrixCommandGroupKey.Factory.asKey("UserService"));
        this.userId = userId;
        this.userServiceClient = client;
    }

    @Override
    protected User run() throws Exception {
        // 在独立线程池中执行
        return userServiceClient.getUser(userId);
    }

    @Override
    protected User getFallback() {
        // 降级逻辑
        User defaultUser = new User();
        defaultUser.setId(userId);
        defaultUser.setName("默认用户");
        return defaultUser;
    }
}

// 使用
User user = new GetUserCommand(userId, userServiceClient).execute();

// Sentinel资源定义
@Service
public class OrderService {

    @SentinelResource(
        value = "getUserInfo",
        fallback = "getUserFallback"
    )
    public User getUser(Long userId) {
        return userServiceClient.getUser(userId);
    }

    public User getUserFallback(Long userId, Throwable ex) {
        User defaultUser = new User();
        defaultUser.setId(userId);
        defaultUser.setName("默认用户");
        return defaultUser;
    }
}

// Sentinel规则配置
@Configuration
public class SentinelConfig {

    @PostConstruct
    public void initRules() {
        // 限流规则
        FlowRule flowRule = new FlowRule("getUserInfo");
        flowRule.setGrade(RuleConstant.FLOW_GRADE_THREAD);
        flowRule.setCount(10);  // 最多10个并发线程
        FlowRuleManager.loadRules(Collections.singletonList(flowRule));

        // 熔断规则
        DegradeRule degradeRule = new DegradeRule("getUserInfo");
        degradeRule.setGrade(RuleConstant.DEGRADE_GRADE_RT);
        degradeRule.setCount(1000);  // RT > 1秒
        degradeRule.setSlowRatioThreshold(0.5);  // 慢调用比例50%
        degradeRule.setTimeWindow(10);  // 熔断10秒
        DegradeRuleManager.loadRules(Collections.singletonList(degradeRule));
    }
}

// 生产者: 订单请求进入队列
@RestController
public class OrderController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostMapping("/order/create")
    public Result createOrder(@RequestBody OrderRequest request) {
        // 发送到消息队列
        rabbitTemplate.convertAndSend("order.exchange", "order.create", request);
        return Result.success("订单已提交,请稍后查看");
    }
}

// 消费者: 异步处理订单
@Component
public class OrderConsumer {

    @RabbitListener(queues = "order.queue")
    public void handleOrder(OrderRequest request) {
        // 处理订单(控制消费速率,如100/秒)
        orderService.createOrder(request);
    }
}

@Configuration
public class SentinelConfig {

    @PostConstruct
    public void initFlowRules() {
        List<FlowRule> rules = new ArrayList<>();

        // 核心接口限流(QPS)
        FlowRule coreRule = new FlowRule("createOrder");
        coreRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        coreRule.setCount(1000);  // 单机1000 QPS
        coreRule.setStrategy(RuleConstant.STRATEGY_DIRECT);
        rules.add(coreRule);

        // 非核心接口限流(并发线程数)
        FlowRule nonCoreRule = new FlowRule("exportReport");
        nonCoreRule.setGrade(RuleConstant.FLOW_GRADE_THREAD);
        nonCoreRule.setCount(10);  // 最多10个并发
        nonCoreRule.setStrategy(RuleConstant.STRATEGY_DIRECT);
        rules.add(nonCoreRule);

        // IP级别限流
        FlowRule ipRule = new FlowRule("queryData");
        ipRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        ipRule.setCount(10);  // 单IP 10 QPS
        ipRule.setLimitApp("origin");  // 根据来源限流
        rules.add(ipRule);

        FlowRuleManager.loadRules(rules);
    }
}

@PostConstruct
public void initDegradeRules() {
    List<DegradeRule> rules = new ArrayList<>();

    // 慢调用比例熔断
    DegradeRule slowCallRule = new DegradeRule("getUserInfo");
    slowCallRule.setGrade(RuleConstant.DEGRADE_GRADE_RT);
    slowCallRule.setCount(1000);  // RT超过1秒
    slowCallRule.setSlowRatioThreshold(0.5);  // 慢调用比例50%
    slowCallRule.setMinRequestAmount(10);  // 最小请求数10个
    slowCallRule.setStatIntervalMs(1000);  // 统计时长1秒
    slowCallRule.setTimeWindow(10);  // 熔断时长10秒
    rules.add(slowCallRule);

    // 异常比例熔断
    DegradeRule exceptionRule = new DegradeRule("paymentService");
    exceptionRule.setGrade(RuleConstant.DEGRADE_GRADE_EXCEPTION_RATIO);
    exceptionRule.setCount(0.5);  // 异常比例50%
    exceptionRule.setMinRequestAmount(10);
    exceptionRule.setStatIntervalMs(1000);
    exceptionRule.setTimeWindow(10);
    rules.add(exceptionRule);

    DegradeRuleManager.loadRules(rules);
}

@Service
public class OrderService {

    // 远程调用用户服务
    @SentinelResource(
        value = "getUserInfo",
        fallback = "getUserFallback"
    )
    public User getUser(Long userId) {
        return userServiceClient.getUser(userId);
    }

    // 降级方法: 返回默认用户
    public User getUserFallback(Long userId, Throwable ex) {
        log.warn("用户服务调用失败,返回默认用户: userId={}", userId, ex);

        // 返回默认用户(游客模式)
        User guestUser = new User();
        guestUser.setId(userId);
        guestUser.setName("游客");
        guestUser.setLevel(UserLevel.GUEST);
        guestUser.setAvatar(DEFAULT_AVATAR);

        return guestUser;
    }
}

@Configuration
public class SentinelMetricConfig {

    @Bean
    public MeterRegistryCustomizer<MeterRegistry> sentinelMetrics() {
        return registry -> {
            // 限流次数
            Gauge.builder("sentinel.block.count", () ->
                getBlockCount()
            ).register(registry);

            // 熔断次数
            Gauge.builder("sentinel.degrade.count", () ->
                getDegradeCount()
            ).register(registry);

            // P99响应时间
            Gauge.builder("sentinel.rt.p99", () ->
                getP99RT()
            ).register(registry);
        };
    }
}

// 告警规则(Prometheus Alertmanager)
groups:
  - name: sentinel_alerts
    rules:
      # 限流告警
      - alert: SentinelBlockTooMany
        expr: rate(sentinel_block_count[1m]) > 100
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "限流次数过多"
          description: "{{ $labels.resource }} 限流次数 > 100/min"

      # 熔断告警
      - alert: SentinelDegradeTooMany
        expr: rate(sentinel_degrade_count[1m]) > 10
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "熔断次数过多"
          description: "{{ $labels.resource }} 熔断次数 > 10/min"

      # 响应时间告警
      - alert: SentinelRTTooHigh
        expr: sentinel_rt_p99 > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "响应时间过长"
          description: "{{ $labels.resource }} P99响应时间 > 1秒"