Java中RabbitMQ限流需在消费者端或网关层实现令牌桶算法，而非依赖RabbitMQ的prefetch机制；推荐单机用Guava RateLimiter、分布式场景用Redis+Lua脚本保证原子性与全局一致性。
Java 中使用 RabbitMQ 实现基于 Token 令牌桶算法的限流，核心在于**不依赖 RabbitMQ 自身限流（如 QoS），而是在消费者端或前置网关/服务层做主动令牌校验**。RabbitMQ 本身不内置令牌桶，需结合内存或分布式令牌桶（如 Redis + Lua）实现精准、可扩展的限流。
1. 为什么不能只靠 RabbitMQ 的 prefetch 设置？
RabbitMQ 的 basic.qos(prefetch=1) 是通道级“最多同时处理 N 条”，属于粗粒度流控，无法按时间窗口（如每秒 100 次）、区分用户/接口、动态调整速率——这些正是令牌桶的核心能力。
- prefetch 控制的是未 ack 消息数，不是请求频次
- 无法实现“突发允许 + 平滑匀速”效果（令牌桶优势）
- 多实例部署时，各消费者独立 prefetch，无全局速率控制
2. 推荐架构：消费者端集成令牌桶（单机轻量场景）
适合中小流量、单消费者实例、对一致性要求不苛刻的场景。用 guava RateLimiter 在消费逻辑前加令牌校验：
// 初始化：每秒生成 50 个 token，允许 10 个预支（应对突发）
private final RateLimiter rateLimiter = RateLimiter.create(50.0, 10, TimeUnit.SECONDS);
<p>@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage msg) {
// 阻塞等待令牌（或用 tryAcquire 非阻塞+降级）
if (!rateLimiter.tryAcquire(1, 100, TimeUnit.MILLISECONDS)) {
log.warn("Rate limit exceeded for message: {}", msg.getId());
// 可选择：拒绝（抛异常触发 nack）、延迟重试、写入死信队列
throw new RuntimeException("Rate limited");
}</p><pre class="brush:php;toolbar:false;">// ✅ 有令牌才真正处理业务
processOrder(msg);}
注意：Guava RateLimiter 基于 SmoothBursty，本质是令牌桶；但它是单机内存态，多实例需改用分布式方案。
3. 生产可用：Redis + Lua 实现分布式令牌桶
多消费者实例共用同一桶，保证全局速率一致。关键点：原子性（用 Lua 脚本避免竞态）、低延迟、支持动态配置。
示例 Lua 脚本（token_bucket.lua）：
-- KEYS[1]: bucket key (e.g., "rate:order:api")
-- ARGV[1]: capacity (max tokens)
-- ARGV[2]: refill rate per second
-- ARGV[3]: current timestamp (ms)
-- ARGV[4]: tokens needed (usually 1)
<p>local bucket_key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now_ms = tonumber(ARGV[3])
local need = tonumber(ARGV[4])</p><p>local state = redis.call('hmget', bucket_key, 'last_ts', 'tokens')
local last_ts = tonumber(state[1]) or now_ms
local tokens = tonumber(state[2]) or capacity</p><p>-- 计算新增令牌：按时间差补发，不超过 capacity
local delta_ms = now_ms - last_ts
local delta_tokens = math.floor(delta_ms * rate / 1000.0)
tokens = math.min(capacity, tokens + delta_tokens)</p><p>-- 检查是否足够
if tokens >= need then
tokens = tokens - need
redis.call('hmset', bucket_key, 'last_ts', now_ms, 'tokens', tokens)
return 1 -- success
else
redis.call('hmset', bucket_key, 'last_ts', now_ms, 'tokens', tokens)
return 0 -- rejected
end</p>
Java 调用（Spring Boot + Lettuce）：
@Resource
private RedisTemplate<string object> redisTemplate;
<p>private static final DefaultRedisScript<long> TOKEN_SCRIPT =
new DefaultRedisScript("token_bucket.lua", Long.class);</long></p>
<p>public boolean tryAcquire(String bucketKey, int capacity, double ratePerSec) {
long nowMs = System.currentTimeMillis();
Long result = redisTemplate.execute(
TOKEN_SCRIPT,
Collections.singletonList(bucketKey),
String.valueOf(capacity),
String.valueOf(ratePerSec),
String.valueOf(nowMs),
"1"
);
return result != null && result == 1L;
}</p>
<p>@RabbitListener(queues = "order.queue")
public void handleOrder(OrderMessage msg) {
if (!tryAcquire("rate:order:consumer", 100, 20.0)) { // 容量100，每秒补20
// 触发限流策略：nack 不 requeue 或发到死信
channel.basicNack(deliveryTag, false, false);
return;
}
processOrder(msg);
}</p></string>
4. 关键细节与建议
-
令牌桶 Key 设计：按业务维度隔离，如
"rate:service:order:create"、"rate:user:1001"，避免一刀切 - 失败处理策略：直接 reject（丢弃）、nack+requeue（可能重复）、nack+不 requeue（进死信）、返回限流响应（若消息含回调地址）
-
监控与告警：记录
rate_limited_count指标，接入 Prometheus；桶中 token 余量可定期采样观察水位 - 冷启动问题：首次调用时桶为空，可通过初始化脚本预设 tokens，或接受首波少量请求被限
- 与消息重试协同：限流失败不应计入业务重试次数，需在 consumer 端明确区分“限流拒绝”和“业务异常”
不复杂但容易忽略：令牌桶必须和消息处理生命周期对齐——校验发生在消息从队列取出后、业务逻辑执行前；且要确保校验失败时消息能被正确 nack 或路由，避免无限循环或堆积。
Java免费学习笔记：立即使用
解锁 Java 大师之旅：从入门到精通的终极指南
