Spring 项目里给 RocketMQ 消息限流：从生产端到消费端的完整方案

一句话结论：RocketMQ 是缓冲区，不是限流器。 它能把流量削平，但不会替你保护下游数据库，也不会阻止生产端把 Broker 打爆。限流要自己加，而且要先把“速率限制”“并发限制”“背压”这三件事分清楚。

一、为什么 RocketMQ 需要额外限流

典型场景：大促或者批量导入，生产端瞬间灌进几十万条消息。Broker 扛住了，但消费端以 consumeThreadMax = 64 的全速消费，分分钟把 MySQL 连接池、下游 HTTP 接口打满，然后整条链路一起雪崩。

很多人第一反应是“RocketMQ 有没有限流开关”。答案是：没有业务语义上的 QPS 限流开关。 它自带的保护机制解决的是另一个问题：

RocketMQ 自带机制	实际作用	是否等于业务限流
Broker 发送队列满后拒绝 / [TIMEOUT_CLEAN_QUEUE]	保护 Broker 自身不被压垮	否，是故障保护，且表现为报错
pullThresholdForQueue / pullThresholdSizeForQueue	客户端缓存太多消息时暂停拉取	否，是背压（backpressure）
consumeThreadMax	限制消费线程数	否，限制的是并发，不是每秒处理量
maxReconsumeTimes + 延迟等级	失败重试与退避	否，是失败处理

所以限流必须落在应用侧。一条完整的消息链路，有三个可以下手的位置：

上游业务 ──▶ [生产端限流] ──▶ Broker（缓冲、堆积） ──▶ [消费端限流 / 并发控制 / 背压] ──▶ DB、HTTP 下游

选择位置的优先级很简单：越靠近你要保护的瓶颈越好。

• 保护 Broker 和发送方自己 → 生产端限流；
• 保护 DB、第三方接口、消费端进程 → 消费端限流 + 并发控制；
• 下游时快时慢、偶尔故障 → 背压 + 熔断降级，而不是一个写死的 QPS。

先把四个概念分清楚

概念	回答的问题	典型手段	被触发时的表现
速率限制	每秒最多处理/发送多少条	令牌桶、漏桶、滑动窗口	拒绝、等待或延迟重投
并发限制	同时有多少条在处理	Semaphore、线程池、consumeThreadMax	线程排队/阻塞
背压	处理不过来时怎么让上游慢下来	调小拉取参数、suspend()/resume()	消息留在 Broker，延迟上升
熔断降级	下游已经挂了怎么办	Sentinel、Resilience4j	快速失败、走降级逻辑

这四个东西不能互相替代。 下面这些写法在线上都很常见，但都是错的：

• 在 onMessage 里 Thread.sleep(50)：只是把消费线程占住，堆积从 Broker 挪到了客户端，监控上“消费 TPS 正常”，实际延迟已经爆炸；
• 把 consumeThreadMax 调小当成 QPS 限流：它只限制并发，每条消息耗时 10ms 时实际 QPS 依然可能很大；
• 多个 Pod 都用 Guava RateLimiter：集群总速率 = 单机值 × Pod 数，扩缩容后阈值直接漂移；
• 只在生产端限流：消费端仍然会被历史堆积打垮。

---

二、实现方式有哪几种

按“作用范围”和“实现手段”两个维度，可以整理成下面这张表：

方案	位置	限制维度	作用范围	额外依赖	适用场景
Guava RateLimiter	生产 / 消费	速率（可叠加 Semaphore）	单实例	Guava	单机部署、快速接入
Redis + Lua 令牌桶	生产 / 消费	速率 + 突发容量	集群总量	Redis	多实例需要精确总量
Redisson RRateLimiter	生产 / 消费	速率	集群总量	Redis + Redisson	不想自己写 Lua
Sentinel	生产 / 消费	QPS、线程数、匀速排队、熔断	单机 / 集群流控	Sentinel	需要熔断、控制台、动态规则
Semaphore / 线程池	消费	并发	单实例	无	保护 DB 连接池、下游连接数
拉取参数 + suspend()/resume()	消费	背压	单实例	无	下游过载时暂停拉取
Broker / 客户端兜底参数	Broker	线程池、队列	集群	无	兜底保护，不是业务限流

