分布式场景下限流算法的挑战与Resilience4j深度剖析

前言：为什么需要Resilience4j？

在分布式系统架构中，服务依赖关系日益复杂，一个服务的故障可能通过调用链引发级联失败，最终导致整个系统崩溃。为了保障系统的稳定性和可用性，容错机制成为核心设计要素。Resilience4j作为一款轻量级Java容错库，专为现代分布式系统打造，提供了限流、熔断、重试、舱壁等完整的容错能力。

与传统容错库（如Netflix Hystrix）相比，Resilience4j具有轻量级、模块化、函数式编程友好的特点，仅依赖Vavr库，无冗余依赖，性能开销极低。它完美支持Java 8+的Lambda表达式和响应式编程，可无缝集成Spring Boot、Spring Cloud等主流框架，成为分布式系统容错设计的首选工具。

本文将从分布式限流的挑战出发，逐步深入Resilience4j的设计思想、架构原理、核心模块、实战用法、源码解析及自定义拓展，帮助开发者全面掌握这款工具的使用与设计精髓。

一、分布式场景下限流算法面临的挑战

在大规模分布式系统中，限流是保护系统免受流量冲击的第一道防线。然而，分布式环境的复杂性为限流算法带来了诸多独特挑战，这些挑战直接影响限流策略的有效性和系统的稳定性。

1.1 多节点时钟不一致问题

分布式系统由多个物理节点组成，各节点的本地时钟可能存在细微偏差（即使通过NTP同步，也可能存在毫秒级误差）。对于基于时间窗口的限流算法（如固定窗口、滑动窗口），时钟不一致会导致限流准确性严重下降。

固定窗口算法的问题：假设系统设定“每分钟允许100个请求”，节点A的时间窗口为0:00-0:59，节点B因时钟快1秒，窗口为0:00-0:58。在0:58-0:59这1秒内，节点B认为窗口已结束，允许新请求通过，而节点A仍在当前窗口计数，导致全局流量超过100次/分钟。

滑动窗口算法的问题：滑动窗口需精确计算窗口移动时间和请求分布，时钟偏差会导致各节点窗口重叠范围不一致，计数统计失真，最终限流阈值形同虚设。

1.2 网络延迟与数据同步难题

分布式限流需跨节点共享流量数据（如总请求数、剩余令牌数），而网络延迟会导致数据同步不及时，引发“限流失效”。

例如，基于Redis的分布式令牌桶算法中，节点A消耗50个令牌后更新Redis，但节点B因网络延迟未获取最新数据，仍按旧值（剩余30个令牌）允许请求通过，导致实际流量远超阈值。在高并发场景下，网络带宽被大量请求占用，进一步加剧同步延迟，形成恶性循环。

1.3 高并发下的性能瓶颈

单机限流算法（如本地令牌桶）可通过内存操作实现低延迟，但分布式场景下，限流逻辑依赖中心化存储（如Redis、ZooKeeper），每次请求需进行网络I/O，在高并发下成为性能瓶颈。

Redis压力过大：每秒10万次请求的系统，每次请求需调用Redis的INCR或HSET操作，Redis节点可能因QPS过高导致响应延迟甚至宕机。

网络开销累积：即使单次Redis操作耗时1ms，10万次/秒的请求也会产生100秒的总网络耗时，直接拖累系统响应速度。

1.4 限流策略的动态调整复杂性

分布式系统的流量具有突发性（如电商秒杀、直播带货），需动态调整限流阈值以适应业务变化。但分布式环境下，动态调整面临两大难题：

策略一致性问题：不同节点可能运行不同版本的业务代码，对限流策略的解析逻辑存在差异，导致新策略无法在所有节点同步生效。

流量监测准确性问题：实时监测全局流量需汇总各节点数据，但网络延迟、数据丢失会导致流量统计失真，基于错误数据的阈值调整可能“雪上加霜”。

1.5 分布式系统拓扑结构变化影响

分布式系统需支持节点动态扩缩容（如K8s自动扩缩容），拓扑结构变化会导致流量重新分配，若限流算法无法感知变化，会出现“资源浪费”或“局部过载”。

新增节点未同步配置：新节点加入集群时，若未及时获取最新限流策略和全局计数，可能无限制放行请求，成为系统短板。

节点下线数据残留：节点下线后，其本地缓存的限流数据未清理，可能导致其他节点误判全局流量，引发过度限流。

1.6 跨地域部署的流量不均衡问题

全球化系统通常跨地域部署（如北美、欧洲、亚太节点），不同地域的流量特征差异显著（如时区性高峰）。若采用全局统一限流阈值，会导致：

高流量地域频繁限流：亚太节点在夜间流量高峰时被限流，影响用户体验；

低流量地域资源闲置：欧洲节点流量低谷时，限流阈值未充分利用硬件资源。

如何根据地域特征动态调整阈值，实现“流量削峰”与“资源利用”的平衡，是分布式限流的核心难题。

1.7 多租户场景下的隔离性挑战

SaaS系统中，多租户共享硬件资源，需保证租户间的资源隔离，避免单个租户的流量过载影响其他租户。传统限流算法难以实现精细化隔离：

共享配额抢占：多个租户共享“1000次/秒”的限流阈值时，某个租户突发流量可能耗尽配额，导致其他租户请求被错误拒绝。

SLA差异化需求：付费租户需更高的限流配额（如200次/秒），免费租户配额较低（如50次/秒），传统限流无法区分租户等级。

1.8 限流与其他容错机制的协同问题

限流需与熔断、降级、重试等机制协同工作，但机制间的交互可能产生副作用：

限流触发重试风暴：限流拒绝请求后，上游服务若开启重试，会产生更多请求，加剧限流压力；

熔断与限流阈值冲突：若限流阈值过低，大量请求被拒绝，可能导致熔断机制误判服务故障而触发熔断；若阈值过高，熔断可能因未及时拦截故障请求而失效。

如何设计机制间的协同策略，避免“1+1 < 2”的负面效果，是分布式容错设计的关键。

二、Resilience4j基础：从概念到架构

Resilience4j是一款专为Java 8+设计的轻量级容错库，旨在通过模块化的容错机制提升分布式系统的弹性。本节将从核心概念、设计思想和架构层面，为读者构建Resilience4j的基础认知。

2.1 Resilience4j概述

Resilience4j的诞生源于对传统容错库（如Hystrix）的改进。Hystrix虽曾是容错领域的标杆，但存在依赖繁重、对函数式编程支持不足等问题，且已于2018年停止维护。Resilience4j应运而生，它具有以下核心特性：

轻量级：仅依赖Vavr（一款函数式编程库），无其他冗余依赖，Jar包大小不足1MB，对项目侵入性极低；

模块化：将限流、熔断、重试等容错机制封装为独立模块，开发者可按需引入，避免“引入全量功能”导致的资源浪费；

函数式友好：原生支持Java 8 Lambda表达式和函数式接口，可通过decorateSupplier等方法轻松包装业务逻辑，代码简洁易读；

响应式支持：完美集成RxJava、Project Reactor等响应式框架，支持异步非阻塞场景下的容错处理；

可扩展性：提供丰富的接口和扩展点，支持自定义限流算法、事件处理逻辑等。

Resilience4j的核心目标是：在不牺牲性能的前提下，为分布式系统提供灵活、可靠的容错能力。

2.2 Resilience4j的设计思想

Resilience4j的设计遵循三大核心思想，这些思想决定了其架构设计和使用方式。

2.2.1 模块化与可组合性

Resilience4j将每种容错机制封装为独立模块（如resilience4j-circuitbreaker、resilience4j-ratelimiter），模块间通过接口解耦，可单独使用或组合使用。这种设计的优势在于：

按需引入：仅引入项目所需的模块（如只需限流则引入ratelimiter），减少依赖体积；

灵活组合：通过“装饰器模式”组合多个模块（如“限流+熔断+重试”），满足复杂业务场景；

低耦合：模块间无强依赖，修改某一模块的实现不影响其他模块。

例如，为一个远程调用添加容错机制时，可先通过限流器控制请求速率，再通过断路器监控调用结果，最后通过重试处理暂时性故障，各机制独立生效又协同工作。

2.2.2 函数式编程驱动

Resilience4j充分利用Java 8的函数式特性，将容错逻辑抽象为“高阶函数”，通过装饰器模式包装业务代码。这种设计带来两大优势：

代码简洁：无需编写大量样板代码（如try-catch块、状态判断逻辑），通过Lambda表达式即可完成容错配置；

无侵入性：业务代码与容错逻辑分离，开发者可专注于核心业务，容错逻辑通过外部装饰实现。

示例代码如下：


// 原始业务逻辑
Supplier<OrderDTO> orderSupplier = () -> paymentService.processPayment(request);

// 用熔断和限流装饰业务逻辑
Supplier<OrderDTO> decoratedSupplier = CircuitBreaker
    .decorateSupplier(circuitBreaker,
        RateLimiter.decorateSupplier(rateLimiter, orderSupplier));

2.2.3 基于事件的监控与扩展

Resilience4j为每个模块设计了完善的事件发布机制，当状态变化或关键操作发生时（如断路器打开、限流触发），会发布相应事件。开发者可通过订阅事件实现：

实时监控：记录事件日志，跟踪系统运行状态；

告警通知：当异常事件发生（如断路器打开）时，发送告警邮件或短信；

自定义扩展：基于事件触发业务逻辑（如限流时动态扩容资源）。

事件机制使Resilience4j具备极强的可观测性和扩展性，是构建“可感知、可调控”弹性系统的基础。

2.3 Resilience4j的架构

Resilience4j采用分层架构设计，各层职责清晰，通过接口交互，确保架构的灵活性和可扩展性。其架构从上到下分为四层：集成层、事件监听与监控层、核心模块层、配置层。

2.3.1 核心模块层

核心模块层是Resilience4j的功能实现层，包含六种核心容错模块，每种模块解决特定的容错问题：

模块名称	核心功能	解决的问题
CircuitBreaker（断路器）	监控服务调用结果，当失败率过高时“断开”调用，避免故障扩散。	服务依赖故障导致的级联失败问题。
RateLimiter（限流器）	控制请求速率，防止系统因流量过大而过载。	突发流量导致的系统资源耗尽问题。
Retry（重试）	当调用失败时自动重试，解决暂时性故障（如网络抖动）。	瞬时故障导致的调用失败问题。
Bulkhead（舱壁）	限制并发请求数量，隔离不同服务的资源占用。	单个服务过载导致的全局资源耗尽问题。
Timeout（超时）	限制调用的最大执行时间，避免长时间阻塞占用资源。	服务响应缓慢导致的线程/连接耗尽问题。
Cache（缓存）	缓存调用结果，减少重复请求，提高响应速度。	高频重复请求导致的后端服务压力过大问题。

