限流器 #

一、限流概述 #

1.1 为什么需要限流 #

text

                    无限流：
                    
客户端 ──► ──► ──► ──► ──► 服务
  │                           │
  └── 大量请求 ────────────────┘
                │
                ▼
            服务过载

                    有限流：
                    
客户端 ──► ──► ──► ──► ──► 限流器 ──► 服务
  │                           │
  └── 大量请求 ────────────────┘
                │
                ▼
           超出限制被拒绝

1.2 限流算法 #

算法	说明	特点
令牌桶	按速率生成令牌	允许突发流量
漏桶	恒定速率处理	平滑流量
滑动窗口	滑动时间窗口	精确控制

二、基本使用 #

2.1 添加依赖 #

xml

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-ratelimiter</artifactId>
</dependency>

2.2 配置限流器 #

yaml

resilience4j:
  ratelimiter:
    configs:
      default:
        limitForPeriod: 10
        limitRefreshPeriod: 1s
        timeoutDuration: 0
    instances:
      userService:
        baseConfig: default
      orderService:
        baseConfig: default
        limitForPeriod: 20

2.3 使用注解 #

java

@Service
public class UserService {

    @RateLimiter(name = "userService", fallbackMethod = "getUserFallback")
    public User getUser(Long id) {
        return userClient.getUserById(id);
    }

    private User getUserFallback(Long id, Throwable t) {
        throw new RuntimeException("请求过于频繁，请稍后重试");
    }
}

三、配置详解 #

3.1 完整配置 #

yaml

resilience4j:
  ratelimiter:
    configs:
      default:
        limitForPeriod: 10
        limitRefreshPeriod: 1s
        timeoutDuration: 5s
        registerHealthIndicator: true
    instances:
      userService:
        baseConfig: default
        limitForPeriod: 100
      orderService:
        baseConfig: default
        limitForPeriod: 50

3.2 配置项说明 #

配置项	默认值	说明
limitForPeriod	50	每个周期的许可数
limitRefreshPeriod	500ns	许可刷新周期
timeoutDuration	5s	等待许可的超时时间
registerHealthIndicator	false	是否注册健康指示器

3.3 计算QPS #

text

QPS = limitForPeriod / limitRefreshPeriod

例如：
limitForPeriod = 10
limitRefreshPeriod = 1s
QPS = 10 / 1 = 10

四、编程式使用 #

4.1 创建限流器 #

java

@Configuration
public class RateLimiterConfig {

    @Bean
    public RateLimiter userServiceRateLimiter() {
        RateLimiterConfig config = RateLimiterConfig.custom()
            .limitForPeriod(10)
            .limitRefreshPeriod(Duration.ofSeconds(1))
            .timeoutDuration(Duration.ofSeconds(5))
            .build();
        
        return RateLimiter.of("userService", config);
    }
}

4.2 使用限流器 #

java

@Service
public class UserService {

    @Autowired
    private RateLimiter rateLimiter;

    public User getUser(Long id) {
        Supplier<User> supplier = RateLimiter.decorateSupplier(
            rateLimiter,
            () -> userClient.getUserById(id)
        );
        
        return Try.ofSupplier(supplier)
            .recover(RequestNotPermitted.class, e -> {
                throw new RuntimeException("请求过于频繁");
            })
            .get();
    }
}

4.3 检查许可 #

java

@Service
public class UserService {

    @Autowired
    private RateLimiter rateLimiter;

    public User getUser(Long id) {
        if (rateLimiter.acquirePermission()) {
            return userClient.getUserById(id);
        } else {
            throw new RuntimeException("请求过于频繁");
        }
    }
}

五、组合使用 #

5.1 限流+熔断 #

java

@Service
public class UserService {

    @Autowired
    private RateLimiter rateLimiter;
    
    @Autowired
    private CircuitBreaker circuitBreaker;

    public User getUser(Long id) {
        Supplier<User> supplier = SupplierUtils
            .compose(
                RateLimiter.decorateSupplier(rateLimiter, () -> userClient.getUserById(id)),
                CircuitBreaker.decorateSupplier(circuitBreaker, () -> userClient.getUserById(id))
            );
        
        return Try.ofSupplier(supplier)
            .recover(throwable -> createDefaultUser(id))
            .get();
    }
}