快速选型：

1. 单实例、只想简单限一下 → Guava RateLimiter（生产端）或 Semaphore（消费端）；
2. 多实例、要控制集群总 QPS → Redis + Lua 令牌桶（精度最好），或 Redisson（少写代码）；
3. 需要熔断降级、控制台、动态规则、集群流控 → Sentinel；
4. 下游是数据库这类“并发敏感”的资源 → 优先 Semaphore / 线程数，而不是 QPS；
5. 下游会临时故障 → 背压（暂停拉取）+ 熔断，别硬扛。

下面给出每种方案的落地代码。示例基于 rocketmq-spring-boot-starter 2.x（4.x remoting 客户端），5.x 的 gRPC 客户端思路一致。

依赖

xml
<dependencies>
    <!-- RocketMQ -->
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.3.1</version>
    </dependency>

    <!-- 单机限流 -->
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>33.2.1-jre</version>
    </dependency>

    <!-- 分布式限流 / 切面 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>

    <!-- 可选：Redisson -->
    <dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson-spring-boot-starter</artifactId>
        <version>3.35.0</version>
    </dependency>

    <!-- 可选：Sentinel -->
    <dependency>
        <groupId>com.alibaba.csp</groupId>
        <artifactId>sentinel-core</artifactId>
        <version>1.8.8</version>
    </dependency>
</dependencies>

版本仅为示例，按项目实际情况对齐即可。

---

三、具体实现

3.1 生产端单机限流：Guava RateLimiter

适合单实例部署，或者“每个实例限自己的速率”就够用的场景。

java
@Component
@RequiredArgsConstructor
public class OrderProducer {

    private final RocketMQTemplate rocketMQTemplate;

    // 单机 500 条/秒，所有发送线程共享同一个实例
    private final RateLimiter sendRateLimiter = RateLimiter.create(500.0);

    public void send(OrderMessage message) {
        // 最多等 50ms，等不到就直接失败，交给上游重试
        if (!sendRateLimiter.tryAcquire(1, 50, TimeUnit.MILLISECONDS)) {
            throw new SendRateLimitedException("发送速率超限，请稍后重试");
        }
        rocketMQTemplate.syncSend("order-topic", message, 3000);
    }
}

几个容易踩的点：

• 必须是单例。 每次 new RateLimiter 都是满桶，等于没限流；
• Guava 默认是“平滑突发”模型，最多会攒下约 1 秒的许可，也就是 500 QPS 的配置允许瞬间放出 500 条。如果需要严格匀速，用 Sentinel 的匀速排队（CONTROL_BEHAVIOR_RATE_LIMITER）或自己实现漏桶；
• tryAcquire 是线程安全的，可以放心被多个业务线程共用；
• 被拒绝后不要无界阻塞业务线程，落库补偿、延迟消息或者直接把可重试错误抛给上游，都比 while 重试安全。

如果追求吞吐而不是严格的发送速率，更好的方向是批量发送 + 异步发送，用信号量控制 in-flight 数量，而不是逐条 syncSend：

java
public void sendBatch(List<OrderMessage> messages) {
    // 一次性发一批，减少网络往返；批量大小受 max-message-size 限制
    rocketMQTemplate.syncSend("order-topic", messages, 5000);
}

3.2 生产端 / 消费端通用：Redis + Lua 令牌桶

多实例部署时，需要的是“所有实例加起来每秒 N 条”。这时把令牌桶放到 Redis 里，用 Lua 保证“读-算-写”原子执行。

Lua 脚本

lua
-- KEYS[1] 令牌桶 key
-- ARGV[1] qps（每秒补充的令牌数）
-- ARGV[2] capacity（桶容量，即允许的突发量）
-- ARGV[3] 本次请求的令牌数
-- ARGV[4] 额外 TTL 秒数