每个模块通过接口定义功能边界（如CircuitBreaker接口、RateLimiter接口），并提供默认实现，开发者可通过接口替换实现类以扩展功能。

2.3.2 配置层

配置层负责管理各模块的配置参数，支持多种配置方式，确保模块行为可按需定制：

代码配置：通过XXXConfig.custom()Builder模式动态创建配置；

配置文件：支持YAML/Properties文件配置（尤其适合Spring Boot环境）；

动态配置：集成Spring Cloud Config、Apollo等配置中心，实现运行时参数更新。

配置层的核心是Registry（注册表），如CircuitBreakerRegistry、RateLimiterRegistry，用于管理模块实例及其配置。通过注册表，开发者可集中创建、获取和销毁模块实例，确保配置的一致性。

2.3.3 事件监听与监控层

事件监听与监控层负责收集模块运行过程中的事件和指标，支撑系统的可观测性：

事件监听：通过EventPublisher发布模块事件（如CircuitBreakerOnOpenEvent），开发者可注册EventConsumer处理事件；

指标收集：集成Micrometer、Prometheus等工具，暴露关键指标（如断路器失败率、限流次数）；

日志输出：默认输出事件日志，便于问题排查。

该层使开发者能实时掌握系统的容错状态，为故障诊断和性能优化提供数据支持。

2.3.4 集成层

集成层负责将Resilience4j与主流框架和技术栈集成，降低使用门槛：

Spring生态：提供resilience4j-spring-boot2 Starter，支持注解式使用（如@CircuitBreaker、@RateLimiter）；

响应式框架：支持RxJava、Project Reactor，提供Flowable、Mono等响应式类型的装饰器；

监控工具：集成Micrometer、Prometheus、Grafana，提供预置的监控仪表盘。

集成层的设计使Resilience4j能无缝融入现有技术体系，无需大幅改造项目架构。

三、Resilience4j核心模块详解

Resilience4j的核心能力通过六大模块实现，每个模块都有明确的应用场景和实现原理。本节将深入剖析各模块的工作机制、配置参数、使用场景及源码核心逻辑。

3.1 断路器（CircuitBreaker）：防止故障扩散的“安全开关”

断路器是Resilience4j最核心的模块，其设计灵感源于电路断路器：当电路过载时自动断开，保护电路；故障排除后手动或自动闭合。在分布式系统中，断路器用于监控服务调用结果，当失败率过高时“断开”调用，避免故障服务消耗更多资源。

3.1.1 核心原理：有限状态机模型

CircuitBreaker基于有限状态机（FSM） 设计，包含三种状态，状态转换由请求结果和配置参数共同驱动：

状态	描述	核心行为
关闭（CLOSED）	默认状态，允许所有请求通过；同时记录请求结果（成功/失败）。	用环形缓冲区记录最近请求结果，当失败率超过阈值时切换到“打开”状态。
打开（OPEN）	拒绝所有请求，直接执行降级逻辑；避免故障服务持续消耗资源。	计时等待`waitDurationInOpenState`后，自动切换到“半开”状态。
半开（HALF_OPEN）	允许有限请求通过，检测故障是否恢复；是“打开→关闭”的过渡状态。	若请求失败率≤阈值，切换到“关闭”状态；否则切换回“打开”状态。

状态转换全流程：

初始状态为CLOSED，所有请求正常执行，断路器通过环形缓冲区记录最近请求的结果（成功/失败）。

当缓冲区中失败率≥failureRateThreshold（如50%），且调用次数≥minimumNumberOfCalls（如10次）时，状态从CLOSED→OPEN。

OPEN状态下，所有请求被拒绝，等待waitDurationInOpenState（如15秒）后，自动切换到HALF_OPEN。

HALF_OPEN状态下，允许permittedNumberOfCallsInHalfOpenState（如10次）请求通过：

若这些请求失败率≤阈值，状态→CLOSED，恢复正常请求；

若失败率＞阈值，状态→OPEN，继续拒绝请求。

3.1.2 核心组件：环形缓冲区（Ring Bit Buffer）

环形缓冲区是断路器记录请求结果的核心数据结构，用于计算失败率。它是一个固定大小的循环数组，每个位置存储单个请求的结果（1=失败，0=成功），具有以下优势：

高效存储：仅用二进制记录结果，内存占用极低（如大小为100的缓冲区仅需100位）；

滑动窗口：新请求结果覆盖旧数据，天然实现“滑动窗口”效果，确保统计最近请求；

快速计算：遍历缓冲区即可计算失败率，无需复杂的数据结构。

滑动窗口配置： Resilience4j支持两种滑动窗口类型（通过slidingWindowType配置）：

COUNT_BASED（基于计数）：缓冲区大小slidingWindowSize表示统计最近N次请求（如20次）；

TIME_BASED（基于时间）：缓冲区按时间分片，slidingWindowSize表示分片数（如10个分片×1秒=10秒窗口）。

3.1.3 核心配置参数

CircuitBreaker的行为完全由配置参数控制，以下是关键参数的详细说明：

参数名	类型	默认值	说明
`failureRateThreshold`	int	50	触发“关闭→打开”的失败率阈值（百分比）。例如50表示失败率≥50%时打开。
`minimumNumberOfCalls`	int	10	计算失败率前的最小调用次数（避免少量请求导致误判）。
`waitDurationInOpenState`	Duration	60s	“打开”状态持续时间，超时后自动切换到“半开”。
`permittedNumberOfCallsInHalfOpenState`	int	10	“半开”状态允许的最大请求数（用于检测故障是否恢复）。
`slidingWindowSize`	int	100	滑动窗口大小（基于计数时为请求数；基于时间时为分片数）。
`slidingWindowType`	enum	COUNT_BASED	滑动窗口类型：`COUNT_BASED`（计数）或`TIME_BASED`（时间）。
`recordExceptions`	Class[]	空	需要记录为“失败”的异常类型（如`IOException`）。
`ignoreExceptions`	Class[]	空	忽略的异常（不视为失败，如业务异常`IllegalArgumentException`）。

YAML配置示例（Spring Boot环境）：


resilience4j:
  circuitbreaker:
    instances:
      paymentService:  # 针对支付服务的断路器
        failureRateThreshold: 40  # 失败率≥40%时打开
        minimumNumberOfCalls: 15  # 至少15次调用后计算失败率
        waitDurationInOpenState: 30s  # 打开状态持续30秒
        permittedNumberOfCallsInHalfOpenState: 5  # 半开状态允许5次请求
        slidingWindowSize: 30  # 统计最近30次请求
        slidingWindowType: COUNT_BASED  # 基于计数的滑动窗口
        recordExceptions:  # 记录为失败的异常
          - java.net.ConnectException
          - java.util.concurrent.TimeoutException
        ignoreExceptions:  # 忽略的异常
          - java.lang.IllegalArgumentException

3.1.4 源码解析：状态转换逻辑

CircuitBreaker的核心逻辑在DefaultCircuitBreaker类中，以下是状态转换的关键源码解析（添加中文注释）：

3.1.4.1 关闭状态下的失败率计算与状态切换


// DefaultCircuitBreaker.java
private void onFailure(long duration, Exception exception) {
    // 记录失败到环形缓冲区（1表示失败）
    ringBuffer.recordFailure();
    // 发布失败事件（供监控和告警）
    eventPublisher.onFailure(createFailureEvent(duration, exception));

    // 若当前为关闭状态，检查是否需要切换到打开状态
    if (stateMachine.getState() == State.CLOSED && shouldTransitionToOpenState()) {
        transitionToOpenState();  // 切换到打开状态
    }
}

// 判断是否需要从关闭→打开
private boolean shouldTransitionToOpenState() {
    // 1. 检查是否达到最小调用次数（避免少量请求导致误判）
    if (ringBuffer.size() < config.getMinimumNumberOfCalls()) {
        return false;
    }
    // 2. 计算环形缓冲区中的失败率
    float failureRate = ringBuffer.calculateFailureRate();
    // 3. 失败率≥阈值则返回true
    return failureRate >= config.getFailureRateThreshold();
}

3.1.4.2 环形缓冲区的失败率计算


// RingBitBuffer.java
public float calculateFailureRate() {
    int failureCount = 0;  // 失败次数
    int totalCount = 0;    // 总请求次数（已记录的）
    // 遍历缓冲区，统计失败次数和总次数
    for (int i = 0; i < buffer.length; i++) {
        if (buffer[i] == 1) {  // 1表示失败
            failureCount++;
        }
        if (buffer[i] != -1) {  // -1表示未使用的位置
            totalCount++;
        }
    }
    // 避免除零异常
    if (totalCount == 0) {
        return 0.0f;
    }
    // 计算失败率（百分比）
    return (float) failureCount / totalCount * 100;
}

3.1.4.3 打开状态到半开状态的过渡


// DefaultCircuitBreaker.java
private void checkState() {
    State currentState = stateMachine.getState();
    if (currentState == State.OPEN) {
        // 计算打开状态已持续的时间
        long elapsedNanosSinceOpenState = System.nanoTime() - stateTransitionTimestamp;
        // 若持续时间≥等待时间，切换到半开状态
        if (elapsedNanosSinceOpenState >= config.getWaitDurationInOpenState().toNanos()) {
            transitionToHalfOpenState();
        } else {
            // 未到时间，抛出异常拒绝请求
            throw new CallNotPermittedException(
                "CircuitBreaker '" + name + "' is OPEN and does not permit further calls");
        }
    }
}

3.1.4.4 半开状态的结果判断


// DefaultCircuitBreaker.java
private void onSuccess(long duration) {
    // 若当前为半开状态，成功后切换到关闭状态
    if (stateMachine.getState() == State.HALF_OPEN) {
        transitionToClosedState();  // 切换到关闭状态
    }
    // 记录成功到缓冲区（0表示成功）
    ringBuffer.recordSuccess();
    // 发布成功事件
    eventPublisher.onSuccess(createSuccessEvent(duration));
}

private void onFailure(long duration, Exception exception) {
    // 若当前为半开状态，失败后切换到打开状态
    if (stateMachine.getState() == State.HALF_OPEN) {
        transitionToOpenState();  // 切换到打开状态
        return;
    }
    // 其他逻辑（关闭状态下的处理）
}

