Guava RateLimiter 限流实战：从令牌桶原理到 Nacos 动态配置

RocketMQ 消费端一秒进来几万条消息数据库连接池只有 20 个连接——不限流DB 瞬间打满。Guava 的 RateLimiter 就是给消费端加一个水龙头令牌桶出令牌消费线程抢到令牌才能执行抢不到就等着。这篇文章从令牌桶的原理讲起到两种落地写法硬编码 Nacos 动态配置 RefreshScope全部用真实代码串下来。没有限流的时候问题长什么样正常流量下一切正常。秒杀、大促、热点事件一来消息队列瞬间堆积几十万条——消费端多线程全力消费每秒几百条 SQL 打到 DB 上。正常 100 条/秒 → MySQL 稳如老狗 高峰 10000 条/秒 → MySQL 连接池耗尽 → 请求超时 → 雪崩MySQLConsumerRocketMQMySQLConsumerRocketMQ连接池满了堆积越来越严重推送消息大批量INSERT × 100...INSERT × 100...INSERT × 100...INSERT等连接……超时消费失败消息重试不是 MySQL 不行是没有控制消费速度。解决办法就是在消费端加个限流器——令牌桶就是干这个的。令牌桶是什么每秒存 N 个有令牌拿走一个有令牌拿走一个没令牌了等着令牌工厂每秒生产 N 个令牌令牌桶最多存 M 个令牌处理请求 1处理请求 2处理请求 3原理就四句话系统以固定速率往桶里放令牌比如每秒 5000 个每个请求处理前必须从桶里拿到一个令牌桶满了就不再放令牌没令牌了就等着新令牌产生Guava 的RateLimiter封装好了这一切。你只需要两行代码// 创建每秒生成 5000 个令牌privateRateLimiterrateLimiterRateLimiter.create(5000);// 消费前拿令牌有就直接返回没有就阻塞等待rateLimiter.acquire();基础写法——硬编码在 Consumer 里ComponentRocketMQMessageListener(consumerGroupgroup_count_following_2_db,topicMQConstants.TOPIC_COUNT_FOLLOWING_2_DB)Slf4jpublicclassCountFollowing2DBConsumerimplementsRocketMQListenerString{ResourceprivateUserCountDOMapperuserCountDOMapper;// 每秒 5000 个令牌——写死在代码里privateRateLimiterrateLimiterRateLimiter.create(5000);OverridepublicvoidonMessage(Stringbody){// 拿令牌没有就阻塞等rateLimiter.acquire();// 消息体 JSON → DTOCountFollowUnfollowMqDTOcountDTOJsonUtils.parseObject(body,CountFollowUnfollowMqDTO.class);intcountcountDTO.getType().equals(FollowUnfollowTypeEnum.FOLLOW.getCode())?1:-1;userCountDOMapper.insertOrUpdateFollowingTotalByUserId(count,countDTO.getUserId());}}这个写法下RateLimiter是这个 Consumer Bean 的一个成员变量当前实例内共享。部署多实例的话每个实例的 Consumer 各自持有一个独立的 RateLimiter——实例 A 的限流器管不了实例 B 的消费速度。所以你要是在 Nacos 里配了rate-limit: 5000部署了 3 个实例打到 DB 的 QPS 上限实际上是3 × 5000 15000。进阶写法——Nacos 动态配置改限流阈值不重启硬编码 5000要是秒杀活动流量翻倍想临时调成 10000——改代码、打包、部署、重启半小时过去了。把阈值抽到 Nacos 配置里加上RefreshScope控制台改完实时生效ConfigurationRefreshScopepublicclassFollowUnfollowMqConsumerRateLimitConfig{Value(${mq-consumer.follow-unfollow.rate-limit})privatedoublerateLimit;BeanRefreshScopepublicRateLimiterrateLimiter(){returnRateLimiter.create(rateLimit);}}# application-dev.yml——阈值写本地mq-consumer:follow-unfollow:rate-limit:5000或者放进 Nacos 配置文件里动态刷新# Nacos 里的 your-service-dev.yamlmq-consumer:follow-unfollow:rate-limit:5000改了 Nacos 里的值RefreshScope销毁旧的 RateLimiter Bean用新的阈值重建一个。Consumer 端注入这个 Bean 即可ComponentRocketMQMessageListener(consumerGroupgroup_follow_unfollow,topicMQConstants.TOPIC_FOLLOW_OR_UNFOLLOW)Slf4jpublicclassFollowUnfollowConsumerimplementsRocketMQListenerMessage{ResourceprivateRateLimiterrateLimiter;// 注入配置类里创建的 BeanOverridepublicvoidonMessage(Messagemessage){rateLimiter.acquire();// 拿令牌// 消费逻辑...}}硬编码 vs Nacos 动态配置方式优点缺点硬编码简单不需要额外配置改阈值要重启Nacos RefreshScope控制台改完实时生效不用重启多了一个配置类生产环境建议后者原因有两个。一是应急——线上 DB 突然扛不住了Nacos 里把 5000 改成 2000所有实例秒级生效不用重启。二是配合扩缩容维持全局 QPS 不变——这个更重要DB 能扛住的写入上限10000/s 部署 1 台实例 rate-limit 10000 → 总 QPS 10000 ✅ 扩容到 3 台实例 不改配置的话 → 总 QPS 30000 DB 打挂 Nacos 里改成 3300 → 总 QPS ≈ 10000 ✅实例数量变了不改代码、不重启Nacos 控制台改一个数字所有实例的总 QPS 重新对齐到 DB 能承受的值。RateLimiter 的核心方法RateLimiterlimiterRateLimiter.create(5000);// 每秒 5000 个令牌// acquire()阻塞获取没令牌就一直等limiter.acquire();// 拿 1 个令牌阻塞limiter.acquire(10);// 拿 10 个令牌阻塞// tryAcquire()非阻塞获取没令牌立刻返回 falseif(limiter.tryAcquire()){// 拿到了执行}else{// 没拿到直接跳过适用于可丢弃的场景}// tryAcquire(timeout)限定时间内的阻塞获取if(limiter.tryAcquire(500,TimeUnit.MILLISECONDS)){// 500ms 内拿到了}else{// 超时没拿到}// getRate() / setRate()查询和动态修改速率doublecurrentRatelimiter.getRate();// 当前速率limiter.setRate(10000);// 动态改成 10000/sMQ 消费端几乎只用acquire()——等他一会儿比跳过这条消息重要得多。放在 consumeMessage 的哪一行很多人会把acquire()放在onMessage的最后几行——觉得先处理业务再限流。错了。acquire()必须放在onMessage的第一行。在消费逻辑之前就把令牌拿了不让多余的请求进到 DB 操作里。OverridepublicvoidonMessage(Messagemessage){// ✅ 第一行就拿令牌——DB 操作之前就被限流了rateLimiter.acquire();// ❌ 不要放在这里——SQL 都执行完了才限流没意义StringbodyJsonStrnewString(message.getBody());// ...DB 操作...}你的各个消费者都设多少阈值不是所有 Consumer 都需要限流。判断要不要加Consumer 有没有 DB 操作没有比如只写 Redis→ 不加Redis 扛得住DB 操作是单条还是批量批量一次写多条→ 阈值设低一点顺带有没有其他外部调用有 → 阈值设低一点外部接口慢// 批量写入型 Consumer——阈值设低一点比如 1000/s// 因为每次消费处理的不是一条 SQL是一批privateRateLimiterrateLimiterRateLimiter.create(1000);// 单条写入型 Consumer——阈值可以高一些比如 5000/sprivateRateLimiterrateLimiterRateLimiter.create(5000);// 无 DB 操作只操作 Redis——不需要限流// Redis 单机 10w QPS消费端这点量打不挂注意RateLimiter 是 JVM 级别的限的是单个实例RateLimiter 是 Guava 的本地限流器——它只是 JVM 内存里的一个计数器不跨网络不跨实例。这意味着两件事一个 Consumer 里的 RateLimiter 对这个实例内所有消费线程生效。ConsumeMode.ORDERLY下一个 Consumer 可能被分配了 4 个 MessageQueue4 个线程各自消费。这 4 个线程共享同一个 RateLimiter——加起来每秒最多 5000 个令牌一个线程acquire()阻塞了其他线程也能拿到令牌。部署了 3 个服务实例打到 DB 的总 QPS 是3 × 5000 15000。每个实例各自一个 RateLimiter互不干扰。所以如果你希望全局限制 DB 写入 QPS 不超过 5000不管多少实例Guava RateLimiter 做不到——它管不了别的 JVM。这种情况需要用 Redis 做分布式限流。