local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local permits = tonumber(ARGV[3])

-- 用 Redis 时间，避免多台应用服务器时钟不一致
-- 注意：redis.call('TIME') 需要 Redis 5+（effect replication）
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)

local data = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
local tokens = tonumber(data[1])
local ts = tonumber(data[2])

if tokens == nil then
    tokens = capacity
    ts = now
end

local delta = math.max(0, now - ts)
tokens = math.min(capacity, tokens + delta * rate / 1000)

local allowed = 0
if tokens >= permits then
    tokens = tokens - permits
    allowed = 1
end

redis.call('HMSET', KEYS[1], 'tokens', tokens, 'ts', now)

-- 低频场景下 key 过期会被重置成满桶，所以 TTL 至少给一个补满周期 + 60s
local ttl = math.floor(capacity / rate * 1000) + 60000
redis.call('PEXPIRE', KEYS[1], ttl)

return allowed

Java 封装

java
@Component
@RequiredArgsConstructor
public class RedisTokenBucketLimiter {

    private static final String LUA = """
            local rate = tonumber(ARGV[1])
            local capacity = tonumber(ARGV[2])
            local permits = tonumber(ARGV[3])
            local t = redis.call('TIME')
            local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
            local data = redis.call('HMGET', KEYS[1], 'tokens', 'ts')
            local tokens = tonumber(data[1])
            local ts = tonumber(data[2])
            if tokens == nil then
                tokens = capacity
                ts = now
            end
            local delta = math.max(0, now - ts)
            tokens = math.min(capacity, tokens + delta * rate / 1000)
            local allowed = 0
            if tokens >= permits then
                tokens = tokens - permits
                allowed = 1
            end
            redis.call('HMSET', KEYS[1], 'tokens', tokens, 'ts', now)
            local ttl = math.floor(capacity / rate * 1000) + 60000
            redis.call('PEXPIRE', KEYS[1], ttl)
            return allowed
            """;

    private static final RedisScript<Long> SCRIPT =
            new DefaultRedisScript<>(LUA, Long.class);

    private final StringRedisTemplate redisTemplate;

    public boolean tryAcquire(String key, double qps, long burst) {
        if (qps <= 0 || burst <= 0) {
            throw new IllegalArgumentException("qps/burst 必须大于 0");
        }
        Long result = redisTemplate.execute(
                SCRIPT,
                List.of("rate_limit:" + key),
                String.valueOf(qps),
                String.valueOf(burst),
                "1");
        return result != null && result == 1L;
    }
}

注解 + 切面

java
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimit {

    /** 限流 key，支持 SpEL，例如 "'send:order:' + #p0.userId" */
    String key();

    /** 每秒令牌数（集群总量） */
    double qps();

    /** 允许的突发量，一般设为 qps 的 1~2 倍 */
    long burst() default 1;

    /** Redis 不可用时是否放行；核心链路建议 false */
    boolean failOpen() default false;
}

java
@Slf4j
@Aspect
@Component
@RequiredArgsConstructor
public class RateLimitAspect {

    private final RedisTokenBucketLimiter limiter;

    private final ExpressionParser parser = new SpelExpressionParser();
    private final DefaultParameterNameDiscoverer nameDiscoverer =
            new DefaultParameterNameDiscoverer();

    @Around("@annotation(rateLimit)")
    public Object around(ProceedingJoinPoint pjp, RateLimit rateLimit) throws Throwable {
        String key = resolveKey(rateLimit.key(), pjp);

        boolean allowed;
        try {
            allowed = limiter.tryAcquire(key, rateLimit.qps(), rateLimit.burst());
        } catch (DataAccessException e) {
            if (!rateLimit.failOpen()) {
                throw new RateLimitedException(key, e);
            }
            // 放行前一定要打日志和指标，否则 Redis 故障期间限流形同虚设
            log.warn("限流依赖的 Redis 不可用，降级放行 key={}", key, e);
            allowed = true;
        }

        if (!allowed) {
            throw new RateLimitedException(key);
        }
        return pjp.proceed();
    }