3.1.5 使用场景与最佳实践

CircuitBreaker适用于所有依赖外部服务（如数据库、第三方API、微服务）的场景，以下是最佳实践：

3.1.5.1 差异化配置核心与非核心服务

核心服务（如支付、订单）：设置较低的failureRateThreshold（如30%）和较短的waitDurationInOpenState（如15秒），确保快速失败和恢复；

非核心服务（如日志、统计）：可设置较高的failureRateThreshold（如60%），允许更多容错空间。

3.1.5.2 合理设计降级逻辑

断路器打开时，需通过降级逻辑（Fallback）保证用户体验：


@Service
public class OrderService {
    private final PaymentService paymentService;

    // 用注解声明断路器和降级方法
    @CircuitBreaker(name = "paymentService", fallbackMethod = "processPaymentFallback")
    public PaymentResult processPayment(PaymentRequest request) {
        return paymentService.pay(request);  // 调用外部支付服务
    }

    // 降级方法：参数需与原方法一致，最后添加Exception参数
    public PaymentResult processPaymentFallback(PaymentRequest request, Exception e) {
        log.warn("Payment service fallback triggered: {}", e.getMessage());
        // 降级策略：返回缓存结果、使用备用服务或提示用户稍后重试
        return new PaymentResult(false, "Payment temporarily unavailable, please try again later");
    }
}

3.1.5.3 结合重试机制使用

对于暂时性故障（如网络抖动），可在断路器外层包裹重试，减少误判：


// 先重试，失败次数达标后再触发断路器
@CircuitBreaker(name = "paymentService", fallbackMethod = "processPaymentFallback")
@Retry(name = "paymentServiceRetry")
public PaymentResult processPayment(PaymentRequest request) {
    return paymentService.pay(request);
}

3.1.5.4 监控与告警

通过事件监听和指标监控断路器状态，及时发现故障：


// 订阅断路器打开事件，发送告警
CircuitBreaker circuitBreaker = CircuitBreakerRegistry.ofDefaults().circuitBreaker("paymentService");
circuitBreaker.getEventPublisher()
    .onOpen(event -> {
        log.error("Circuit breaker {} opened at {}: failure rate exceeded threshold",
            event.getCircuitBreakerName(), event.getCreationTime());
        alertService.sendAlert("Circuit Breaker Open: " + event.getCircuitBreakerName());
    });

3.2 限流器（RateLimiter）：控制流量的“阀门”

限流器用于控制单位时间内的请求数量，防止系统因流量过大而过载。Resilience4j的RateLimiter基于令牌桶算法实现，支持灵活的速率配置和超时等待机制。

3.2.1 核心原理：令牌桶算法

令牌桶算法的核心思想是：系统以固定速率生成令牌并放入桶中，每个请求需获取一个令牌才能执行；若桶中无令牌，请求被限流（等待或拒绝）。Resilience4j的RateLimiter通过以下机制实现：

令牌生成：每隔limitRefreshPeriod（如10秒）生成limitForPeriod（如100个）令牌，桶中令牌数不超过limitForPeriod；

令牌获取：每个请求调用acquirePermission()获取令牌，若有令牌则消耗1个并执行；若无令牌且设置了超时，则等待令牌生成，超时后被限流；

并发控制：通过原子变量storedTokens和lastRefillTime记录当前令牌数和上次生成时间，确保并发安全。

3.2.2 核心配置参数

参数名	类型	默认值	说明
`limitForPeriod`	int	50	每个刷新周期内生成的令牌总数（即单位时间内的最大请求数）。
`limitRefreshPeriod`	Duration	500ms	令牌刷新周期（令牌生成的时间间隔）。
`timeoutDuration`	Duration	500ms	请求获取令牌的最大等待时间，超时未获取则被限流。

参数关系示例：

若limitForPeriod=100、limitRefreshPeriod=10s，则系统允许的最大QPS为10（100令牌/10秒）；

若limitForPeriod=10、limitRefreshPeriod=1s，则最大QPS为10。

YAML配置示例：


resilience4j:
  ratelimiter:
    instances:
      orderApi:  # 订单接口限流器
        limitForPeriod: 200  # 每10秒生成200个令牌
        limitRefreshPeriod: 10s  # 刷新周期10秒
        timeoutDuration: 1s  # 请求最多等待1秒获取令牌

3.2.3 源码解析：令牌桶核心逻辑

RateLimiter的核心逻辑在DefaultRateLimiter类中，以下是令牌生成和获取的关键源码（添加中文注释）：

3.2.3.1 令牌获取入口


// DefaultRateLimiter.java
@Override
public boolean acquirePermission() {
    // 无超时获取令牌（立即返回）
    return acquirePermission(0, TimeUnit.NANOSECONDS);
}

@Override
public boolean acquirePermission(long timeoutDuration, TimeUnit timeUnit) {
    long timeoutNanos = timeUnit.toNanos(timeoutDuration);
    long startNanos = System.nanoTime();

    while (true) {
        // 1. 计算当前可用令牌数（生成新令牌并更新）
        long nowNanos = System.nanoTime();
        long newTokens = calculateNewTokens(nowNanos);  // 计算新生成的令牌
        long currentTokens = Math.min(config.getLimitForPeriod(), storedTokens.get() + newTokens);
        storedTokens.set(currentTokens);  // 更新令牌数
        lastRefillTime.set(nowNanos);     // 更新上次生成时间

        // 2. 若有可用令牌，消耗1个并返回成功
        if (currentTokens > 0) {
            storedTokens.decrementAndGet();  // 消耗令牌
            return true;
        }

        // 3. 无令牌，检查是否需要等待
        long waitTimeNanos = calculateWaitTimeNanos(nowNanos);  // 计算需等待的时间
        if (timeoutNanos <= 0) {  // 无超时设置，直接返回失败
            return false;
        }
        if (waitTimeNanos > timeoutNanos) {  // 等待时间超过超时时间，返回失败
            return false;
        }

        // 4. 等待令牌生成
        try {
            // 等待剩余时间（毫秒+纳秒）
            Thread.sleep(waitTimeNanos / 1_000_000, (int) (waitTimeNanos % 1_000_000));
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();  // 恢复中断状态
            return false;
        }

        // 5. 更新剩余超时时间，继续循环尝试获取令牌
        timeoutNanos -= (System.nanoTime() - startNanos);
        startNanos = System.nanoTime();
    }
}

3.2.3.2 新令牌计算逻辑


// DefaultRateLimiter.java
private long calculateNewTokens(long nowNanos) {
    long lastRefillNanos = lastRefillTime.get();
    // 若当前时间≤上次生成时间，无需生成新令牌（可能是时钟回拨）
    if (nowNanos <= lastRefillNanos) {
        return 0;
    }

    // 计算时间差（当前时间 - 上次生成时间）
    long nanosSinceLastRefill = nowNanos - lastRefillNanos;
    // 计算刷新周期的纳秒数
    long refreshPeriodNanos = config.getLimitRefreshPeriod().toNanos();
    // 时间差内包含的完整刷新周期数
    long completedPeriods = nanosSinceLastRefill / refreshPeriodNanos;

    // 每个周期生成的令牌数×周期数 = 新生成的令牌数
    return completedPeriods * config.getLimitForPeriod();
}

3.2.3.3 等待时间计算


// DefaultRateLimiter.java
private long calculateWaitTimeNanos(long nowNanos) {
    long lastRefillNanos = lastRefillTime.get();
    // 刷新周期的纳秒数
    long refreshPeriodNanos = config.getLimitRefreshPeriod().toNanos();
    // 距离下次令牌生成的剩余时间 = 刷新周期 - （当前时间 - 上次生成时间）% 刷新周期
    long nanosUntilNextRefill = refreshPeriodNanos - ((nowNanos - lastRefillNanos) % refreshPeriodNanos);
    return nanosUntilNextRefill;
}

3.2.4 使用场景与最佳实践

RateLimiter适用于控制接口或服务的访问速率，以下是典型场景和实践建议：

3.2.4.1 保护第三方API调用

当调用第三方API（如支付接口、短信服务）时，通常有QPS限制，可用限流器确保不超过阈值：


@Service
public class PaymentGatewayService {
    // 注入限流器（通过Spring自动配置）
    private final RateLimiter rateLimiter;

    public PaymentGatewayService(RateLimiterRegistry rateLimiterRegistry) {
        this.rateLimiter = rateLimiterRegistry.rateLimiter("thirdPartyPayment");
    }

    public PaymentResponse callThirdPartyApi(PaymentRequest request) {
        // 尝试获取令牌，超时1秒
        boolean permission = rateLimiter.acquirePermission(1, TimeUnit.SECONDS);
        if (!permission) {
            // 限流处理：返回降级响应或记录排队
            return new PaymentResponse(false, "Too many requests, please try again later");
        }
        // 调用第三方API
        return thirdPartyPaymentClient.pay(request);
    }
}

3.2.4.2 结合Spring注解使用

在Spring Boot中，通过@RateLimiter注解可简化使用：


@RestController
@RequestMapping("/orders")
public class OrderController {
    // 对接口添加限流，使用"orderApi"配置
    @RateLimiter(name = "orderApi", fallbackMethod = "createOrderFallback")
    @PostMapping
    public ResponseEntity<OrderDTO> createOrder(@RequestBody OrderRequest request) {
        OrderDTO order = orderService.createOrder(request);
        return ResponseEntity.ok(order);
    }

    // 限流降级方法
    public ResponseEntity<OrderDTO> createOrderFallback(OrderRequest request, Exception e) {
        return ResponseEntity.status(429).body(new OrderDTO("limited", "Too many requests"));
    }
}

3.2.4.3 动态调整限流阈值

结合配置中心（如Apollo）可实现限流阈值的动态调整：


// 监听配置变化，动态更新限流器参数
@Configuration
public class RateLimiterDynamicConfig {
    @Autowired
    private RateLimiterRegistry rateLimiterRegistry;

    @EventListener
    public void onConfigChange(ConfigChangeEvent event) {
        ConfigChange limitChange = event.getChange("resilience4j.ratelimiter.instances.orderApi.limitForPeriod");
        if (limitChange != null) {
            int newLimit = Integer.parseInt(limitChange.getNewValue());
            // 更新限流器配置
            RateLimiter rateLimiter = rateLimiterRegistry.rateLimiter("orderApi");
            rateLimiter.changeConfig(config -> config.limitForPeriod(newLimit));
        }
    }
}