5.2 注解组合 #

java

@Service
public class UserService {

    @RateLimiter(name = "userService")
    @CircuitBreaker(name = "userService", fallbackMethod = "fallback")
    public User getUser(Long id) {
        return userClient.getUserById(id);
    }

    private User fallback(Long id, Throwable t) {
        return createDefaultUser(id);
    }
}

六、自定义限流策略 #

6.1 基于IP限流 #

java

@Configuration
public class RateLimiterConfig {

    @Bean
    public RateLimiter ipRateLimiter() {
        return RateLimiter.of("ipRateLimiter", 
            RateLimiterConfig.custom()
                .limitForPeriod(100)
                .limitRefreshPeriod(Duration.ofSeconds(1))
                .build()
        );
    }
}

6.2 基于用户限流 #

java

@Service
public class UserService {

    private final Map<Long, RateLimiter> userRateLimiters = new ConcurrentHashMap<>();

    public User getUser(Long userId, Long id) {
        RateLimiter rateLimiter = userRateLimiters.computeIfAbsent(userId, 
            uid -> RateLimiter.of("user-" + uid, 
                RateLimiterConfig.custom()
                    .limitForPeriod(10)
                    .limitRefreshPeriod(Duration.ofSeconds(1))
                    .build()
            )
        );
        
        return RateLimiter.decorateSupplier(rateLimiter, 
            () -> userClient.getUserById(id)
        ).get();
    }
}

七、事件监听 #

7.1 配置事件监听 #

java

@Configuration
public class RateLimiterEventConfig {

    @Bean
    public RateLimiterEventListener rateLimiterEventListener(
            RateLimiterRegistry registry) {
        
        registry.getAllRateLimiters().forEach(rateLimiter -> {
            rateLimiter.getEventPublisher()
                .onSuccess(event -> log.info("Rate limiter success: {}", event))
                .onFailure(event -> log.warn("Rate limiter failure: {}", event));
        });
        
        return new RateLimiterEventListener();
    }
}

7.2 事件类型 #

事件类型	说明
RateLimiterOnSuccessEvent	获取许可成功
RateLimiterOnFailureEvent	获取许可失败

八、监控指标 #

8.1 获取指标 #

java

@RestController
public class RateLimiterController {

    @Autowired
    private RateLimiterRegistry registry;

    @GetMapping("/rate-limiter/{name}/metrics")
    public Map<String, Object> getMetrics(@PathVariable String name) {
        RateLimiter rateLimiter = registry.rateLimiter(name);
        RateLimiter.Metrics metrics = rateLimiter.getMetrics();
        
        Map<String, Object> result = new HashMap<>();
        result.put("availablePermissions", metrics.getAvailablePermissions());
        result.put("numberOfWaitingThreads", metrics.getNumberOfWaitingThreads());
        return result;
    }
}

8.2 Actuator端点 #

yaml

management:
  endpoints:
    web:
      exposure:
        include: health,ratelimiters

九、最佳实践 #

9.1 生产环境配置 #

yaml

resilience4j:
  ratelimiter:
    configs:
      default:
        limitForPeriod: 100
        limitRefreshPeriod: 1s
        timeoutDuration: 0
        registerHealthIndicator: true
    instances:
      userService:
        baseConfig: default
        limitForPeriod: 200
      orderService:
        baseConfig: default
        limitForPeriod: 100

9.2 注意事项 #

注意点	说明
QPS计算	合理计算限流阈值
超时设置	避免长时间等待
降级处理	提供友好的限流提示
监控告警	配置限流事件告警

十、总结 #

10.1 核心要点 #

要点	说明
令牌桶算法	允许突发流量
配置项	limitForPeriod、limitRefreshPeriod
注解使用	@RateLimiter
组合使用	与熔断器组合使用

10.2 下一步 #

现在你已经掌握了限流器的使用，接下来让我们学习重试机制，了解如何实现自动重试！