    private String resolveKey(String expression, ProceedingJoinPoint pjp) {
        if (!expression.contains("#")) {
            return expression;
        }
        MethodSignature signature = (MethodSignature) pjp.getSignature();
        String[] names = nameDiscoverer.getParameterNames(signature.getMethod());
        Object[] args = pjp.getArgs();

        StandardEvaluationContext context = new StandardEvaluationContext();
        for (int i = 0; i < args.length; i++) {
            // 同时注册 #p0/#a0 和真实参数名，避免 -parameters 未开启时取不到名字
            context.setVariable("p" + i, args[i]);
            context.setVariable("a" + i, args[i]);
            if (names != null && i < names.length) {
                context.setVariable(names[i], args[i]);
            }
        }
        return parser.parseExpression(expression).getValue(context, String.class);
    }
}

java
public class RateLimitedException extends RuntimeException {

    public RateLimitedException(String key) {
        super("rate limited: " + key);
    }

    public RateLimitedException(String key, Throwable cause) {
        super("rate limited: " + key, cause);
    }
}

使用方式

生产端：

java
@RateLimit(key = "'send:order:' + #p0.userId", qps = 20, burst = 40)
public void send(OrderMessage message) {
    rocketMQTemplate.syncSend("order-topic", message);
}

消费端（限集群总消费速率）：

java
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order-consumer")
@Service
public class OrderConsumer implements RocketMQListener<OrderMessage> {

    @RateLimit(key = "'consume:order'", qps = 200, burst = 200)
    @Override
    public void onMessage(OrderMessage message) {
        handle(message);
    }
}

注意事项：

• 消费端被限流后抛出的异常会被 RocketMQ 当成消费失败，消息进入重试队列（带延迟等级）。这通常是你想要的，但要配合 maxReconsumeTimes 和死信队列一起考虑（见第四节）；
• key 的粒度决定限流效果：按 topic 限、按 tag 限、按租户/商家限，语义完全不同。按业务 key 限时注意热 key；
• 每条消息一次 Redis 往返，2000 QPS 以内一般没问题。更高吞吐可以“批量预取配额”（一次取 100 个令牌），代价是精度和公平性下降；
• Redis 不可用时的 failOpen 要显式选择：限流失效 还是 业务不可用，这是业务问题，不是技术问题。

3.3 消费端本地：信号量 + 令牌桶（阻塞式背压）

如果下游是 MySQL 这类对并发敏感的资源，限 QPS 不如限并发。Semaphore 比 consumeThreadMax 更细：你可以让 20 个消费线程里最多只有 8 个真正打到下游，其余线程自动阻塞，形成背压。

java
@Component
public class LocalConsumeThrottle {

    // 平滑限速：200 条/秒
    private final RateLimiter rateLimiter = RateLimiter.create(200.0);

    // 并发上限：同时最多 16 条在处理
    private final Semaphore concurrency = new Semaphore(16);

    public void acquire() {
        // 先占并发额度，再拿速率令牌；阻塞是刻意的，用来把压力顶回 Broker
        concurrency.acquireUninterruptibly();
        rateLimiter.acquire();
    }

    public void release() {
        concurrency.release();
    }
}

java
@Service
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = "order-topic",
        consumerGroup = "order-consumer",
        consumeThreadMax = 20)   // 略大于并发上限即可，避免大量线程空转
public class OrderConsumer implements RocketMQListener<OrderMessage> {

    private final LocalConsumeThrottle throttle;

    @Override
    public void onMessage(OrderMessage message) {
        throttle.acquire();
        try {
            handle(message);
        } finally {
            throttle.release();
        }
    }
}

为什么这样能形成背压：消费线程被阻塞 → 消息迟迟不 ACK → 客户端的未消费消息数上涨 → 达到 pullThresholdForQueue 后暂停拉取 → 消息留在 Broker。整个过程不产生重试，也不会丢掉消息，代价是端到端延迟上升。

如果你不能接受延迟上升，就不该用阻塞式限流，而应该扩容消费端和下游——这是容量问题，不是限流问题。