3.3 重试（Retry）：解决暂时性故障的“二次尝试”

重试机制用于在调用失败时自动重新尝试，适用于处理暂时性故障（如网络抖动、服务临时不可用）。Resilience4j的Retry支持灵活的重试策略（固定间隔、指数退避）和异常过滤。

3.3.1 核心原理

Retry通过拦截方法调用，在捕获到指定异常时按配置策略重试，核心流程如下：

执行原始业务逻辑；

若成功，返回结果；

若失败，检查异常是否符合重试条件（retryExceptions）；

若符合条件且未达最大重试次数，等待waitDuration后重试；

若重试成功，返回结果；若重试次数耗尽仍失败，抛出异常或执行降级。

Retry支持两种重试间隔策略：

固定间隔：每次重试间隔相同（waitDuration）；

指数退避：重试间隔按指数增长（如1s→2s→4s），避免集中重试加剧系统压力。

3.3.2 核心配置参数

参数名	类型	默认值	说明
`maxAttempts`	int	3	最大尝试次数（包括首次调用），即重试次数=maxAttempts-1。
`waitDuration`	Duration	500ms	重试间隔时间（固定间隔策略）。
`retryExceptions`	Class[]	空	需要触发重试的异常类型（如`IOException`）。
`ignoreExceptions`	Class[]	空	忽略的异常（不触发重试，如`IllegalArgumentException`）。
`backoffType`	enum	FIXED	退避策略：`FIXED`（固定间隔）或`EXPONENTIAL`（指数退避）。
`exponentialBackoffMultiplier`	float	2.0f	指数退避的乘数（间隔=waitDuration×multiplier^重试次数）。

YAML配置示例：


resilience4j:
  retry:
    instances:
      inventoryService:  # 库存服务重试配置
        maxAttempts: 4  # 最大尝试4次（即重试3次）
        waitDuration: 1s  # 初始间隔1秒
        backoffType: EXPONENTIAL  # 指数退避策略
        exponentialBackoffMultiplier: 2  # 乘数2（1s→2s→4s）
        retryExceptions:  # 触发重试的异常
          - java.net.ConnectException
          - java.util.concurrent.TimeoutException
        ignoreExceptions:  # 忽略的异常
          - java.lang.IllegalStateException

3.3.3 源码解析：重试逻辑

Retry的核心逻辑在DefaultRetry类中，以下是重试流程的关键源码（添加中文注释）：


// DefaultRetry.java
@Override
public <T> T executeSupplier(Supplier<T> supplier) {
    int attempt = 0;
    Throwable lastThrowable;

    while (true) {
        attempt++;  // 尝试次数+1
        try {
            // 执行原始业务逻辑
            T result = supplier.get();
            // 发布重试成功事件（若已重试过）
            if (attempt > 1) {
                eventPublisher.onSuccess(createSuccessEvent(attempt));
            }
            return result;
        } catch (Exception e) {
            lastThrowable = e;
            // 检查异常是否需要忽略（不重试）
            if (isIgnoreException(e)) {
                throw e;
            }
            // 检查异常是否符合重试条件
            if (!isRetryException(e)) {
                throw e;
            }
            // 检查是否已达最大尝试次数
            if (attempt >= config.getMaxAttempts()) {
                // 发布重试耗尽事件
                eventPublisher.onError(createErrorEvent(attempt, e));
                throw e;
            }
            // 计算重试间隔时间
            long waitDuration = calculateWaitDuration(attempt);
            // 发布重试事件
            eventPublisher.onRetry(createRetryEvent(attempt, waitDuration, e));
            // 等待重试间隔
            sleep(waitDuration);
        }
    }
}

// 计算重试间隔（支持指数退避）
private long calculateWaitDuration(int attempt) {
    if (config.getBackoffType() == BackoffType.EXPONENTIAL) {
        // 指数退避：waitDuration × multiplier^(attempt-1)
        int exponent = attempt - 1;
        float multiplier = config.getExponentialBackoffMultiplier();
        return (long) (config.getWaitDuration().toMillis() * Math.pow(multiplier, exponent));
    } else {
        // 固定间隔：直接返回waitDuration
        return config.getWaitDuration().toMillis();
    }
}

3.3.4 使用场景与最佳实践

Retry适用于可能出现暂时性故障的场景，以下是实践建议：

3.3.4.1 数据库连接重试

数据库连接可能因临时网络问题失败，重试可提高成功率：


@Repository
public class OrderRepository {
    private final JdbcTemplate jdbcTemplate;
    private final Retry retry;

    public OrderRepository(JdbcTemplate jdbcTemplate, RetryRegistry retryRegistry) {
        this.jdbcTemplate = jdbcTemplate;
        this.retry = retryRegistry.retry("dbConnection");
    }

    public void saveOrder(Order order) {
        // 用重试装饰数据库操作
        Retry.decorateRunnable(retry, () -> {
            jdbcTemplate.update("INSERT INTO orders (...) VALUES (...)",
                order.getId(), order.getAmount());
        }).run();
    }
}

3.3.4.2 避免盲目重试

幂等性保证：重试的操作必须是幂等的（如GET请求、带唯一ID的POST请求），避免重复创建订单、重复扣款等问题；

合理设置间隔：使用指数退避策略（EXPONENTIAL），避免短时间内大量重试加剧系统压力；

明确重试异常：仅对暂时性异常（如TimeoutException、ConnectException）重试，对业务异常（如“余额不足”）不重试。

3.4 舱壁（Bulkhead）：资源隔离的“防火墙”

舱壁模式源于船舶设计：将船舱划分为多个独立舱室，某一舱室进水时不会淹没整艘船。在分布式系统中，舱壁用于隔离不同服务的资源（线程/并发数），防止单个服务过载影响全局。

3.4.1 核心原理

Resilience4j提供两种舱壁实现：

信号量舱壁（SEMAPHORE）：通过信号量限制并发请求数量，适用于阻塞式调用；

线程池舱壁（THREAD_POOL）：为服务分配独立线程池，通过线程池参数限制资源使用，适用于异步调用。

信号量舱壁的核心逻辑：

维护一个信号量计数器，初始值为maxConcurrentCalls；

每个请求获取信号量（计数器-1），完成后释放（计数器+1）；

若计数器=0，新请求等待maxWaitDuration，超时未获取则被拒绝。

线程池舱壁的核心逻辑：

为服务创建独立线程池（corePoolSize、maxPoolSize、queueCapacity）；

请求提交到线程池执行，若线程池满（线程数达maxPoolSize且队列满），新请求被拒绝。

3.4.2 核心配置参数

参数名	类型	默认值	说明（信号量舱壁）
`type`	enum	SEMAPHORE	舱壁类型：`SEMAPHORE`（信号量）或`THREAD_POOL`（线程池）。
`maxConcurrentCalls`	int	25	最大并发请求数（信号量舱壁）。
`maxWaitDuration`	Duration	0ms	请求等待获取信号量的最大时间，超时则被拒绝。

线程池舱壁额外参数：

参数名	类型	默认值	说明
`corePoolSize`	int	10	核心线程池大小。
`maxPoolSize`	int	10	最大线程池大小。
`queueCapacity`	int	100	任务队列容量。
`keepAliveDuration`	Duration	20ms	空闲线程存活时间。

YAML配置示例：


resilience4j:
  bulkhead:
    instances:
      paymentService:  # 支付服务用线程池舱壁
        type: THREAD_POOL
        corePoolSize: 5
        maxPoolSize: 10
        queueCapacity: 20
        keepAliveDuration: 30s
      inventoryService:  # 库存服务用信号量舱壁
        type: SEMAPHORE
        maxConcurrentCalls: 15
        maxWaitDuration: 500ms

3.4.3 源码解析：信号量舱壁

信号量舱壁的核心逻辑在SemaphoreBulkhead类中：


// SemaphoreBulkhead.java
public class SemaphoreBulkhead implements Bulkhead {
    private final Semaphore semaphore;  // Java原生信号量
    private final BulkheadConfig config;
    private final String name;

    public SemaphoreBulkhead(String name, BulkheadConfig config) {
        this.name = name;
        this.config = config;
        // 初始化信号量， permits = maxConcurrentCalls
        this.semaphore = new Semaphore(config.getMaxConcurrentCalls(), true);
    }

    @Override
    public <T> T executeSupplier(Supplier<T> supplier) {
        try {
            // 尝试获取信号量，超时时间为maxWaitDuration
            boolean acquired = semaphore.tryAcquire(config.getMaxWaitDuration().toMillis(), TimeUnit.MILLISECONDS);
            if (!acquired) {
                // 未获取到信号量，发布拒绝事件并抛出异常
                eventPublisher.onCallRejected(createCallRejectedEvent());
                throw new BulkheadFullException("Bulkhead '" + name + "' is full");
            }
            // 发布允许事件
            eventPublisher.onCallPermitted(createCallPermittedEvent());
            // 执行业务逻辑
            return supplier.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BulkheadException(e);
        } finally {
            // 释放信号量
            semaphore.release();
        }
    }
}

3.4.4 使用场景与最佳实践

舱壁模式适用于隔离不同服务的资源，以下是典型场景：

3.4.4.1 核心服务资源隔离

为核心服务（如支付）分配独立线程池，确保其资源不被非核心服务占用：


@Service
public class PaymentService {
    private final Bulkhead bulkhead;
    private final PaymentClient paymentClient;

    public PaymentService(BulkheadRegistry bulkheadRegistry, PaymentClient paymentClient) {
        this.bulkhead = bulkheadRegistry.bulkhead("paymentService");  // 线程池舱壁
        this.paymentClient = paymentClient;
    }

    public CompletableFuture<PaymentResult> processPaymentAsync(PaymentRequest request) {
        // 用舱壁装饰异步任务
        return Bulkhead.decorateSupplier(bulkhead, () -> paymentClient.pay(request))
            .toCompletableFuture()
            .exceptionally(ex -> {
                // 舱壁满时的降级处理
                if (ex instanceof BulkheadFullException) {
                    return new PaymentResult(false, "System busy, please try again later");
                }
                throw new RuntimeException(ex);
            });
    }
}

3.4.4.2 选择合适的舱壁类型

信号量舱壁：适用于轻量级、短耗时的同步调用（如数据库查询），开销低；

线程池舱壁：适用于重量级、长耗时的调用（如第三方API），隔离性更强，但线程切换开销高。

3.5 超时（Timeout）：避免资源长期占用的“定时器”

