RocketMQ 的消息是推模式，还是拉模式？

#中间件#面试八股

RocketMQ 的消息是推模式，还是拉模式？

结论：RocketMQ 在协议层只有拉模式（Pull），所谓推模式（Push）是客户端对"拉 + 长轮询（Long Polling）"的封装。理解这一点，才能解释为什么推模式也会重复消费、为什么消费进度由客户端决定、以及为什么会出现空轮询。

一、长轮询把"拉"伪装成"推"

Consumer                          Broker
   │  1. PullRequest(offset=100)     │
   ├───────────────────────────────>│  队列里没有新消息
   │                                │  不立即返回，挂起请求(最长15s)
   │        ┌── 期间消息写入 ─────────┤
   │        │                       │  唤醒挂起的请求
   │  2. 返回消息(offset=100..110)   │
   │<───────────────────────────────┤
   │  3. 回调 MessageListener 处理   │
   ├─ 处理成功，更新本地消费位点 ──────┤
   │  4. 下次 PullRequest(offset=111)│
   └───────────────────────────────>│

如果挂起期间一直没有新消息，服务端在 brokerSuspendMaxTimeMillis（默认 15000 毫秒）后返回空结果，客户端立刻发起下一次拉取。这就是"推"的实时性来源：没有消息时连接不空转，有消息时立即醒。

二、推与拉的真实差别

对比项	推模式 PushConsumer	拉模式 PullConsumer
使用方式	注册监听器，框架回调	业务自己循环调用 `pull`
消费进度	框架自动上报（每 5 秒）	业务自己维护 offset
流控	框架内置队列级阈值	完全由业务控制
实时性	高（长轮询 + 批量）	取决于轮询间隔
适用场景	绝大多数在线业务	对账、批量回放、按需限速
重复消费	位点未提交即重投	位点由业务掌握，可控

所以"推模式"并没有改变谁主动：始终是消费者主动拉，Broker 只负责在无数据时把请求 hold 住。Kafka 的 poll 也是同一思路，只是没有服务端挂起，靠客户端参数控制。

三、推模式下的流控参数

参数	默认值	作用
`pullBatchSize`	32	一次网络拉取最多多少条
`consumeMessageBatchMaxSize`	1	一次回调最多交给业务多少条
`pullThresholdForQueue`	1000	单队列本地缓存消息条数上限，超过则暂停拉取
`pullThresholdSizeForQueue`	100 MB	单队列本地缓存体积上限
`consumeConcurrentlyMaxSpan`	2000	单队列位点最大跨度，防止处理太慢导致乱序扩散
`pullInterval`	0	两次拉取的最小间隔，用于主动限速

这些阈值是消费端自我保护：如果业务处理很慢，缓存的 1000 条消息会占满内存甚至 OOM，框架就靠"停拉"来反压。

四、两种客户端写法

// 推模式：注册监听器，框架负责拉取、位点、重试
DefaultMQPushConsumer push = new DefaultMQPushConsumer("pay_consumer");
push.setNamesrvAddr("10.0.0.11:9876");
push.subscribe("PAY_TOPIC", "SUCCESS || REFUND");
push.setPullBatchSize(64);
push.setConsumeMessageBatchMaxSize(32);
push.registerMessageListener((MessageListenerConcurrently) (msgs, ctx) -> {
    for (MessageExt m : msgs) {
        handle(m);                 // 业务必须幂等
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
push.start();

// 拉模式：业务自己定位 offset，适合对账与回放
DefaultMQPullConsumer pull = new DefaultMQPullConsumer("recon_consumer");
pull.setNamesrvAddr("10.0.0.11:9876");
pull.start();
Set<MessageQueue> mqs = pull.fetchSubscribeMessageQueues("PAY_TOPIC");
for (MessageQueue mq : mqs) {
    long offset = pull.fetchConsumeOffset(mq, false);
    while (offset < pull.maxOffset(mq)) {
        PullResult pr = pull.pull(mq, "*", offset, 32);
        for (MessageExt m : pr.getMsgFoundList()) {
            archive(m);            // 只入归档库，不推进线上位点
        }
        offset = pr.getNextBeginOffset();
    }
}

五、常见追问

推模式会丢消息吗？ 消息不会因为"推"而丢，但处理失败返回 RECONSUME_LATER 会重投，反过来会重复。丢消息的风险在"先返回成功再异步处理"这种写法上。
消费进度为什么由客户端上报？ 因为拉取与处理都在客户端，Broker 不知道业务是否处理成功；位点每 5 秒持久化一次，这也是重启后可能重复消费的原因。
拉模式能保证实时性吗？ 可以，但要自己实现长轮询或缩短间隔；间隔太短会放大 Broker 压力，间隔太长延迟就上去了。
广播模式也能用拉模式吗？ 能，但位点存在本地文件，多实例各自维护，重置与排障都比集群模式麻烦。

小结

RocketMQ 只有拉模式，推模式是长轮询封装，Broker 负责挂起请求而不是主动下发。
长轮询用"挂起 15 秒"换实时性，既避免空轮询又保证消息到达即唤醒。
推模式省心但把流控与幂等责任留给了业务；拉模式适合对账、回放等需要精确控制位点的场景。
无论推还是拉，消费位点都由客户端决定，重复消费是常态，幂等必须做。

笔记加载中…