3.4 消费端集群总量：Redisson RRateLimiter

不想维护 Lua 脚本，可以直接用 Redisson：

java
@Configuration
public class RateLimiterConfig {

    @Bean
    public RRateLimiter orderConsumeRateLimiter(RedissonClient redissonClient) {
        RRateLimiter limiter = redissonClient.getRateLimiter("rate:consume:order");
        // OVERALL：所有实例共享 2000 条/秒；PER_CLIENT 则是每实例
        // 只在 key 不存在时生效，改配置需要 setRate()
        limiter.trySetRate(RateType.OVERALL, 2000, 1, RateIntervalUnit.SECONDS);
        return limiter;
    }
}

java
@Service
@RequiredArgsConstructor
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order-consumer")
public class OrderConsumer implements RocketMQListener<OrderMessage> {

    private final RRateLimiter orderConsumeRateLimiter;

    @Override
    public void onMessage(OrderMessage message) {
        // 最多等 200ms；不想占线程就直接 tryAcquire(1) 快速失败
        if (!orderConsumeRateLimiter.tryAcquire(1, 200, TimeUnit.MILLISECONDS)) {
            throw new RateLimitedException("consume:order");
        }
        handle(message);
    }
}

Redisson 的 RRateLimiter 本质是按 interval 重置配额的固定窗口实现，不是严格令牌桶：窗口边界上可能出现接近 2 倍的瞬时突发。要求严格平滑，还是用 3.2 的 Lua 令牌桶或 Sentinel 匀速排队。

3.5 消费端限流 + 熔断：Sentinel

需要“限流 + 熔断 + 控制台 + 动态规则”时，Sentinel 更省事。核心是把消费逻辑包成一个资源，流控规则按资源名配置。

java
@Configuration
public class SentinelConfig {

    @PostConstruct
    public void initFlowRules() {
        // QPS 限流：快速失败
        FlowRule qpsRule = new FlowRule();
        qpsRule.setResource("consume:order-topic:order-consumer");
        qpsRule.setGrade(RuleConstant.FLOW_GRADE_QPS);
        qpsRule.setCount(200);
        qpsRule.setControlBehavior(RuleConstant.CONTROL_BEHAVIOR_DEFAULT);

        // 并发限流：同时最多 16 条在处理，保护下游连接池
        FlowRule threadRule = new FlowRule();
        threadRule.setResource("consume:order-topic:order-consumer");
        threadRule.setGrade(RuleConstant.FLOW_GRADE_THREAD);
        threadRule.setCount(16);

        FlowRuleManager.loadRules(List.of(qpsRule, threadRule));
    }
}

java
@Service
@RocketMQMessageListener(topic = "order-topic", consumerGroup = "order-consumer")
public class OrderConsumer implements RocketMQListener<OrderMessage> {

    private static final String RESOURCE = "consume:order-topic:order-consumer";

    @Override
    public void onMessage(OrderMessage message) {
        Entry entry = null;
        try {
            entry = SphU.entry(RESOURCE);
            handle(message);
        } catch (BlockException e) {
            // 被限流：抛出去走 RocketMQ 重试（带延迟），或者记录后 return 丢弃
            throw new RateLimitedException(RESOURCE, e);
        } finally {
            if (entry != null) {
                entry.exit();
            }
        }
    }
}

补充说明：

• 需要“匀速排队”时，把流控行为改成 CONTROL_BEHAVIOR_RATE_LIMITER 并设置 maxQueueingTimeMs，请求会排队而不是立刻拒绝；
• 集群流控需要额外的 token server（sentinel-cluster-client-default + 独立部署），运维成本比 Redis 方案高；
• Sentinel 官方提供了 sentinel-apache-rocketmq 适配模块，通过 ConsumeMessageHook 自动埋点，资源名形如 topic:group。如果需要按业务维度（租户、消息类型）限流，手动 SphU.entry 更灵活；
• 规则可以接 Nacos 配置中心动态下发，避免改阈值就重启。