超时机制用于限制调用的最大执行时间，防止因服务响应缓慢导致线程/连接长期占用。Resilience4j的Timeout模块支持同步和异步调用的超时控制。

3.5.1 核心原理

Timeout通过以下机制实现超时控制：

同步调用：启动一个计时器，若方法执行时间超过timeoutDuration，则中断线程并抛出TimeoutException；

异步调用：基于CompletableFuture，若未来结果在超时时间内未完成，则触发超时处理。

3.5.2 核心配置参数

参数名	类型	默认值	说明
`timeoutDuration`	Duration	1s	最大执行时间，超过则触发超时。

YAML配置示例：


resilience4j:
  timelimiter:
    instances:
      reportGeneration:  # 报表生成超时配置
        timeoutDuration: 30s  # 报表生成最多允许30秒

3.5.3 使用场景

Timeout适用于所有可能出现响应延迟的场景，如：

复杂的数据库查询（设置较长超时，如10秒）；

第三方API调用（根据API SLA设置超时，如5秒）；

大数据处理任务（如报表生成，设置较长超时）。

使用示例：


@Service
public class ReportService {
    private final TimeLimiter timeLimiter;
    private final ReportGenerator generator;

    public ReportService(TimeLimiterRegistry timeLimiterRegistry, ReportGenerator generator) {
        this.timeLimiter = timeLimiterRegistry.timeLimiter("reportGeneration");
        this.generator = generator;
    }

    public Report generateReport(ReportRequest request) {
        // 用超时装饰同步任务
        return TimeLimiter.decorateSupplier(timeLimiter, () -> generator.generate(request))
            .get();  // 若超时，会抛出TimeoutException
    }
}

3.6 缓存（Cache）：减少重复请求的“加速器”

缓存模块用于存储高频请求的结果，减少对后端服务的重复调用，提高响应速度。Resilience4j的Cache支持与Caffeine、Guava等主流缓存框架集成。

3.6.1 核心原理

Cache的核心逻辑是“先查缓存，未命中则执行方法并缓存结果”：

调用方法前，通过key查询缓存；

若缓存命中，直接返回缓存值；

若缓存未命中，执行方法，将结果存入缓存后返回。

3.6.2 核心配置参数

参数名	类型	默认值	说明
`cacheProvider`	String	CAFFEINE	缓存框架：`CAFFEINE`或`GUAVA`。
`maxSize`	int	1000	缓存最大容量。
`expirationDuration`	Duration	5m	缓存条目过期时间。

YAML配置示例：


resilience4j:
  cache:
    instances:
      productCache:  # 商品缓存
        cacheProvider: CAFFEINE
        maxSize: 5000  # 最多缓存5000个商品
        expirationDuration: 10m  # 10分钟过期

3.6.3 使用场景

Cache适用于高频调用且结果相对稳定的场景，如：

商品详情查询、用户信息查询；

静态配置数据（如字典表、权限配置）；

计算密集型但结果稳定的任务（如报表统计结果）。

使用示例：


@Service
public class ProductService {
    private final Cache<String, ProductDTO> productCache;
    private final ProductRepository repository;

    public ProductService(CacheRegistry cacheRegistry, ProductRepository repository) {
        this.productCache = cacheRegistry.cache("productCache");  // 获取缓存实例
        this.repository = repository;
    }

    public ProductDTO getProduct(String productId) {
        // 先查缓存，未命中则从数据库查询并缓存
        return productCache.computeIfAbsent(productId, id -> repository.findById(id)
            .orElseThrow(() -> new ProductNotFoundException(id)));
    }
}

3.7 事件（Event）：监控与扩展的“消息总线”

事件模块是Resilience4j的“神经中枢”，各模块在状态变化或关键操作时发布事件，开发者可通过订阅事件实现监控、告警和自定义扩展。

3.7.1 核心事件类型

各模块发布的事件类型如下：

模块	核心事件类型	触发时机
CircuitBreaker	`CircuitBreakerOnOpenEvent`、`OnCloseEvent`等	状态切换（打开/关闭/半开）、请求成功/失败。
RateLimiter	`RateLimiterOnSuccessEvent`、`OnFailureEvent`	获取令牌成功/失败（被限流）。
Retry	`RetryOnRetryEvent`、`OnSuccessEvent`等	开始重试、重试成功/失败。
Bulkhead	`BulkheadOnCallPermittedEvent`、`OnRejectedEvent`	请求被允许/拒绝。

3.7.2 事件订阅与处理

通过EventPublisher订阅事件，示例如下：


// 订阅断路器事件
CircuitBreaker circuitBreaker = CircuitBreaker.of("paymentService");
circuitBreaker.getEventPublisher()
    .onOpen(event -> log.error("Circuit breaker {} opened", event.getCircuitBreakerName()))
    .onFailure(event -> log.error("Request failed: {}", event.getThrowable().getMessage()));

// 订阅限流器事件
RateLimiter rateLimiter = RateLimiter.of("orderApi");
rateLimiter.getEventPublisher()
    .onFailure(event -> log.warn("Rate limited: {}", event.getRateLimiterName()));

3.7.3 事件持久化与分析

可将事件存储到数据库或消息队列，用于后续分析：


// 将断路器事件存储到数据库
public class DatabaseEventConsumer implements EventConsumer<CircuitBreakerEvent> {
    private final JdbcTemplate jdbcTemplate;

    @Override
    public void accept(CircuitBreakerEvent event) {
        jdbcTemplate.update(
            "INSERT INTO circuit_breaker_events (name, type, timestamp, details) VALUES (?, ?, ?, ?)",
            event.getCircuitBreakerName(),
            event.getClass().getSimpleName(),
            event.getCreationTime(),
            event.toString()
        );
    }
}

// 注册事件消费者
circuitBreaker.getEventPublisher().onEvent(new DatabaseEventConsumer(jdbcTemplate));

四、Resilience4j在项目中的模块化使用

Resilience4j的模块化设计使其能灵活集成到各类项目中，本节将详细介绍在Spring Boot和非Spring环境中的使用方法，以及多模块组合的最佳实践。

4.1 Spring Boot环境集成

Spring Boot提供了对Resilience4j的自动配置支持，通过引入Starter依赖和简单配置即可快速使用。

4.1.1 依赖引入

在pom.xml中添加所需模块的Starter依赖：


<!-- 核心依赖 -->
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot2</artifactId>
    <version>1.7.1</version>
</dependency>
<!-- 若需使用注解式编程 -->
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-annotations</artifactId>
    <version>1.7.1</version>
</dependency>
<!-- 若需监控集成 -->
<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-micrometer</artifactId>
    <version>1.7.1</version>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
    <version>1.8.4</version>
</dependency>

4.1.2 配置文件设置

在application.yml中配置各模块的参数：


# 应用名称
spring:
  application:
    name: order-service

# Resilience4j配置
resilience4j:
  # 断路器配置
  circuitbreaker:
    instances:
      paymentService:
        failureRateThreshold: 50
        waitDurationInOpenState: 10s
        slidingWindowSize: 20
  # 限流器配置
  ratelimiter:
    instances:
      orderApi:
        limitForPeriod: 100
        limitRefreshPeriod: 10s
  # 重试配置
  retry:
    instances:
      inventoryService:
        maxAttempts: 3
        waitDuration: 1s
  # 舱壁配置
  bulkhead:
    instances:
      shippingService:
        type: SEMAPHORE
        maxConcurrentCalls: 15
  # 超时配置
  timelimiter:
    instances:
      reportService:
        timeoutDuration: 30s

# 监控端点配置（暴露Prometheus指标）
management:
  endpoints:
    web:
      exposure:
        include: health,info,prometheus
  metrics:
    export:
      prometheus:
        enabled: true

4.1.3 注解式使用

通过注解在业务方法上启用容错机制，示例如下：


@Service
public class OrderService {
    private final PaymentService paymentService;
    private final InventoryService inventoryService;

    // 构造器注入依赖
    public OrderService(PaymentService paymentService, InventoryService inventoryService) {
        this.paymentService = paymentService;
        this.inventoryService = inventoryService;
    }

    /**
     * 创建订单：组合限流、熔断、重试机制
     */
    @RateLimiter(name = "orderApi", fallbackMethod = "createOrderFallback")  // 限流
    @CircuitBreaker(name = "paymentService", fallbackMethod = "createOrderFallback")  // 熔断
    @Retry(name = "inventoryService")  // 重试
    public OrderDTO createOrder(OrderRequest request) {
        // 1. 检查库存
        boolean hasStock = inventoryService.checkStock(request.getProductId(), request.getQuantity());
        if (!hasStock) {
            throw new InsufficientStockException("Product " + request.getProductId() + " is out of stock");
        }
        // 2. 处理支付
        PaymentResult paymentResult = paymentService.processPayment(
            new PaymentRequest(request.getUserId(), request.getAmount()));
        if (!paymentResult.isSuccess()) {
            throw new PaymentFailedException(paymentResult.getErrorMessage());
        }
        // 3. 创建订单
        return new OrderDTO(
            UUID.randomUUID().toString(),
            request.getProductId(),
            request.getQuantity(),
            "SUCCESS"
        );
    }

    /**
     * 降级方法：限流、熔断时执行
     */
    public OrderDTO createOrderFallback(OrderRequest request, Exception e) {
        log.warn("Order creation fallback triggered: {}", e.getMessage());
        // 返回降级订单（如排队中、稍后重试）
        return new OrderDTO(
            null,
            request.getProductId(),
            request.getQuantity(),
            "PENDING"
        );
    }
}

4.1.4 依赖注入与手动使用

除注解外，也可通过注入Registry手动创建模块实例：


@Service
public class ShippingService {
    // 注入舱壁注册表
    private final BulkheadRegistry bulkheadRegistry;
    // 注入超时注册表
    private final TimeLimiterRegistry timeLimiterRegistry;
    private final ShippingClient shippingClient;

    public ShippingService(BulkheadRegistry bulkheadRegistry,
                          TimeLimiterRegistry timeLimiterRegistry,
                          ShippingClient shippingClient) {
        this.bulkheadRegistry = bulkheadRegistry;
        this.timeLimiterRegistry = timeLimiterRegistry;
        this.shippingClient = shippingClient;
    }

    public ShippingResult scheduleShipping(ShippingRequest request) {
        // 获取舱壁实例
        Bulkhead bulkhead = bulkheadRegistry.bulkhead("shippingService");
        // 获取超时实例
        TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter("reportService");

        // 组合舱壁和超时机制
        Supplier<ShippingResult> shippingSupplier = () -> shippingClient.schedule(request);
        Supplier<ShippingResult> decoratedSupplier = Bulkhead
            .decorateSupplier(bulkhead, TimeLimiter.decorateSupplier(timeLimiter, shippingSupplier));

        try {
            return decoratedSupplier.get();
        } catch (BulkheadFullException e) {
            return new ShippingResult(false, "Too many shipping requests, please try again later");
        } catch (TimeoutException e) {
            return new ShippingResult(false, "Shipping schedule timed out");
        }
    }
}

4.2 非Spring环境使用（函数式API）

在非Spring环境（如普通Java应用、响应式应用）中，可通过函数式API使用Resilience4j。

4.2.1 基本使用步骤

创建配置：通过XXXConfig.custom()构建模块配置；

创建注册表：通过XXXRegistry.of(config)创建注册表；

获取模块实例：通过注册表获取模块实例；

装饰业务逻辑：通过decorateXXX方法包装业务代码。

4.2.2 示例代码


public class OrderProcessingApp {
    public static void main(String[] args) {
        // 1. 创建断路器配置
        CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
            .failureRateThreshold(50)
            .waitDurationInOpenState(Duration.ofSeconds(10))
            .slidingWindowSize(20)
            .build();
        // 创建断路器注册表
        CircuitBreakerRegistry circuitBreakerRegistry = CircuitBreakerRegistry.of(circuitBreakerConfig);
        // 获取断路器实例
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("paymentService");

        // 2. 创建限流器配置
        RateLimiterConfig rateLimiterConfig = RateLimiterConfig.custom()
            .limitForPeriod(100)
            .limitRefreshPeriod(Duration.ofSeconds(10))
            .build();
        RateLimiterRegistry rateLimiterRegistry = RateLimiterRegistry.of(rateLimiterConfig);
        RateLimiter rateLimiter = rateLimiterRegistry.rateLimiter("orderApi");

        // 3. 业务逻辑
        PaymentService paymentService = new PaymentService();
        OrderRequest request = new OrderRequest("user123", "product456", 2, new BigDecimal("99.99"));

        // 4. 组合断路器和限流器
        Supplier<OrderDTO> orderSupplier = () -> processOrder(paymentService, request);
        Supplier<OrderDTO> decoratedSupplier = CircuitBreaker
            .decorateSupplier(circuitBreaker,
                RateLimiter.decorateSupplier(rateLimiter, orderSupplier));

        // 5. 执行带容错的业务逻辑
        try {
            OrderDTO order = decoratedSupplier.get();
            System.out.println("Order created: " + order.getId());
        } catch (Exception e) {
            System.err.println("Order failed: " + e.getMessage());
        }
    }

    private static OrderDTO processOrder(PaymentService paymentService, OrderRequest request) {
        PaymentResult result = paymentService.processPayment(request.getUserId(), request.getAmount());
        if (!result.isSuccess()) {
            throw new RuntimeException("Payment failed: " + result.getErrorMessage());
        }
        return new OrderDTO(UUID.randomUUID().toString(), request.getProductId(), "SUCCESS");
    }
}

4.2.3 响应式编程集成

Resilience4j支持RxJava和Project Reactor，示例如下（Reactor）：


public class ReactiveOrderService {
    private final CircuitBreaker circuitBreaker;
    private final RateLimiter rateLimiter;
    private final ReactivePaymentClient paymentClient;

    public ReactiveOrderService(CircuitBreaker circuitBreaker,
                               RateLimiter rateLimiter,
                               ReactivePaymentClient paymentClient) {
        this.circuitBreaker = circuitBreaker;
        this.rateLimiter = rateLimiter;
        this.paymentClient = paymentClient;
    }

    public Mono<OrderDTO> createOrder(OrderRequest request) {
        // 响应式业务逻辑
        Mono<PaymentResult> paymentMono = paymentClient.processPayment(
            new PaymentRequest(request.getUserId(), request.getAmount()));

        // 用断路器和限流器装饰响应式流
        return paymentMono
            .transform(CircuitBreakerOperator.of(circuitBreaker))  // 熔断
            .transform(RateLimiterOperator.of(rateLimiter))  // 限流
            .map(paymentResult -> {
                if (!paymentResult.isSuccess()) {
                    throw new PaymentFailedException(paymentResult.getErrorMessage());
                }
                return new OrderDTO(UUID.randomUUID().toString(), request.getProductId(), "SUCCESS");
            })
            .onErrorResume(e -> {
                // 降级处理
                return Mono.just(new OrderDTO(null, request.getProductId(), "PENDING"));
						});
		}
}

4.3 多模块组合最佳实践

在实际项目中，单一容错机制往往无法应对复杂场景，需组合多个模块形成“容错组合拳”。以下是常见的组合方案及最佳实践：

4.3.1 限流 + 熔断：流量入口与故障隔离

组合逻辑：先通过限流器控制请求速率，防止系统过载；对通过限流的请求，用断路器监控后端服务状态，故障时快速熔断。

适用场景：外部接口（如API网关）调用后端服务。

实现示例：


@RateLimiter(name = "apiGatewayLimiter")  // 先限流
@CircuitBreaker(name = "backendServiceBreaker", fallbackMethod = "apiFallback")  // 再熔断
public ApiResponse callBackendService(ApiRequest request) {
    return backendService.handleRequest(request);
}

public ApiResponse apiFallback(ApiRequest request, Exception e) {
    return new ApiResponse("error", "Service temporarily unavailable");
}

4.3.2 重试 + 超时：应对暂时性故障与响应延迟

组合逻辑：为可能出现暂时性故障的操作添加重试机制，并设置超时时间，避免重试过程中长时间阻塞。

适用场景：数据库查询、消息发送等。

实现示例：


@Retry(name = "dbRetry")  // 重试暂时性故障
@TimeLimiter(name = "dbTimeout")  // 超时控制
public CompletableFuture<DataResult> queryData(String query) {
    return CompletableFuture.supplyAsync(() -> dbClient.query(query));
}

4.3.3 舱壁 + 限流：资源隔离与流量控制

组合逻辑：通过舱壁限制并发请求数量，防止单个服务占用过多资源；结合限流器控制单位时间请求数，双重保护系统。

适用场景：核心服务（如支付、订单）的资源隔离。

实现示例：


@Bulkhead(name = "paymentBulkhead")  // 限制并发数
@RateLimiter(name = "paymentRateLimiter")  // 控制请求速率
public PaymentResult processPayment(PaymentRequest request) {
    return paymentGateway.pay(request);
}

4.3.4 缓存 + 限流：减轻后端压力与流量控制

组合逻辑：优先查询缓存，减少后端服务调用；对缓存未命中的请求，通过限流器控制访问后端的速率。

适用场景：高频读场景（如商品详情、用户信息）。

实现示例：


@Cache(name = "productCache")  // 先查缓存
public ProductDTO getProduct(String productId) {
    // 缓存未命中时执行，实际调用后端
    return loadProductFromBackend(productId);
}

// 对缓存未命中的后端调用添加限流
@RateLimiter(name = "productBackendLimiter")
private ProductDTO loadProductFromBackend(String productId) {
    return productService.getById(productId);
}

4.3.5 全链路组合：复杂场景的多层防护

组合逻辑：限流→缓存→重试→超时→舱壁→熔断，从流量入口到后端服务实现全链路防护。

适用场景：核心业务链路（如订单创建）。

实现示例：


@RateLimiter(name = "orderApiLimiter")  // 1. 限流
public OrderDTO createOrder(OrderRequest request) {
    return processOrderWithFallback(request);
}

@Cache(name = "productCache", fallbackMethod = "checkProductStockFallback")  // 2. 缓存查库存
private boolean checkProductStock(String productId, int quantity) {
    return inventoryService.checkStock(productId, quantity);
}

@Retry(name = "inventoryRetry")  // 3. 重试库存检查
@TimeLimiter(name = "inventoryTimeout")  // 4. 库存检查超时
private boolean checkProductStockFallback(String productId, int quantity, Exception e) {
    return inventoryService.checkStockWithFallback(productId, quantity);
}

@Bulkhead(name = "paymentBulkhead")  // 5. 支付并发控制
@CircuitBreaker(name = "paymentBreaker", fallbackMethod = "processPaymentFallback")  // 6. 支付熔断
private PaymentResult processPayment(PaymentRequest request) {
    return paymentService.process(request);
}

4.4 模块化使用注意事项

执行顺序控制：

注解式使用时，各模块的执行顺序由注解处理器决定，通常为：@Retry → @CircuitBreaker → @RateLimiter → @Bulkhead → @TimeLimiter。若需自定义顺序，需使用函数式API手动组合。

降级逻辑设计：

降级方法需与原方法参数一致，且最后一个参数为Exception。降级逻辑应轻量、可靠，避免依赖外部服务。

配置参数协同：

多模块组合时，参数需协同设计。例如：限流器limitForPeriod应大于舱壁maxConcurrentCalls，避免限流未触发而舱壁先满；重试次数应与超时时间匹配，避免总耗时过长。

避免过度防护：

并非所有方法都需要全量容错机制，需根据业务重要性和稳定性需求选择模块。例如，内部低延迟服务可仅用超时和重试，无需熔断和限流。

五、Resilience4j源码深度剖析

Resilience4j的源码设计简洁优雅，大量使用设计模式和函数式编程思想。本节将深入剖析核心模块的源码结构、关键逻辑和设计模式，帮助读者理解其底层实现。

5.1 核心框架设计与架构模式

Resilience4j的核心框架基于接口抽象和默认实现的分离设计，每个模块都遵循“接口→抽象类→实现类”的层次结构，并通过注册表管理实例生命周期。

5.1.1 核心接口与实现类结构

以CircuitBreaker为例，核心接口与实现类关系如下：


CircuitBreaker（接口） → AbstractCircuitBreaker（抽象类） → DefaultCircuitBreaker（实现类）

接口（CircuitBreaker）：定义核心功能（如executeSupplier、getState）；

抽象类（AbstractCircuitBreaker）：实现通用逻辑（如事件发布、状态检查）；

实现类（DefaultCircuitBreaker）：实现具体业务逻辑（如状态转换、失败率计算）。

这种结构确保了接口稳定性和实现灵活性，便于扩展新的实现类（如自定义断路器）。

5.1.2 注册表模式（Registry）

各模块通过Registry管理实例，如CircuitBreakerRegistry、RateLimiterRegistry。注册表的核心职责：