3.6 消费端背压：拉取参数 + suspend/resume

RocketMQ 4.x 的 PushConsumer 本质是“客户端拉取 + 本地投递线程池”。所以除了限流，还可以直接调拉取参数，让客户端少缓存一点：

java
@Slf4j
@Service
@RocketMQMessageListener(
        topic = "order-topic",
        consumerGroup = "order-consumer",
        consumeThreadMax = 20,
        maxReconsumeTimes = 8)
public class OrderConsumer implements RocketMQListener<OrderMessage>,
        RocketMQPushConsumerLifecycleListener {

    private volatile DefaultMQPushConsumer consumer;
    private volatile boolean paused = false;

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        // 每个队列最多缓存 200 条、20MiB，超过就暂停拉取
        consumer.setPullThresholdForQueue(200);
        consumer.setPullThresholdSizeForQueue(20);
        consumer.setPullBatchSize(32);
        consumer.setConsumeMessageBatchMaxSize(1);
        this.consumer = consumer;
    }

    // 下游过载时暂停消费，恢复后继续；需要 @EnableScheduling
    @Scheduled(fixedDelay = 1000)
    public void adjustByDownstreamHealth() {
        if (consumer == null) {
            return;
        }
        // 示例：DB 连接池使用率、下游 RT、错误率，任一超阈值即视为过载
        boolean overloaded = DownstreamHealth.isOverloaded();
        if (overloaded && !paused) {
            consumer.suspend();
            paused = true;
            log.warn("下游过载，暂停当前实例的消费");
        } else if (!overloaded && paused) {
            consumer.resume();
            paused = false;
            log.info("下游恢复，继续消费");
        }
    }

    @Override
    public void onMessage(OrderMessage message) {
        handle(message);
    }
}

几点必须知道：

• suspend() 只暂停当前实例，其他实例照常消费。要全局暂停，需要配置中心广播或者 Broker 侧操作；
• suspend() 不会中断已经取到本地的消息，正在处理的消息会正常完成；
• 暂停期间消息留在 Broker，消费延迟持续上涨，这是预期行为；恢复后要关注是否会把下游再次打满；
• pullThresholdForQueue 调得太小会降低吞吐，太大则失去背压意义，要结合单条消息大小和消费耗时压测。

3.7 兜底参数：Broker 与客户端

这些参数不是限流，但能在异常流量下保命。

生产端：

yaml
rocketmq:
  producer:
    group: order-producer
    send-message-timeout: 3000
    retry-times-when-send-failed: 2
    retry-times-when-send-async-failed: 2

Broker（broker.conf）：

properties
# 发送线程数与队列：队列满会触发快速失败，保护 Broker
sendMessageThreadPoolNums=16
sendThreadPoolQueueCapacity=10000
waitTimeMillsInSendQueue=200

消费端常用参数：

参数	含义	默认值（4.x）
consumeThreadMin / consumeThreadMax（客户端）	消费线程数	20 / 20
consumeThreadMax（@RocketMQMessageListener）	容器覆盖后的消费线程数	64
pullThresholdForQueue	单队列本地最多缓存多少条	1000
pullThresholdSizeForQueue	单队列本地最多缓存多少 MiB	100
consumeMessageBatchMaxSize	一次回调最多多少条	1
pullBatchSize	一次拉取最多多少条	32
consumeConcurrentlyMaxSpan	并发消费时队列中未消费消息的最大跨度	2000
maxReconsumeTimes	最大重试次数	-1（并发消费实际 16 次）

---

四、被限流之后怎么办

限流的代码只有几行，真正决定系统是否稳定的是“被拒绝之后做什么”。

场景	推荐动作	必须注意
生产端，非核心消息（日志、埋点）	直接丢弃 + 打点	丢弃必须可观测
生产端，核心消息	落本地库/Redis 队列，定时补偿；或发延迟消息	保证最终投递，消费端必须幂等
消费端，消息可丢	记录日志/指标后 return（ACK）	不要误吞不可丢的消息
消费端，消息不可丢	抛异常走重试；或手动转延迟 Topic	重试流量也占额度，避免重试风暴
下游故障	熔断 + 暂停消费	故障期间继续拉取只会放大损失