集中创建和缓存模块实例；

管理默认配置和实例专属配置；

支持实例事件监听和生命周期管理。

Registry核心源码：


// AbstractRegistry.java（通用注册表实现）
public abstract class AbstractRegistry<T, C> {
    // 存储实例：名称→实例
    private final ConcurrentMap<String, T> entries = new ConcurrentHashMap<>();
    // 默认配置
    private final C defaultConfig;
    // 实例创建工厂
    private final Function<String, T> entrySupplier;

    public AbstractRegistry(C defaultConfig, Function<String, T> entrySupplier) {
        this.defaultConfig = defaultConfig;
        this.entrySupplier = entrySupplier;
    }

    // 获取或创建实例
    public T get(String name) {
        return entries.computeIfAbsent(name, entrySupplier);
    }

    // 自定义配置创建实例
    public T get(String name, C config) {
        return entries.computeIfAbsent(name, (k) -> createEntry(name, config));
    }

    // 子类实现实例创建逻辑
    protected abstract T createEntry(String name, C config);
}

5.1.3 装饰器模式（Decorator）

Resilience4j通过装饰器模式为业务逻辑添加容错功能，核心是decorateXXX方法（如decorateSupplier、decorateRunnable）。装饰器模式的优势：

业务逻辑与容错逻辑分离，无侵入性；

支持动态组合多个装饰器（如同时装饰限流和熔断）。

装饰器核心源码：


// CircuitBreaker.java（断路器装饰器）
public static <T> Supplier<T> decorateSupplier(CircuitBreaker circuitBreaker, Supplier<T> supplier) {
    return () -> {
        // 执行前检查状态
        circuitBreaker.checkPermission();
        long start = System.nanoTime();
        try {
            // 执行原始业务逻辑
            T result = supplier.get();
            // 记录成功
            circuitBreaker.onSuccess(System.nanoTime() - start);
            return result;
        } catch (Exception e) {
            // 记录失败
            circuitBreaker.onFailure(System.nanoTime() - start, e);
            throw e;
        }
    };
}

5.2 断路器（CircuitBreaker）源码深度剖析

CircuitBreaker是Resilience4j最复杂的模块，其核心是状态机和环形缓冲区。以下是关键源码解析：

5.2.1 状态机实现

CircuitBreaker的状态由StateMachine管理，核心源码如下：


// StateMachine.java
public class StateMachine {
    // 原子引用存储当前状态
    private final AtomicReference<State> state = new AtomicReference<>(State.CLOSED);

    // 切换到关闭状态
    public void transitionToClosedState() {
        state.set(State.CLOSED);
    }

    // 切换到打开状态
    public void transitionToOpenState() {
        state.set(State.OPEN);
    }

    // 切换到半开状态
    public void transitionToHalfOpenState() {
        state.set(State.HALF_OPEN);
    }

    // 获取当前状态
    public State getState() {
        return state.get();
    }

    // 状态枚举
    public enum State {
        CLOSED, OPEN, HALF_OPEN
    }
}

5.2.2 环形缓冲区（RingBitBuffer）实现

环形缓冲区用于记录请求结果，核心源码如下：


// RingBitBuffer.java
public class RingBitBuffer {
    // 存储请求结果：1=失败，0=成功，-1=未使用
    private final int[] buffer;
    // 缓冲区大小
    private final int size;
    // 当前写入位置
    private final AtomicInteger index = new AtomicInteger(0);
    // 已记录的请求数
    private final AtomicInteger count = new AtomicInteger(0);

    public RingBitBuffer(int size) {
        this.size = size;
        this.buffer = new int[size];
        // 初始化缓冲区为-1（未使用）
        Arrays.fill(buffer, -1);
    }

    // 记录成功
    public void recordSuccess() {
        record(0);
    }

    // 记录失败
    public void recordFailure() {
        record(1);
    }

    // 记录结果
    private void record(int value) {
        int currentIndex = index.getAndIncrement() % size;  // 循环写入
        buffer[currentIndex] = value;
        count.incrementAndGet();  // 计数+1
    }

    // 获取已记录的请求数（不超过缓冲区大小）
    public int size() {
        return Math.min(count.get(), size);
    }

    // 计算失败率（百分比）
    public float calculateFailureRate() {
        int failureCount = 0;
        int totalCount = 0;
        for (int i = 0; i < buffer.length; i++) {
            if (buffer[i] == 1) {
                failureCount++;
            }
            if (buffer[i] != -1) {
                totalCount++;
            }
        }
        return totalCount == 0 ? 0.0f : (float) failureCount / totalCount * 100;
    }
}

5.2.3 核心执行逻辑

DefaultCircuitBreaker的executeSupplier方法是执行业务逻辑的入口：


// DefaultCircuitBreaker.java
@Override
public <T> T executeSupplier(Supplier<T> supplier) {
    // 1. 检查状态（打开状态直接拒绝）
    checkState();

    long startNanos = System.nanoTime();
    try {
        // 2. 执行原始业务逻辑
        T result = supplier.get();
        // 3. 记录成功并处理状态转换
        onSuccess(System.nanoTime() - startNanos);
        return result;
    } catch (Exception e) {
        // 4. 记录失败并处理状态转换
        onFailure(System.nanoTime() - startNanos, e);
        throw e;
    }
}

// 检查状态：打开状态且未到半开时间则抛出异常
private void checkState() {
    State currentState = stateMachine.getState();
    if (currentState == State.OPEN) {
        long elapsedNanos = System.nanoTime() - stateTransitionTimestamp;
        if (elapsedNanos < config.getWaitDurationInOpenState().toNanos()) {
            throw new CallNotPermittedException("CircuitBreaker is OPEN");
        } else {
            // 打开状态超时，切换到半开
            transitionToHalfOpenState();
        }
    }
}

5.3 限流器（RateLimiter）源码深度剖析

RateLimiter基于令牌桶算法实现，核心是令牌生成和获取逻辑。

5.3.1 令牌桶核心变量


// DefaultRateLimiter.java
public class DefaultRateLimiter implements RateLimiter {
    // 当前令牌数
    private final AtomicLong storedTokens = new AtomicLong(0);
    // 上次令牌生成时间（纳秒）
    private final AtomicLong lastRefillTime = new AtomicLong(System.nanoTime());
    // 配置
    private final RateLimiterConfig config;
    // 名称
    private final String name;
    // 事件发布器
    private final RateLimiterEventPublisher eventPublisher;
}

5.3.2 令牌生成与获取逻辑

令牌生成的核心是calculateNewTokens方法，令牌获取的核心是acquirePermission方法（详见3.2.3节源码）。

5.4 重试（Retry）源码深度剖析

Retry的核心是重试策略和异常过滤，核心逻辑在DefaultRetry的executeSupplier方法中：


// DefaultRetry.java
@Override
public <T> T executeSupplier(Supplier<T> supplier) {
    int attempt = 0;
    Throwable lastThrowable;

    while (true) {
        attempt++;  // 尝试次数+1
        try {
            // 执行业务逻辑
            T result = supplier.get();
            // 重试成功事件（若已重试过）
            if (attempt > 1) {
                eventPublisher.onSuccess(createSuccessEvent(attempt));
            }
            return result;
        } catch (Exception e) {
            lastThrowable = e;
            // 检查是否忽略异常
            if (isIgnoreException(e)) {
                throw e;
            }
            // 检查是否符合重试条件
            if (!isRetryException(e)) {
                throw e;
            }
            // 检查是否达到最大重试次数
            if (attempt >= config.getMaxAttempts()) {
                eventPublisher.onError(createErrorEvent(attempt, e));
                throw e;
            }
            // 计算重试间隔
            long waitDuration = calculateWaitDuration(attempt);
            // 发布重试事件
            eventPublisher.onRetry(createRetryEvent(attempt, waitDuration, e));
            // 等待重试间隔
            sleep(waitDuration);
        }
    }
}

// 检查异常是否符合重试条件
private boolean isRetryException(Exception e) {
    return config.getRetryExceptions().stream()
        .anyMatch(exceptionType -> exceptionType.isInstance(e));
}

// 计算重试间隔（支持指数退避）
private long calculateWaitDuration(int attempt) {
    if (config.getBackoffType() == BackoffType.EXPONENTIAL) {
        int exponent = attempt - 1;
        return (long) (config.getWaitDuration().toMillis() * Math.pow(config.getExponentialBackoffMultiplier(), exponent));
    } else {
        return config.getWaitDuration().toMillis();
    }
}

六、Resilience4j自定义拓展

Resilience4j的高扩展性使其能满足个性化业务需求。本节将介绍如何自定义限流算法、事件处理和缓存策略，以及扩展核心模块功能。

6.1 自定义限流算法

Resilience4j默认实现了令牌桶算法，若需使用其他算法（如漏桶、滑动窗口），可通过实现RateLimiter接口扩展。

6.1.1 自定义限流器步骤

实现RateLimiter接口：定义自定义限流逻辑；

定义配置类：继承RateLimiterConfig，添加自定义参数；

实现注册表：继承AbstractRegistry，创建自定义限流器实例；

注册与使用：在项目中使用自定义注册表创建限流器。

6.1.2 滑动窗口限流器示例

以下是基于滑动窗口算法的自定义限流器实现：


// 1. 自定义配置类
public class SlidingWindowRateLimiterConfig extends RateLimiterConfig {
    private final int windowSize;  // 窗口大小（秒）
    private final int limitPerWindow;  // 窗口内最大请求数

    private SlidingWindowRateLimiterConfig(int windowSize, int limitPerWindow) {
        this.windowSize = windowSize;
        this.limitPerWindow = limitPerWindow;
    }

    public int getWindowSize() {
        return windowSize;
    }

    public int getLimitPerWindow() {
        return limitPerWindow;
    }

    // 构建器
    public static class Builder {
        private int windowSize = 10;
        private int limitPerWindow = 100;

        public Builder windowSize(int windowSize) {
            this.windowSize = windowSize;
            return this;
        }

        public Builder limitPerWindow(int limitPerWindow) {
            this.limitPerWindow = limitPerWindow;
            return this;
        }

        public SlidingWindowRateLimiterConfig build() {
            return new SlidingWindowRateLimiterConfig(windowSize, limitPerWindow);
        }
    }
}

// 2. 自定义限流器实现
public class SlidingWindowRateLimiter implements RateLimiter {
    private final String name;
    private final SlidingWindowRateLimiterConfig config;
    // 滑动窗口计数器：存储每个时间片的请求数
    private final ConcurrentHashMap<Long, AtomicInteger> windowCounts = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public SlidingWindowRateLimiter(String name, SlidingWindowRateLimiterConfig config) {
        this.name = name;
        this.config = config;
        // 定时清理过期时间片（每窗口大小/2执行一次）
        scheduler.scheduleAtFixedRate(this::cleanExpiredWindows,
            config.getWindowSize() / 2,
            config.getWindowSize() / 2,
            TimeUnit.SECONDS);
    }

    @Override
    public String getName() {
        return name;
    }

    @Override
    public boolean acquirePermission() {
        return acquirePermission(0, TimeUnit.NANOSECONDS);
    }

    @Override
    public boolean acquirePermission(long timeoutDuration, TimeUnit timeUnit) {
        long currentTime = System.currentTimeMillis() / 1000;  // 秒级时间片
        long windowStart = currentTime - config.getWindowSize() + 1;  // 窗口起始时间

        // 计算当前窗口内的总请求数
        int totalRequests = windowCounts.entrySet().stream()
            .filter(entry -> entry.getKey() >= windowStart)
            .mapToInt(entry -> entry.getValue().get())
            .sum();

        // 若总请求数≥阈值，返回失败
        if (totalRequests >= config.getLimitPerWindow()) {
            return false;
        }

        // 记录当前时间片的请求数
        windowCounts.computeIfAbsent(currentTime, k -> new AtomicInteger(0))
            .incrementAndGet();
        return true;
    }

    // 清理过期时间片
    private void cleanExpiredWindows() {
        long windowStart = System.currentTimeMillis() / 1000 - config.getWindowSize() + 1;
        windowCounts.keySet().removeIf(time -> time < windowStart);
    }

    // 其他方法实现...
}

// 3. 自定义注册表
public class SlidingWindowRateLimiterRegistry extends AbstractRegistry<RateLimiter, SlidingWindowRateLimiterConfig> {
    public SlidingWindowRateLimiterRegistry(SlidingWindowRateLimiterConfig defaultConfig) {
        super(defaultConfig, name -> new SlidingWindowRateLimiter(name, defaultConfig));
    }

    @Override
    protected RateLimiter createEntry(String name, SlidingWindowRateLimiterConfig config) {
        return new SlidingWindowRateLimiter(name, config);
    }
}

// 4. 使用自定义限流器
public class CustomRateLimiterExample {
    public static void main(String[] args) {
        // 创建自定义配置
        SlidingWindowRateLimiterConfig config = new SlidingWindowRateLimiterConfig.Builder()
            .windowSize(10)  // 10秒窗口
            .limitPerWindow(100)  // 窗口内最多100次请求
            .build();
        // 创建注册表
        SlidingWindowRateLimiterRegistry registry = new SlidingWindowRateLimiterRegistry(config);
        // 获取限流器实例
        RateLimiter rateLimiter = registry.rateLimiter("customLimiter");

        // 使用限流器
        for (int i = 0; i < 150; i++) {
            boolean permitted = rateLimiter.acquirePermission();
            System.out.println("Request " + i + ": " + (permitted ? "Permitted" : "Denied"));
        }
    }
}

6.2 自定义事件处理

通过实现EventConsumer接口，可自定义事件处理逻辑（如事件持久化、告警通知）。

6.2.1 事件持久化到数据库


// 自定义断路器事件消费者
public class DbCircuitBreakerEventConsumer implements EventConsumer<CircuitBreakerEvent> {
    private final JdbcTemplate jdbcTemplate;

    public DbCircuitBreakerEventConsumer(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @Override
    public void accept(CircuitBreakerEvent event) {
        String sql = "INSERT INTO circuit_breaker_events (" +
            "event_id, circuit_breaker_name, event_type, creation_time, details) " +
            "VALUES (?, ?, ?, ?, ?)";
        jdbcTemplate.update(sql,
            UUID.randomUUID().toString(),
            event.getCircuitBreakerName(),
            event.getClass().getSimpleName(),
            new Timestamp(event.getCreationTime().toEpochMilli()),
            event.toString()
        );
    }
}

// 注册事件消费者
@Configuration
public class EventConfig {
    @Bean
    public DbCircuitBreakerEventConsumer dbCircuitBreakerEventConsumer(JdbcTemplate jdbcTemplate) {
        return new DbCircuitBreakerEventConsumer(jdbcTemplate);
    }

    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry(DbCircuitBreakerEventConsumer consumer) {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom().build();
        // 创建注册表并注册事件消费者
        return CircuitBreakerRegistry.of(config, registryEventConsumer -> {
            registryEventConsumer.onEntryAdded(event -> {
                CircuitBreaker circuitBreaker = event.getAddedEntry();
                circuitBreaker.getEventPublisher().onEvent(consumer);
            });
        });
    }
}

6.2.2 事件触发告警通知


// 限流事件告警消费者
public class RateLimiterAlertConsumer implements EventConsumer<RateLimiterEvent> {
    private final AlertService alertService;
    // 告警频率控制（避免告警风暴）
    private final RateLimiter alertLimiter = RateLimiter.ofDefaults("alertLimiter");

    public RateLimiterAlertConsumer(AlertService alertService) {
        this.alertService = alertService;
    }

    @Override
    public void accept(RateLimiterEvent event) {
        // 仅处理限流失败事件
        if (event instanceof RateLimiterOnFailureEvent) {
            // 控制告警频率（每分钟最多10条）
            if (alertLimiter.acquirePermission()) {
                alertService.sendSms("Rate limiter " + event.getRateLimiterName() + " is triggering frequently");
                alertService.sendEmail("admin@example.com", "Rate Limiter Alert", event.toString());
            }
        }
    }
}

6.3 自定义缓存策略

Resilience4j默认支持Caffeine和Guava缓存，若需集成其他缓存框架（如Redis），可实现Cache接口。

6.3.1 Redis缓存实现


// Redis缓存实现
public class RedisCache<K, V> implements Cache<K, V> {
    private final RedisTemplate<K, V> redisTemplate;
    private final Duration expirationDuration;

    public RedisCache(RedisTemplate<K, V> redisTemplate, Duration expirationDuration) {
        this.redisTemplate = redisTemplate;
        this.expirationDuration = expirationDuration;
    }

    @Override
    public V computeIfAbsent(K key, Supplier<V> supplier) throws Exception {
        // 先查Redis缓存
        V value = redisTemplate.opsForValue().get(key);
        if (value != null) {
            return value;
        }
        // 缓存未命中，执行Supplier获取值
        value = supplier.get();
        // 存入Redis并设置过期时间
        redisTemplate.opsForValue().set(key, value, expirationDuration);
        return value;
    }

    @Override
    public void put(K key, V value) {
        redisTemplate.opsForValue().set(key, value, expirationDuration);
    }

    @Override
    public V get(K key) {
        return redisTemplate.opsForValue().get(key);
    }

    @Override
    public void invalidate(K key) {
        redisTemplate.delete(key);
    }

    @Override
    public void invalidateAll() {
        // Redis中通常通过前缀删除，需自定义实现
        throw new UnsupportedOperationException("Invalidate all not supported");
    }
}

// Redis缓存注册表
public class RedisCacheRegistry {
    private final RedisTemplate<String, Object> redisTemplate;
    private final ConcurrentMap<String, Cache<String, Object>> caches = new ConcurrentHashMap<>();
    private final Duration defaultExpiration;

    public RedisCacheRegistry(RedisTemplate<String, Object> redisTemplate, Duration defaultExpiration) {
        this.redisTemplate = redisTemplate;
        this.defaultExpiration = defaultExpiration;
    }

    public Cache<String, Object> cache(String name) {
        return caches.computeIfAbsent(name, k -> new RedisCache<>(redisTemplate, defaultExpiration));
    }

    public Cache<String, Object> cache(String name, Duration expiration) {
        return caches.computeIfAbsent(name, k -> new RedisCache<>(redisTemplate, expiration));
    }
}

// 使用Redis缓存
@Service
public class ProductService {
    private final Cache<String, ProductDTO> productCache;

    public ProductService(RedisCacheRegistry cacheRegistry) {
        this.productCache = cacheRegistry.cache("productCache", Duration.ofMinutes(10));
    }

    public ProductDTO getProduct(String id) throws Exception {
        // 使用Redis缓存
        return productCache.computeIfAbsent(id, this::loadProductFromDb);
    }

    private ProductDTO loadProductFromDb(String id) {
        // 从数据库加载商品
        return productRepository.findById(id)
            .orElseThrow(() -> new ProductNotFoundException(id));
    }
}

6.4 扩展核心模块功能

通过继承核心模块的抽象类或实现接口，可扩展其功能（如添加自定义指标、增强状态管理）。

6.4.1 扩展断路器添加自定义指标


// 扩展断路器，添加成功/失败计数指标
public class MetricCircuitBreaker extends AbstractCircuitBreaker {
    private final MeterRegistry meterRegistry;
    private final Counter successCounter;
    private final Counter failureCounter;

    public MetricCircuitBreaker(String name, CircuitBreakerConfig config,
                               CircuitBreakerEventPublisher eventPublisher,
                               MeterRegistry meterRegistry) {
        super(name, config, eventPublisher);
        this.meterRegistry = meterRegistry;
        // 初始化指标计数器
        this.successCounter = meterRegistry.counter("circuitbreaker.success", "name", name);
        this.failureCounter = meterRegistry.counter("circuitbreaker.failure", "name", name);
    }

    @Override
    protected void onSuccess(long duration) {
        super.onSuccess(duration);
        successCounter.increment();  // 成功计数+1
    }

    @Override
    protected void onFailure(long duration, Exception exception) {
        super.onFailure(duration, exception);
        failureCounter.increment();  // 失败计数+1
    }
}

// 扩展断路器注册表
public class MetricCircuitBreakerRegistry extends AbstractRegistry<CircuitBreaker, CircuitBreakerConfig> {
    private final MeterRegistry meterRegistry;

    public MetricCircuitBreakerRegistry(CircuitBreakerConfig defaultConfig, MeterRegistry meterRegistry) {
        super(defaultConfig);
        this.meterRegistry = meterRegistry;
    }

    @Override
    protected CircuitBreaker createEntry(String name, CircuitBreakerConfig config) {
        CircuitBreakerEventPublisher eventPublisher = CircuitBreakerEventPublisher.of(config);
        return new MetricCircuitBreaker(name, config, eventPublisher, meterRegistry);
    }
}