几个关键细节：

1. 重试是有代价的。 并发消费失败后进入 %RETRY%{group}，按延迟等级（1s、5s、10s、30s、1m……2h）重投。如果限流是持续性的，重试消息会不断堆积，最终进入 %DLQ%{group}。死信队列一定要有人处理。
2. 被限流的尝试也会占用消费线程。 如果每条消息都快速失败重投，消费端看起来“很忙”，但真正处理成功的吞吐并不高。这时应该先解决下游容量，而不是继续调小限流值。
3. 幂等是前提。 只要用了重试或补偿投递，就必须假设消息会重复。
4. 监控要闭环。 至少关注：消费延迟（Broker 堆积量）、消费 TPS、限流拒绝次数、重试次数、死信数量。限流阈值应该跟着这些指标调，而不是拍脑袋。
5. 不要在消费线程里 sleep。 用 RateLimiter、Semaphore 或背压参数，至少它们是可测量、可中断、有明确语义的。

---

五、常见坑与经验

1. 单机限流 ≠ 集群限流。 Guava RateLimiter 的 200 QPS 在 8 个 Pod 上是 1600 QPS。多实例要么用 Redis/Redisson/Sentinel 集群流控，要么在配置里按实例数折算——但扩缩容时一定会忘。
2. 限流器要单例。 每次请求 new RateLimiter 或 new Semaphore，等于没有限流。
3. 令牌桶有突发。 桶初始是满的，服务刚启动或 Redis key 刚创建时，第一批请求可能瞬间放行 burst 条。下游冷启动扛不住时，把 burst 设小，或者用匀速排队。
4. 顺序消息不要随便限流。 ConsumeMode.ORDERLY 的并发度等于队列数，引入异步线程池或随机拒绝会破坏顺序。顺序链路优先扩容队列和下游，而不是硬限流。
5. 阈值来自压测，不来自感觉。 先测出下游的安全 QPS/并发，再留 20%~30% 余量，最后才是配置限流值。
6. 限流阈值要能动态调整。 写死在注解里，每次都发版；接到 Nacos/Apollo 或 Sentinel 控制台，才能在大促时快速调。
7. 限流不解决堆积。 限流保护的是下游，堆积需要扩容消费者、增加队列数、提升下游处理能力，或者业务降级。
8. Redis 限流要考虑故障策略。 Redis 挂了是放行还是拒绝，要提前定好，并准备降级开关。
9. 消费端限流会增加延迟。 对延迟敏感的业务（支付回调、验证码），限流可能比扩容更伤用户体验。
10. 5.x 的变化。 gRPC 客户端的 SimpleConsumer 采用手动 receive + ack，拉取节奏完全由业务控制，天然适合做限流；PushConsumer 的限流思路和 4.x 一致。

---

六、总结

给 RocketMQ 做限流，本质上是在三个位置保护三类对象：

• 生产端保护 Broker 和发送方：Guava（单机）、Redis Lua / Redisson（集群）、Sentinel（流控 + 熔断）；
• 消费端保护下游：Semaphore/线程数控并发，令牌桶控速率，拉取参数和 suspend() 做背压；
• Broker 参数只做兜底，不能当业务限流用。

落地时的最小闭环：

1. 先定位瓶颈在下游的哪个资源（连接池、QPS、第三方配额）；
2. 选位置：生产端入口限流，还是消费端出口限流；
3. 选范围：单机用 Guava/Semaphore，集群总量用 Redis Lua 或 Sentinel 集群流控；
4. 先定义“被限流后怎么办”，再写限流代码；
5. 阈值来自压测，规则支持动态调整，指标接入监控；
6. 记住限流只是保护手段，堆积和容量问题最终还是要靠扩容与降级解决。

如果只能记一句话：限流限的是你保护的对象，不是消息本身；先想清楚拒绝之后消息去哪，再决定用什么限流器。